on 0bd3d4e help sign in
1595 lines 50 kB history raw
1{-# OPTIONS_GHC -Wno-deprecations #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.IsStream.Transform
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
9-- Portability : GHC
11module Streamly.Internal.Data.Stream.IsStream.Transform {-# DEPRECATED "Please use \"Streamly.Data.Stream.*\" instead." #-}
12 (
13 -- * Piping
14 -- | Pass through a 'Pipe'.
15 transform
17 -- * Folding
18 , foldrS
19 , foldrSShared
20 , foldrT
22 -- * Mapping
23 -- | Stateless one-to-one maps.
24 , map
25 , sequence
26 , mapM
27 , smapM
29 -- * Mapping Side Effects (Observation)
30 -- | See also the intersperse*_ combinators.
31 , trace
32 , trace_
33 , tap
34 , tapOffsetEvery
35 , tapAsync
36 , tapAsyncK
37 , distributeAsync_
38 , pollCounts
40 -- * Scanning By 'Fold'
41 , scan
42 , scanMany
43 , postscan
45 -- * Scanning
46 -- | Left scans. Stateful, mostly one-to-one maps.
47 , scanl'
48 , scanlM'
49 , scanlMAfter'
50 , postscanlMAfter'
51 , postscanl'
52 , postscanlM'
53 , prescanl'
54 , prescanlM'
55 , scanl1'
56 , scanl1M'
58 -- XXX Once we have pipes the contravariant transformations can be
59 -- represented by attaching pipes before a transformation.
60 --
61 -- , lscanl'
62 -- , lscanlM'
63 -- , lscanl1'
64 -- , lscanl1M'
65 --
66 -- , lpostscanl'
67 -- , lpostscanlM'
68 -- , lprescanl'
69 -- , lprescanlM'
71 -- * Filtering
72 -- | Produce a subset of the stream using criteria based on the values of
73 -- the elements. We can use a concatMap and scan for filtering but these
74 -- combinators are more efficient and convenient.
76 , with
77 , deleteBy
78 , filter
79 , filterM
80 , uniq
81 , uniqBy
82 , nubBy
83 , prune
84 , repeated
86 -- * Trimming
87 -- | Produce a subset of the stream trimmed at ends.
89 , take
90 , takeLast
91 , takeLastInterval
92 , takeWhile
93 , takeWhileM
94 , takeWhileLast
95 , takeWhileAround
96 , drop
97 , dropLast
98 , dropLastInterval
99 , dropWhile
100 , dropWhileM
101 , dropWhileLast
102 , dropWhileAround
104 -- * Inserting Elements
105 -- | Produce a superset of the stream. This is the opposite of
106 -- filtering/sampling. We can always use concatMap and scan for inserting
107 -- but these combinators are more efficient and convenient.
109 -- Element agnostic (Opposite of sampling)
110 , intersperse
111 , intersperseM -- XXX naming
112 , intersperseMWith
114 , intersperseMSuffix
115 , intersperseMSuffixWith
116 , interjectSuffix
118 -- , interspersePrefix
119 -- , interspersePrefixBySpan
121 -- * Inserting Side Effects/Time
122 , intersperseM_ -- XXX naming
123 , delay
124 , intersperseMSuffix_
125 , delayPost
126 , intersperseMPrefix_
127 , delayPre
129 -- * Element Aware Insertion
130 -- | Opposite of filtering
131 , insertBy
132 -- , intersperseByBefore
133 -- , intersperseByAfter
135 -- * Reordering
136 , reverse
137 , reverse'
138 , reassembleBy
140 -- * Position Indexing
141 , indexed
142 , indexedR
144 -- * Time Indexing
145 , timestamped
146 , timestampWith
147 -- , timestampedR -- timer
148 , timeIndexed
149 , timeIndexWith
151 -- * Searching
152 , findIndices -- XXX indicesBy
153 , elemIndices -- XXX indicesOf
155 -- * Rolling map
156 -- | Map using the previous element.
157 , rollingMapM
158 , rollingMap
159 , rollingMap2
161 -- * Maybe Streams
162 -- Move these to Streamly.Data.Maybe.Stream?
163 , catMaybes -- XXX justs (like lefts/rights)
164 , mapMaybe
165 , mapMaybeM
167 -- * Either Streams
168 -- Move these to Streamly.Data.Either.Stream?
169 , lefts
170 , rights
171 , both
173 -- * Concurrent Evaluation
174 -- ** Concurrent Pipelines
175 -- | Run streaming stages concurrently.
177 , mkAsync
178 , mkParallel
179 , applyAsync
180 , (|$)
181 , (|&)
183 -- ** Concurrency Control
184 , maxThreads
186 -- ** Buffering and Sampling
187 -- | Evaluate strictly using a buffer of results. When the buffer becomes
188 -- full we can block, drop the new elements, drop the oldest element and
189 -- insert the new at the end or keep dropping elements uniformly to match
190 -- the rate of the consumer.
191 , maxBuffer
192 , sampleOld
193 , sampleNew
194 , sampleRate
196 -- ** Rate Limiting
197 -- | Evaluate the stream at uniform intervals to maintain a specified
198 -- evaluation rate.
199 , Rate (..)
200 , rate
201 , avgRate
202 , minRate
203 , maxRate
204 , constRate
206 -- * Diagnostics
207 , inspectMode
209 -- * Deprecated
210 , scanx
211 )
212where
214#include "inline.hs"
216import Control.Concurrent (threadDelay)
217import Control.Monad (void)
218import Control.Monad.IO.Class (MonadIO(..))
219import Control.Monad.Trans.Class (MonadTrans(..))
220import Data.Either (isLeft, isRight, fromLeft, fromRight)
221import Data.Kind (Type)
222import Data.Maybe (isJust, fromJust)
223import Streamly.Internal.Data.Fold.Type (Fold (..))
224import Streamly.Internal.Data.Pipe.Type (Pipe (..))
225import Streamly.Internal.Data.Stream.IsStream.Combinators
226 ( inspectMode, maxBuffer, maxThreads, rate, avgRate, minRate
227 , maxRate, constRate)
228import Streamly.Internal.Data.Stream.IsStream.Common
229 ( absTimesWith
230 , drop
231 , findIndices
232 , map
233 , postscanlM'
234 , relTimesWith
235 , reverse
236 , reverse'
237 , scanlMAfter'
238 , postscanlMAfter'
239 , smapM
240 , take
241 , takeWhile
242 , interjectSuffix
243 , intersperseM
244 , mkAsync
245 , mkParallel
246 , zipWith
247 )
248import Streamly.Internal.Control.Concurrent (MonadAsync)
249import Streamly.Internal.Data.Stream.IsStream.Type
250 (IsStream(..), fromStreamD, toStreamD, toConsK)
251import Streamly.Internal.Data.Stream.Serial (SerialT)
252import Streamly.Internal.Data.SVar (Rate(..))
253import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
255import qualified Streamly.Data.Fold as FL
256import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
257import qualified Streamly.Internal.Data.Stream.Parallel as Par
258import qualified Streamly.Internal.Data.Stream.Serial as Serial
259import qualified Streamly.Internal.Data.Stream.StreamD as D
260 (transform, foldrT, tap, tapOffsetEvery, mapM, scan
261 , scanMany, postscan, scanlx', scanlM', scanl', postscanl', prescanl'
262 , prescanlM', scanl1M', scanl1', filter, filterM, uniq, deleteBy, takeWhileM
263 , dropWhile, dropWhileM, insertBy, intersperse
264 , intersperseM_, intersperseMSuffix, intersperseMSuffix_
265 , intersperseMSuffixWith, indexed, indexedR, rollingMap, rollingMapM
266 , rollingMap2, mapMaybe, mapMaybeM)
267import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
268 (foldrS, foldrSShared, mapMWith)
269import qualified Prelude
271import Prelude hiding
272 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
273 , reverse, foldr1 , scanl, scanl1, zipWith)
275--
276-- $setup
277-- >>> :m
278-- >>> :set -fno-warn-deprecations
279-- >>> import Control.Concurrent (threadDelay)
280-- >>> import Data.Function ((&))
281-- >>> import Streamly.Prelude ((|$))
282-- >>> import Prelude hiding ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
283-- >>> import qualified Streamly.Prelude as Stream
284-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
285-- >>> import qualified Streamly.Data.Fold as Fold
286-- >>> import qualified Streamly.Internal.Data.Fold as Fold
287-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
288--
289-- >>> hSetBuffering stdout LineBuffering
291------------------------------------------------------------------------------
292-- Piping
293------------------------------------------------------------------------------
295-- | Use a 'Pipe' to transform a stream.
296--
297-- /Pre-release/
298--
299{-# INLINE transform #-}
300transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
301transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
303------------------------------------------------------------------------------
304-- Transformation Folds
305------------------------------------------------------------------------------
307-- | Right fold to a streaming monad.
308--
309-- > foldrS Stream.cons Stream.nil === id
310--
311-- 'foldrS' can be used to perform stateless stream to stream transformations
312-- like map and filter in general. It can be coupled with a scan to perform
313-- stateful transformations. However, note that the custom map and filter
314-- routines can be much more efficient than this due to better stream fusion.
315--
316-- >>> Stream.toList $ Stream.foldrS Stream.cons Stream.nil $ Stream.fromList [1..5]
317-- [1,2,3,4,5]
318--
319-- Find if any element in the stream is 'True':
320--
321-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then (Stream.fromPure True) else xs) (Stream.fromPure False) $ (Stream.fromList (2:4:5:undefined) :: Stream.SerialT IO Int)
322-- [True]
323--
324-- Map (+2) on odd elements and filter out the even elements:
325--
326-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then (x + 2) `Stream.cons` xs else xs) Stream.nil $ (Stream.fromList [1..5] :: Stream.SerialT IO Int)
327-- [3,5,7]
328--
329-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
330-- is much more efficient:
331--
332-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
333--
334-- /Pre-release/
335{-# INLINE foldrS #-}
336foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
337foldrS f z xs =
338 fromStream
339 $ K.foldrS
340 (\y ys -> toStream $ f y (fromStream ys))
341 (toStream z)
342 (toStream xs)
344{-# INLINE foldrSShared #-}
345foldrSShared :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
346foldrSShared f z xs =
347 fromStream
348 $ K.foldrSShared
349 (\y ys -> toStream $ f y (fromStream ys))
350 (toStream z)
351 (toStream xs)
353-- | Right fold to a transformer monad. This is the most general right fold
354-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
355-- implementation can be more efficient:
356--
357-- > foldrS = foldrT
358-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
359--
360-- 'foldrT' can be used to translate streamly streams to other transformer
361-- monads e.g. to a different streaming type.
362--
363-- /Pre-release/
364{-# INLINE foldrT #-}
365foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
366 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
367foldrT f z s = D.foldrT f z (toStreamD s)
369------------------------------------------------------------------------------
370-- Transformation by Mapping
371------------------------------------------------------------------------------
373-- |
374-- @
375-- mapM f = sequence . map f
376-- @
377--
378-- Apply a monadic function to each element of the stream and replace it with
379-- the output of the resulting action.
380--
381-- @
382-- >>> drain $ Stream.mapM putStr $ Stream.fromList ["a", "b", "c"]
383-- abc
384--
385-- >>> :{
386-- drain $ Stream.replicateM 10 (return 1)
387-- & (fromSerial . Stream.mapM (\x -> threadDelay 1000000 >> print x))
388-- :}
389-- 1
390-- ...
391-- 1
392--
393-- > drain $ Stream.replicateM 10 (return 1)
394-- & (fromAsync . Stream.mapM (\x -> threadDelay 1000000 >> print x))
395-- @
396--
397-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
398--
399-- @since 0.1.0
400{-# INLINE_EARLY mapM #-}
401mapM :: forall t m a b. (IsStream t, MonadAsync m) =>
402 (a -> m b) -> t m a -> t m b
403mapM f = fromStream . K.mapMWith (toConsK (consM @t)) f . toStream
405{-# RULES "mapM serial" mapM = mapMSerial #-}
406{-# INLINE mapMSerial #-}
407mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
408mapMSerial = Serial.mapM
410-- |
411-- @
412-- sequence = mapM id
413-- @
414--
415-- Replace the elements of a stream of monadic actions with the outputs of
416-- those actions.
417--
418-- @
419-- >>> drain $ Stream.sequence $ Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
420-- abc
421--
422-- >>> :{
423-- drain $ Stream.replicateM 3 (return $ threadDelay 1000000 >> print 1)
424-- & (fromSerial . Stream.sequence)
425-- :}
426-- 1
427-- 1
428-- 1
429--
430-- >>> :{
431-- drain $ Stream.replicateM 3 (return $ threadDelay 1000000 >> print 1)
432-- & (fromAsync . Stream.sequence)
433-- :}
434-- 1
435-- 1
436-- 1
437--
438-- @
439--
440-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
441--
442-- @since 0.1.0
443{-# INLINE sequence #-}
444sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
445sequence = mapM id
447------------------------------------------------------------------------------
448-- Mapping side effects
449------------------------------------------------------------------------------
451-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
452-- add a tap to log the contents flowing through the stream. The fold is used
453-- only for effects, its result is discarded.
454--
455-- @
456-- Fold m a b
457-- |
458-- -----stream m a ---------------stream m a-----
459--
460-- @
461--
462-- >>> Stream.drain $ Stream.tap (Fold.drainBy print) (Stream.enumerateFromTo 1 2)
463-- 1
464-- 2
465--
466-- Compare with 'trace'.
467--
468-- @since 0.7.0
469{-# INLINE tap #-}
470tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
471tap f xs = fromStreamD $ D.tap f (toStreamD xs)
473-- XXX Remove this. It can be expressed in terms of Fold.sampleFromThen.
474--
475-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
476-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
477-- means start at the first element in the stream. If the offset is outside
478-- this range then @offset `mod` n@ is used as offset.
479--
480-- >>> Stream.drain $ Stream.tapOffsetEvery 0 2 (Fold.rmapM print Fold.toList) $ Stream.enumerateFromTo 0 10
481-- [0,2,4,6,8,10]
483--
484-- /Pre-release/
485--
486{-# INLINE tapOffsetEvery #-}
487tapOffsetEvery :: (IsStream t, Monad m)
488 => Int -> Int -> FL.Fold m a b -> t m a -> t m a
489tapOffsetEvery offset n f xs =
490 fromStreamD $ D.tapOffsetEvery offset n f (toStreamD xs)
492-- | Redirect a copy of the stream to a supplied fold and run it concurrently
493-- in an independent thread. The fold may buffer some elements. The buffer size
494-- is determined by the prevailing 'maxBuffer' setting.
495--
496-- @
497-- Stream m a -> m b
498-- |
499-- -----stream m a ---------------stream m a-----
500--
501-- @
502--
503-- @
504-- >>> Stream.drain $ Stream.tapAsync (Fold.drainBy print) (Stream.enumerateFromTo 1 2)
505-- 1
506-- 2
507--
508-- @
509--
510-- Exceptions from the concurrently running fold are propagated to the current
511-- computation. Note that, because of buffering in the fold, exceptions may be
512-- delayed and may not correspond to the current element being processed in the
513-- parent stream, but we guarantee that before the parent stream stops the tap
514-- finishes and all exceptions from it are drained.
515--
516-- >>> tapAsync f = Stream.tapAsyncK (Stream.fold f . Stream.adapt)
517--
518-- Compare with 'tap'.
519--
520-- /Pre-release/
521{-# INLINE tapAsync #-}
522tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
523tapAsync f xs = fromStreamD $ Par.tapAsyncF f (toStreamD xs)
525-- XXX We should keep only tapAsyncK as tapAsync can be implemented in terms
526-- of tapAsyncK and it has better performance. Rename this to tapAsync.
528-- | Like 'tapAsyncF' but uses a stream fold function instead of a 'Fold' type.
529--
530-- /Pre-release/
531{-# INLINE tapAsyncK #-}
532tapAsyncK :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
533tapAsyncK f m = fromStream $ Par.tapAsyncK (f . fromStream) (toStream m)
535-- XXX rename this to tapManyAsync_?
536--
537-- | Concurrently distribute a stream to a collection of fold functions,
538-- discarding the outputs of the folds.
539--
540-- @
541-- > Stream.drain $ Stream.distributeAsync_ [Stream.mapM_ print, Stream.mapM_ print] (Stream.enumerateFromTo 1 2)
542-- 1
543-- 2
544-- 1
545-- 2
546--
547-- @
548--
549-- @
550-- distributeAsync_ = flip (foldr tapAsync)
551-- @
552--
553-- /Pre-release/
554--
555{-# INLINE distributeAsync_ #-}
556distributeAsync_ :: (Foldable f, IsStream t, MonadAsync m)
557 => f (t m a -> m b) -> t m a -> t m a
558distributeAsync_ = flip (Prelude.foldr tapAsyncK)
560-- | @pollCounts predicate transform fold stream@ counts those elements in the
561-- stream that pass the @predicate@. The resulting count stream is sent to
562-- another thread which transforms it using @transform@ and then folds it using
563-- @fold@. The thread is automatically cleaned up if the stream stops or
564-- aborts due to exception.
565--
566-- For example, to print the count of elements processed every second:
567--
568-- @
569-- > Stream.drain $ Stream.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (FLold.drainBy print)
570-- $ Stream.enumerateFrom 0
571-- @
572--
573-- Note: This may not work correctly on 32-bit machines.
574--
575-- /Pre-release/
576--
577{-# INLINE pollCounts #-}
578pollCounts ::
579 (IsStream t, MonadAsync m)
580 => (a -> Bool)
581 -> (t m Int -> m b)
582 -> t m a
583 -> t m a
584pollCounts predicate f xs =
585 fromStreamD
586 $ Concur.tapCountD predicate (f . fromStreamD)
587 $ toStreamD xs
589-- | Apply a monadic function to each element flowing through the stream and
590-- discard the results.
591--
592-- >>> Stream.drain $ Stream.trace print (Stream.enumerateFromTo 1 2)
593-- 1
594-- 2
595--
596-- Compare with 'tap'.
597--
598-- @since 0.7.0
599{-# INLINE trace #-}
600trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
601trace f = mapM (\x -> void (f x) >> return x)
603-- | Perform a side effect before yielding each element of the stream and
604-- discard the results.
605--
606-- @
607-- >>> Stream.drain $ Stream.trace_ (print "got here") (Stream.enumerateFromTo 1 2)
608-- "got here"
609-- "got here"
610--
611-- @
612--
613-- Same as 'intersperseMPrefix_' but always serial.
614--
615-- See also: 'trace'
616--
617-- /Pre-release/
618{-# INLINE trace_ #-}
619trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
620trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
622------------------------------------------------------------------------------
623-- Scanning with a Fold
624------------------------------------------------------------------------------
626-- | Scan a stream using the given monadic fold.
627--
628-- >>> Stream.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum (Stream.fromList [1..10])
629-- [0,1,3,6]
630--
631-- @since 0.7.0
632{-# INLINE scan #-}
633scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
634scan fld m = fromStreamD $ D.scan fld $ toStreamD m
636-- | Like 'scan' but restarts scanning afresh when the scanning fold
637-- terminates.
638--
639-- /Pre-release/
640{-# INLINE scanMany #-}
641scanMany :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
642scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
644-- | Postscan a stream using the given monadic fold.
645--
646-- The following example extracts the input stream up to a point where the
647-- running average of elements is no more than 10:
648--
649-- >>> import Data.Maybe (fromJust)
650-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
651-- >>> :{
652-- Stream.toList
653-- $ Stream.map (fromJust . fst)
654-- $ Stream.takeWhile (\(_,x) -> x <= 10)
655-- $ Stream.postscan (Fold.tee Fold.last avg) (Stream.enumerateFromTo 1.0 100.0)
656-- :}
657-- [1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]
658--
659-- @since 0.7.0
660{-# INLINE postscan #-}
661postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
662postscan fld = fromStreamD . D.postscan fld . toStreamD
664------------------------------------------------------------------------------
665-- Scanning - Transformation by Folding
666------------------------------------------------------------------------------
668-- XXX It may be useful to have a version of scan where we can keep the
669-- accumulator independent of the value emitted. So that we do not necessarily
670-- have to keep a value in the accumulator which we are not using. We can pass
671-- an extraction function that will take the accumulator and the current value
672-- of the element and emit the next value in the stream. That will also make it
673-- possible to modify the accumulator after using it. In fact, the step function
674-- can return new accumulator and the value to be emitted. The signature would
675-- be more like mapAccumL. Or we can change the signature of scanx to
676-- accommodate this.
677--
678-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
679-- user supplied extraction function (the third argument) at each step. This is
680-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
681-- extraction.
682--
683-- /Since 0.2.0/
684--
685-- /Since: 0.7.0 (Monad m constraint)/
686{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
687{-# INLINE scanx #-}
688scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
689scanx step begin done = fromStreamD . D.scanlx' step begin done . toStreamD
691-- XXX this needs to be concurrent
692-- XXX because of the use of D.cons for appending, scanlM' has quadratic
693-- complexity when iterated over a stream. We should use StreamK style scanlM'
694-- for linear performance on iteration.
695--
696-- | Like 'scanl'' but with a monadic step function and a monadic seed.
697--
698-- /Since: 0.4.0/
699--
700-- /Since: 0.8.0 (signature change)/
701{-# INLINE scanlM' #-}
702scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
703scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
705-- XXX because of the use of D.cons for appending, scanl' has quadratic
706-- complexity when iterated over a stream. We should use StreamK style scanlM'
707-- for linear performance on iteration.
709-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
710-- however it adds an extra element.
711--
712-- @
713-- >>> Stream.toList $ Stream.scanl' (+) 0 $ fromList [1,2,3,4]
714-- [0,1,3,6,10]
715--
716-- @
717--
718-- @
719-- >>> Stream.toList $ Stream.scanl' (flip (:)) [] $ Stream.fromList [1,2,3,4]
720-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
721--
722-- @
723--
724-- The output of 'scanl'' is the initial value of the accumulator followed by
725-- all the intermediate steps and the final result of 'foldl''.
726--
727-- By streaming the accumulated state after each fold step, we can share the
728-- state across multiple stages of stream composition. Each stage can modify or
729-- extend the state, do some processing with it and emit it for the next stage,
730-- thus modularizing the stream processing. This can be useful in
731-- stateful or event-driven programming.
732--
733-- Consider the following monolithic example, computing the sum and the product
734-- of the elements in a stream in one go using a @foldl'@:
735--
736-- @
737-- >>> Stream.foldl' (\(s, p) x -> (s + x, p * x)) (0,1) $ Stream.fromList [1,2,3,4]
738-- (10,24)
739--
740-- @
741--
742-- Using @scanl'@ we can make it modular by computing the sum in the first
743-- stage and passing it down to the next stage for computing the product:
744--
745-- @
746-- >>> :{
747-- Stream.foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
748-- $ Stream.scanl' (\(s, _) x -> (s + x, x)) (0,1)
749-- $ Stream.fromList [1,2,3,4]
750-- :}
751-- (10,24)
752--
753-- @
754--
755-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
756-- lazy expressions inside the accumulator, it is recommended that a strict
757-- data structure is used for accumulator.
758--
759-- >>> scanl' step z = scan (Fold.foldl' step z)
760-- >>> scanl' f z xs = scanlM' (\a b -> return (f a b)) (return z) xs
761-- >>> scanl' f z xs = z `Stream.cons` postscanl' f z xs
762--
763-- See also: 'usingStateT'
764--
765-- @since 0.2.0
766{-# INLINE scanl' #-}
767scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
768scanl' step z m = fromStreamD $ D.scanl' step z $ toStreamD m
770-- | Like 'scanl'' but does not stream the initial value of the accumulator.
771--
772-- >>> postscanl' step z = postscan (Fold.foldl' step z)
773-- >>> postscanl' f z = postscanlM' (\a b -> return (f a b)) (return z)
774-- >>> postscanl' f z xs = Stream.drop 1 $ Stream.scanl' f z xs
775--
776-- @since 0.7.0
777{-# INLINE postscanl' #-}
778postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
779postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
781-- XXX prescanl does not sound very useful, enable only if there is a
782-- compelling use case.
783--
784-- | Like scanl' but does not stream the final value of the accumulator.
785--
786-- /Pre-release/
787{-# INLINE prescanl' #-}
788prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
789prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
791-- XXX this needs to be concurrent
792-- | Like prescanl' but with a monadic step function and a monadic seed.
793--
794-- /Pre-release/
795{-# INLINE prescanlM' #-}
796prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
797prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
799-- XXX this needs to be concurrent
800-- | Like 'scanl1'' but with a monadic step function.
801--
802-- @since 0.6.0
803{-# INLINE scanl1M' #-}
804scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
805scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
807-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
808-- is used as the initial value of the accumulator. Does nothing if the stream
809-- is empty.
810--
811-- @
812-- >>> Stream.toList $ Stream.scanl1' (+) $ fromList [1,2,3,4]
813-- [1,3,6,10]
814--
815-- @
816--
817-- @since 0.6.0
818{-# INLINE scanl1' #-}
819scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
820scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
822------------------------------------------------------------------------------
823-- Filtering
824------------------------------------------------------------------------------
826-- | Modify a @t m a -> t m a@ stream transformation that accepts a predicate
827-- @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a transformation @t m
828-- a -> t m (s, a)@. Convenient to filter with index or time.
829--
830-- @
831-- filterWithIndex = with indexed filter
832-- filterWithAbsTime = with timestamped filter
833-- filterWithRelTime = with timeIndexed filter
834-- @
835--
836-- /Pre-release/
837{-# INLINE with #-}
838with :: forall (t :: (Type -> Type) -> Type -> Type) m a b s. Functor (t m) =>
839 (t m a -> t m (s, a))
840 -> (((s, a) -> b) -> t m (s, a) -> t m (s, a))
841 -> (((s, a) -> b) -> t m a -> t m a)
842with f comb g = fmap snd . comb g . f
844-- | Include only those elements that pass a predicate.
845--
846-- @since 0.1.0
847{-# INLINE filter #-}
848filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
849filter p m = fromStreamD $ D.filter p $ toStreamD m
851-- | Same as 'filter' but with a monadic predicate.
852--
853-- @since 0.4.0
854{-# INLINE filterM #-}
855filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
856filterM p m = fromStreamD $ D.filterM p $ toStreamD m
858-- | Drop repeated elements that are adjacent to each other using the supplied
859-- comparison function.
860--
861-- @uniq = uniqBy (==)
862--
863-- To strip duplicate path separators:
864--
865-- @
866-- f x y = x == '/' && x == y
867-- Stream.toList $ Stream.uniqBy f $ Stream.fromList "//a//b"
868-- "/a/b"
869-- @
870--
871-- Space: @O(1)@
872--
873-- See also: 'nubBy'.
874--
875-- /Pre-release/
876--
877{-# INLINE uniqBy #-}
878uniqBy :: (IsStream t, Monad m, Functor (t m)) =>
879 (a -> a -> Bool) -> t m a -> t m a
880uniqBy eq = catMaybes . rollingMap f
882 where
884 f pre curr =
885 case pre of
886 Nothing -> Just curr
887 Just x -> if x `eq` curr then Nothing else Just curr
889-- | Drop repeated elements that are adjacent to each other.
890--
891-- @since 0.6.0
892{-# INLINE uniq #-}
893uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
894uniq = fromStreamD . D.uniq . toStreamD
896-- | Strip all leading and trailing occurrences of an element passing a
897-- predicate and make all other consecutive occurrences uniq.
898--
899-- @
900-- prune p = dropWhileAround p $ uniqBy (x y -> p x && p y)
901-- @
902--
903-- @
904-- > Stream.prune isSpace (Stream.fromList " hello world! ")
905-- "hello world!"
906--
907-- @
908--
909-- Space: @O(1)@
910--
911-- /Unimplemented/
912{-# INLINE prune #-}
913prune ::
914 -- (IsStream t, Monad m, Eq a) =>
915 (a -> Bool) -> t m a -> t m a
916prune = error "Not implemented yet!"
918-- Possible implementation:
919-- @repeated =
920-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
921--
922-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
923-- single element.
924--
925-- | Emit only repeated elements, once.
926--
927-- /Unimplemented/
928repeated :: -- (IsStream t, Monad m, Eq a) =>
929 t m a -> t m a
930repeated = undefined
932-- We can have more efficient implementations for nubOrd and nubInt by using
933-- Set and IntSet to find/remove duplication. For Hashable we can use a
934-- hashmap. Use rewrite rules to specialize to more efficient impls.
935--
936-- | Drop repeated elements anywhere in the stream.
937--
938-- /Caution: not scalable for infinite streams/
939--
940-- /See also: nubWindowBy/
941--
942-- /Unimplemented/
943--
944{-# INLINE nubBy #-}
945nubBy :: -- (IsStream t, Monad m) =>
946 (a -> a -> Bool) -> t m a -> t m a
947nubBy = undefined -- fromStreamD . D.nubBy . toStreamD
949-- | Deletes the first occurrence of the element in the stream that satisfies
950-- the given equality predicate.
951--
952-- @
953-- >>> Stream.toList $ Stream.deleteBy (==) 3 $ Stream.fromList [1,3,3,5]
954-- [1,3,5]
955--
956-- @
957--
958-- @since 0.6.0
959{-# INLINE deleteBy #-}
960deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
961deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
963------------------------------------------------------------------------------
964-- Lossy Buffering
965------------------------------------------------------------------------------
967-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we
968-- may want to have the evaluation rate independent of the sampling rate. To
969-- support that we can decouple evaluation and sampling in independent stages.
970-- The sampling stage would strictly evaluate and sample, the evaluation stage
971-- would control the evaluation rate.
973-- | Evaluate the input stream continuously and keep only the oldest @n@
974-- elements in the buffer, discard the new ones when the buffer is full. When
975-- the output stream is evaluated it consumes the values from the buffer in a
976-- FIFO manner.
977--
978-- /Unimplemented/
979--
980{-# INLINE sampleOld #-}
981sampleOld :: -- (IsStream t, Monad m) =>
982 Int -> t m a -> t m a
983sampleOld = undefined
985-- | Evaluate the input stream continuously and keep only the latest @n@
986-- elements in a ring buffer, keep discarding the older ones to make space for
987-- the new ones. When the output stream is evaluated it consumes the values
988-- from the buffer in a FIFO manner.
989--
990-- /Unimplemented/
991--
992{-# INLINE sampleNew #-}
993sampleNew :: -- (IsStream t, Monad m) =>
994 Int -> t m a -> t m a
995sampleNew = undefined
997-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
998-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
999-- consumer pattern.
1001-- /Unimplemented/
1003{-# INLINE sampleRate #-}
1004sampleRate :: -- (IsStream t, Monad m) =>
1005 Double -> t m a -> t m a
1006sampleRate = undefined
1008------------------------------------------------------------------------------
1009-- Trimming
1010------------------------------------------------------------------------------
1012-- | Same as 'takeWhile' but with a monadic predicate.
1014-- @since 0.4.0
1015{-# INLINE takeWhileM #-}
1016takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1017takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
1019-- See the lastN fold for impl hints. Use a Data.Array based ring buffer.
1021-- takeLast n = Stream.concatM . fmap Array.toStream . fold Fold.lastN
1023-- | Take @n@ elements at the end of the stream.
1025-- O(n) space, where n is the number elements taken.
1027-- /Unimplemented/
1028{-# INLINE takeLast #-}
1029takeLast :: -- (IsStream t, Monad m) =>
1030 Int -> t m a -> t m a
1031takeLast = undefined -- fromStreamD $ D.takeLast n $ toStreamD m
1033-- | Take time interval @i@ seconds at the end of the stream.
1035-- O(n) space, where n is the number elements taken.
1037-- /Unimplemented/
1038{-# INLINE takeLastInterval #-}
1039takeLastInterval :: -- (IsStream t, Monad m) =>
1040 Double -> t m a -> t m a
1041takeLastInterval = undefined -- fromStreamD $ D.takeLast n $ toStreamD m
1043-- | Take all consecutive elements at the end of the stream for which the
1044-- predicate is true.
1046-- O(n) space, where n is the number elements taken.
1048-- /Unimplemented/
1049{-# INLINE takeWhileLast #-}
1050takeWhileLast :: -- (IsStream t, Monad m) =>
1051 (a -> Bool) -> t m a -> t m a
1052takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m
1054-- | Like 'takeWhile' and 'takeWhileLast' combined.
1056-- O(n) space, where n is the number elements taken from the end.
1058-- /Unimplemented/
1059{-# INLINE takeWhileAround #-}
1060takeWhileAround :: -- (IsStream t, Monad m) =>
1061 (a -> Bool) -> t m a -> t m a
1062takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
1064-- | Drop elements in the stream as long as the predicate succeeds and then
1065-- take the rest of the stream.
1067-- @since 0.1.0
1068{-# INLINE dropWhile #-}
1069dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1070dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
1072-- | Same as 'dropWhile' but with a monadic predicate.
1074-- @since 0.4.0
1075{-# INLINE dropWhileM #-}
1076dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1077dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
1079-- | Drop @n@ elements at the end of the stream.
1081-- O(n) space, where n is the number elements dropped.
1083-- /Unimplemented/
1084{-# INLINE dropLast #-}
1085dropLast :: -- (IsStream t, Monad m) =>
1086 Int -> t m a -> t m a
1087dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m
1089-- | Drop time interval @i@ seconds at the end of the stream.
1091-- O(n) space, where n is the number elements dropped.
1093-- /Unimplemented/
1094{-# INLINE dropLastInterval #-}
1095dropLastInterval :: -- (IsStream t, Monad m) =>
1096 Int -> t m a -> t m a
1097dropLastInterval = undefined
1099-- | Drop all consecutive elements at the end of the stream for which the
1100-- predicate is true.
1102-- O(n) space, where n is the number elements dropped.
1104-- /Unimplemented/
1105{-# INLINE dropWhileLast #-}
1106dropWhileLast :: -- (IsStream t, Monad m) =>
1107 (a -> Bool) -> t m a -> t m a
1108dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m
1110-- | Like 'dropWhile' and 'dropWhileLast' combined.
1112-- O(n) space, where n is the number elements dropped from the end.
1114-- /Unimplemented/
1115{-# INLINE dropWhileAround #-}
1116dropWhileAround :: -- (IsStream t, Monad m) =>
1117 (a -> Bool) -> t m a -> t m a
1118dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
1120------------------------------------------------------------------------------
1121-- Inserting Elements
1122------------------------------------------------------------------------------
1124-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
1125-- @stream@ that is less than @elem@ when compared using @cmp@.
1127-- @
1128-- insertBy cmp x = 'mergeBy' cmp ('fromPure' x)
1129-- @
1131-- @
1132-- >>> Stream.toList $ Stream.insertBy compare 2 $ Stream.fromList [1,3,5]
1133-- [1,2,3,5]
1135-- @
1137-- @since 0.6.0
1138{-# INLINE insertBy #-}
1139insertBy ::
1140 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
1141insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
1143-- | Insert a pure value between successive elements of a stream.
1145-- >>> Stream.toList $ Stream.intersperse ',' $ Stream.fromList "hello"
1146-- "h,e,l,l,o"
1148-- @since 0.7.0
1149{-# INLINE intersperse #-}
1150intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
1151intersperse a = fromStreamD . D.intersperse a . toStreamD
1153-- | Insert a side effect before consuming an element of a stream except the
1154-- first one.
1156-- >>> Stream.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') $ Stream.fromList "hello"
1157-- h.e.l.l.o
1159-- /Pre-release/
1160{-# INLINE intersperseM_ #-}
1161intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1162intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
1164-- | Intersperse a monadic action into the input stream after every @n@
1165-- elements.
1167-- @
1168-- > Stream.toList $ Stream.intersperseMWith 2 (return ',') $ Stream.fromList "hello"
1169-- "he,ll,o"
1171-- @
1173-- /Unimplemented/
1174{-# INLINE intersperseMWith #-}
1175intersperseMWith :: -- IsStream t =>
1176 Int -> m a -> t m a -> t m a
1177intersperseMWith _n _f _xs = undefined
1179-- | Insert an effect and its output after consuming an element of a stream.
1181-- >>> Stream.toList $ Stream.trace putChar $ intersperseMSuffix (putChar '.' >> return ',') $ Stream.fromList "hello"
1182-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
1184-- /Pre-release/
1185{-# INLINE intersperseMSuffix #-}
1186intersperseMSuffix :: (IsStream t, Monad m) => m a -> t m a -> t m a
1187intersperseMSuffix m = fromStreamD . D.intersperseMSuffix m . toStreamD
1189-- | Insert a side effect after consuming an element of a stream.
1191-- @
1192-- >>> Stream.mapM_ putChar $ Stream.intersperseMSuffix_ (threadDelay 1000000) $ Stream.fromList "hello"
1193-- hello
1195-- @
1197-- /Pre-release/
1199{-# INLINE intersperseMSuffix_ #-}
1200intersperseMSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1201intersperseMSuffix_ m = fromStreamD . D.intersperseMSuffix_ m . toStreamD
1203-- XXX Use an offset argument, like tapOffsetEvery
1205-- | Like 'intersperseMSuffix' but intersperses an effectful action into the
1206-- input stream after every @n@ elements and after the last element.
1208-- >>> Stream.toList $ Stream.intersperseMSuffixWith 2 (return ',') $ Stream.fromList "hello"
1209-- "he,ll,o,"
1211-- /Pre-release/
1213{-# INLINE intersperseMSuffixWith #-}
1214intersperseMSuffixWith :: (IsStream t, Monad m)
1215 => Int -> m a -> t m a -> t m a
1216intersperseMSuffixWith n eff =
1217 fromStreamD . D.intersperseMSuffixWith n eff . toStreamD
1219-- | Insert a side effect before consuming an element of a stream.
1221-- >>> Stream.toList $ Stream.trace putChar $ Stream.intersperseMPrefix_ (putChar '.' >> return ',') $ Stream.fromList "hello"
1222-- .h.e.l.l.o"hello"
1224-- Same as 'trace_' but may be concurrent.
1226-- /Concurrent/
1228-- /Pre-release/
1230{-# INLINE intersperseMPrefix_ #-}
1231intersperseMPrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
1232intersperseMPrefix_ m = mapM (\x -> void m >> return x)
1234------------------------------------------------------------------------------
1235-- Inserting Time
1236------------------------------------------------------------------------------
1238-- Note: delay must be serial.
1240-- | Introduce a delay of specified seconds before consuming an element of the
1241-- stream except the first one.
1243-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1244-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1245-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1246-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1248-- @since 0.8.0
1250{-# INLINE delay #-}
1251delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1252delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
1254-- Note: delay must be serial.
1256-- | Introduce a delay of specified seconds after consuming an element of a
1257-- stream.
1259-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPost 1 $ Stream.enumerateFromTo 1 3
1260-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1261-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1262-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1264-- /Pre-release/
1266{-# INLINE delayPost #-}
1267delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1268delayPost n = intersperseMSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
1270-- Note: delay must be serial, that's why 'trace_' is used.
1272-- | Introduce a delay of specified seconds before consuming an element of a
1273-- stream.
1275-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPre 1 $ Stream.enumerateFromTo 1 3
1276-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1277-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1278-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1280-- /Pre-release/
1282{-# INLINE delayPre #-}
1283delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1284delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
1286------------------------------------------------------------------------------
1287-- Reorder in sequence
1288------------------------------------------------------------------------------
1290-- | Buffer until the next element in sequence arrives. The function argument
1291-- determines the difference in sequence numbers. This could be useful in
1292-- implementing sequenced streams, for example, TCP reassembly.
1294-- /Unimplemented/
1296{-# INLINE reassembleBy #-}
1297reassembleBy
1298 :: -- (IsStream t, Monad m) =>
1299 Fold m a b
1300 -> (a -> a -> Int)
1301 -> t m a
1302 -> t m b
1303reassembleBy = undefined
1305------------------------------------------------------------------------------
1306-- Position Indexing
1307------------------------------------------------------------------------------
1309-- |
1310-- > indexed = Stream.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
1311-- > indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
1313-- Pair each element in a stream with its index, starting from index 0.
1315-- >>> Stream.toList $ Stream.indexed $ Stream.fromList "hello"
1316-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1318-- @since 0.6.0
1319{-# INLINE indexed #-}
1320indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1321indexed = fromStreamD . D.indexed . toStreamD
1323-- |
1324-- > indexedR n = Stream.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
1325-- > indexedR n = Stream.zipWith (,) (Stream.enumerateFromThen n (n - 1))
1327-- Pair each element in a stream with its index, starting from the
1328-- given index @n@ and counting down.
1330-- >>> Stream.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
1331-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
1333-- @since 0.6.0
1334{-# INLINE indexedR #-}
1335indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1336indexedR n = fromStreamD . D.indexedR n . toStreamD
1338-------------------------------------------------------------------------------
1339-- Time Indexing
1340-------------------------------------------------------------------------------
1342-- Note: The timestamp stream must be the second stream in the zip so that the
1343-- timestamp is generated after generating the stream element and not before.
1344-- If we do not do that then the following example will generate the same
1345-- timestamp for first two elements:
1347-- Stream.mapM_ print $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
1349-- | Pair each element in a stream with an absolute timestamp, using a clock of
1350-- specified granularity. The timestamp is generated just before the element
1351-- is consumed.
1353-- >>> Stream.mapM_ print $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1354-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1355-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1356-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1358-- /Pre-release/
1360{-# INLINE timestampWith #-}
1361timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
1362 => Double -> t m a -> t m (AbsTime, a)
1363timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
1365-- TBD: check performance vs a custom implementation without using zipWith.
1367-- /Pre-release/
1369{-# INLINE timestamped #-}
1370timestamped :: (IsStream t, MonadAsync m, Functor (t m))
1371 => t m a -> t m (AbsTime, a)
1372timestamped = timestampWith 0.01
1374-- | Pair each element in a stream with relative times starting from 0, using a
1375-- clock with the specified granularity. The time is measured just before the
1376-- element is consumed.
1378-- >>> Stream.mapM_ print $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1379-- (RelTime64 (NanoSecond64 ...),1)
1380-- (RelTime64 (NanoSecond64 ...),2)
1381-- (RelTime64 (NanoSecond64 ...),3)
1383-- /Pre-release/
1385{-# INLINE timeIndexWith #-}
1386timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
1387 => Double -> t m a -> t m (RelTime64, a)
1388timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
1390-- | Pair each element in a stream with relative times starting from 0, using a
1391-- 10 ms granularity clock. The time is measured just before the element is
1392-- consumed.
1394-- >>> Stream.mapM_ print $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1395-- (RelTime64 (NanoSecond64 ...),1)
1396-- (RelTime64 (NanoSecond64 ...),2)
1397-- (RelTime64 (NanoSecond64 ...),3)
1399-- /Pre-release/
1401{-# INLINE timeIndexed #-}
1402timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
1403 => t m a -> t m (RelTime64, a)
1404timeIndexed = timeIndexWith 0.01
1406------------------------------------------------------------------------------
1407-- Searching
1408------------------------------------------------------------------------------
1410-- | Find all the indices where the value of the element in the stream is equal
1411-- to the given value.
1413-- > elemIndices a = findIndices (== a)
1415-- @since 0.5.0
1416{-# INLINE elemIndices #-}
1417elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1418elemIndices a = findIndices (== a)
1420------------------------------------------------------------------------------
1421-- Rolling map
1422------------------------------------------------------------------------------
1424-- XXX this is not a one-to-one map so calling it map may not be right.
1425-- We can perhaps call it zipWithTail or rollWith.
1427-- | Apply a function on every two successive elements of a stream. The first
1428-- argument of the map function is the previous element and the second argument
1429-- is the current element. When the current element is the first element, the
1430-- previous element is 'Nothing'.
1432-- /Pre-release/
1434{-# INLINE rollingMap #-}
1435rollingMap :: (IsStream t, Monad m) => (Maybe a -> a -> b) -> t m a -> t m b
1436rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
1438-- | Like 'rollingMap' but with an effectful map function.
1440-- /Pre-release/
1442{-# INLINE rollingMapM #-}
1443rollingMapM :: (IsStream t, Monad m) => (Maybe a -> a -> m b) -> t m a -> t m b
1444rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
1446-- | Like 'rollingMap' but requires at least two elements in the stream,
1447-- returns an empty stream otherwise.
1449-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
1451-- /Pre-release/
1453{-# INLINE rollingMap2 #-}
1454rollingMap2 :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
1455rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
1457------------------------------------------------------------------------------
1458-- Maybe Streams
1459------------------------------------------------------------------------------
1461-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1462-- elements, and return a stream of values extracted from 'Just'.
1464-- Equivalent to:
1466-- @
1467-- mapMaybe f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.map f
1468-- @
1470-- @since 0.3.0
1471{-# INLINE mapMaybe #-}
1472mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
1473mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
1475-- | Like 'mapMaybe' but maps a monadic function.
1477-- Equivalent to:
1479-- @
1480-- mapMaybeM f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.mapM f
1481-- @
1483-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
1485-- @since 0.3.0
1486{-# INLINE_EARLY mapMaybeM #-}
1487mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
1488 => (a -> m (Maybe b)) -> t m a -> t m b
1489mapMaybeM f = fmap fromJust . filter isJust . mapM f
1491{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
1492{-# INLINE mapMaybeMSerial #-}
1493mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
1494mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
1496-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
1498-- /Pre-release/
1500{-# INLINE catMaybes #-}
1501catMaybes :: (IsStream t, Monad m, Functor (t m)) => t m (Maybe a) -> t m a
1502catMaybes = fmap fromJust . filter isJust
1504------------------------------------------------------------------------------
1505-- Either streams
1506------------------------------------------------------------------------------
1508-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1510-- /Pre-release/
1512{-# INLINE lefts #-}
1513lefts :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a
1514lefts = fmap (fromLeft undefined) . filter isLeft
1516-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1518-- /Pre-release/
1520{-# INLINE rights #-}
1521rights :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b
1522rights = fmap (fromRight undefined) . filter isRight
1524-- | Remove the either wrapper and flatten both lefts and as well as rights in
1525-- the output stream.
1527-- /Pre-release/
1529{-# INLINE both #-}
1530both :: Functor (t m) => t m (Either a a) -> t m a
1531both = fmap (either id id)
1533------------------------------------------------------------------------------
1534-- Concurrent Application
1535------------------------------------------------------------------------------
1537-- | Parallel transform application operator; applies a stream transformation
1538-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
1539-- is evaluated asynchronously in an independent thread yielding elements to a
1540-- buffer and the transformation function runs in another thread consuming the
1541-- input from the buffer. '|$' is just like regular function application
1542-- operator '$' except that it is concurrent.
1544-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
1545-- look at it as a transformation that converts a transform function to a
1546-- buffered concurrent transform function.
1548-- The following code prints a value every second even though each stage adds a
1549-- 1 second delay.
1552-- >>> :{
1553-- Stream.drain $
1554-- Stream.mapM (\x -> threadDelay 1000000 >> print x)
1555-- |$ Stream.replicateM 3 (threadDelay 1000000 >> return 1)
1556-- :}
1557-- 1
1558-- 1
1559-- 1
1561-- /Concurrent/
1563-- /Since: 0.3.0 ("Streamly")/
1565-- @since 0.8.0
1566{-# INLINE (|$) #-}
1567(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
1568-- (|$) f = f . mkAsync
1569(|$) f = f . mkParallel
1571infixr 0 |$
1573-- | Same as '|$'.
1575-- /Internal/
1577{-# INLINE applyAsync #-}
1578applyAsync :: (IsStream t, MonadAsync m)
1579 => (t m a -> t m b) -> (t m a -> t m b)
1580applyAsync = (|$)
1582-- | Same as '|$' but with arguments reversed.
1584-- (|&) = flip (|$)
1586-- /Concurrent/
1588-- /Since: 0.3.0 ("Streamly")/
1590-- @since 0.8.0
1591{-# INLINE (|&) #-}
1592(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
1593x |& f = f |$ x
1595infixl 1 |&