@@ -28,6 +28,10 @@
28-- provided for convenience and for consistency with other pure APIs in the
28-- provided for convenience and for consistency with other pure APIs in the
29-- @base@ package.
29-- @base@ package.
31+-- In many cases, short definitions of the combinators are provided in the
32+-- documentation for illustration. The actual implementation may differ for
33+-- performance reasons.
31-- Functions having a 'MonadAsync' constraint work concurrently when used with
35-- Functions having a 'MonadAsync' constraint work concurrently when used with
32-- appropriate stream type combinator. Please be careful to not use 'parallely'
36-- appropriate stream type combinator. Please be careful to not use 'parallely'
33-- with infinite streams.
37-- with infinite streams.
@@ -41,74 +45,128 @@
41module Streamly.Prelude
45module Streamly.Prelude
43 -- * Construction
47 -- * Construction
44− -- | All other stream construction and generation combinators described
48+ -- ** Primitives
45− -- later, and even more custom combinators can be expressed in terms of
46− -- these primitives. However, the special versions provided in this module
47− -- can be more efficient in some situations.
50 -- | Primitives to construct a stream from pure values or monadic actions.
49 -- | Primitives to construct a stream from pure values or monadic actions.
50+ -- All other stream construction and generation combinators described later
51+ -- can be expressed in terms of these primitives. However, the special
52+ -- versions provided in this module can be much more efficient in most
53+ -- cases. Users can create custom combinators using these primitives.
63+ -- | Generate a monadic stream from a seed value or values.
60− -- ** From Streams
71+ -- ** Enumeration
61− -- | You can construct streams by appending or merging existing streams.
72+ -- | The haskell 'Enum' typeclass provides enumerations for member types.
62− -- When constructing streams from streams, '<>' and 'mempty' are the
73+ -- However, the typeclass is tied to lists, it is impossible to use it to
63− -- intuitive equivalents of 'K.cons' and 'K.nil', respectively. These
74+ -- write enumeration functions producing something other than a list in a
64− -- primitives can be very useful when constructing your own custom stream
75+ -- generic way.
65− -- combinators. Also see the variants of '<>' defined in the "Streamly"
76+ -- We can use the 'Enum' facilities to enumerate the type producing a list
66− -- module. Note that appending streams is inexpensive, it is much more
77+ -- and then convert it to a stream:
67− -- efficient than appending lists.
78+ --
69− -- * Generation
80+ -- 'fromList' $ 'Prelude.enumFromThen' from then
70− -- ** Unfold and Iterate
81+ -- @
71− -- | Note that the generative steps of unfold and iterate are inherently
82+ --
72− -- serial as the next step depends on the result of the previous step.
83+ -- However, this is not particularly efficient. Note that using
73− -- However, consumption of the result from the previous step can happen in
84+ -- enumeration functions e.g. 'Prelude.enumFromThen' is slightly faster
74− -- parallel with the generation by the next step.
85+ -- than the idioms like @[from, then..]@.
87+ -- For the cases where speed matters, we have provided specific 'Enum' like
88+ -- implementations for enumeration.
102+ -- ** From Generators
103+ -- | Generate a monadic stream from a seed value and a generator function.
75 , unfoldr
104 , unfoldr
76 , unfoldrM
105 , unfoldrM
77 , iterate
106 , iterate
78 , iterateM
107 , iterateM
80− -- ** Replicate and Repeat
111+ -- ** From Containers
81− -- | Generate a monadic stream from a seed value or function. Note that
82− -- these functions can generate a stream fully concurrently as, unlike
83− -- unfolds, there is no dependency between steps, therefore, an unbounded
84− -- number of steps can run concurrently. All of these can be expressed in
85− -- terms of 'K.cons' and 'K.nil' primitives.
92 -- | Convert an input structure, container or source into a stream. All of
112 -- | Convert an input structure, container or source into a stream. All of
93 -- these can be expressed in terms of primitives.
113 -- these can be expressed in terms of primitives.
96 , P.fromList
114 , P.fromList
97 , fromListM
115 , fromListM
98 , K.fromFoldable
116 , K.fromFoldable
99 , fromFoldableM
117 , fromFoldableM
102− -- * Deconstruction
119+ -- ** From External Containers
103− , uncons
120+ , fromHandle
105 -- * Elimination
122 -- * Elimination
125+ -- | It is easy to express all the folds in terms of the 'uncons' primitive,
126+ -- however the specific implementations provided later are generally more
127+ -- efficient. Folds are inherently serial as each step needs to use the
128+ -- result of the previous step.
107 -- ** General Folds
131 -- ** General Folds
108− -- | All the folds can be implemented in terms of 'uncons', however the
132+-- | Right and left folds.
109− -- specific implementations provided here are generally more efficient.
133+-- As a simple rule, always use lazy right fold for construction and strict
110− -- Folds are inherently serial as each step needs to use the result of
134+-- left fold for reduction. By construction we mean using a constructor as the
111− -- the previous step.
135+-- outermost operation in the fold function, by reduction we mean using a
136+-- function as the outermost operation in the fold function.
138+-- +-----------------------------------+--------------------------------------+
139+-- | Right Fold | Left Fold |
140+-- +===================================+======================================+
141+-- | Construction consumes input | Construction consumes all input, |
142+-- | lazily and streams it in FIFO | and constructs in reverse (LIFO) |
144+-- +-----------------------------------+--------------------------------------+
145+-- | Reduction ends up buffering all | Strict reduction works |
146+-- | input before it can be reduced | incrementally, without buffering. |
147+-- +-----------------------------------+--------------------------------------+
149+-- Almost always, we need lazy construction and strict reduction, therefore,
150+-- strict @foldr@ and lazy @foldl@ are rarely useful. If needed, strict @foldr@
151+-- and lazy @foldl@ can be expressed in terms of the available versions. For
152+-- example, a lazy @foldl@ can be replaced by a strict @foldl@ to reverse the
153+-- structure followed by a @foldr@.
155+-- The following equations may help understand the relation between the two
159+-- foldr f z xs = foldl (flip f) z (reverse xs)
160+-- foldl f z xs = foldr (flip f) z (reverse xs)
166+-- foldl f z xs = foldr g id xs z where g x k = k . flip f x
167+-- foldr f z xs = foldl g id xs z where g k x = k . f x
@@ -118,108 +176,113 @@ module Streamly.Prelude
121− -- ** Specialized Folds
179+ -- ** To Elements
122− -- | These folds can be expressed in terms of the general fold routines but
180+ -- | Folds that extract selected elements of a stream or their properties.
123− -- the special versions here can be more efficient in many cases.
181+ , (!!)
125− -- Filtering folds: extract parts of the stream
131− -- Substreams
190+ -- ** To Parts
132− , isPrefixOf
191+ -- | Folds that extract selected parts of a stream.
133− , isSubsequenceOf
192+ , tail
134− , stripPrefix
193+ , init
136− -- Conditional folds: may terminate early based on a condition
195+ -- ** To Boolean
196+ -- | Folds that summarize the stream to a boolean value.
140 , notElem
199 , notElem
155− -- Full folds - need to go through all elements
205+ -- ** To Summary
206+ -- | Folds that summarize the stream to a single value.
162 , product
209 , product
164− -- ** Fold To
211+ -- ** To Summary (Maybe)
212+ -- | Folds that summarize a non-empty stream to a 'Just' value and return
213+ -- 'Nothing' for an empty stream.
165 -- | Convert or divert a stream into an output structure, container or
221 -- | Convert or divert a stream into an output structure, container or
168 , toHandle
224 , toHandle
170 -- * Transformation
226 -- * Transformation
171− -- | One to one transformations, each element in the input stream is
172− -- transformed into a corresponding element in the output stream.
173− -- Therefore, the length of the stream and the ordering of elements in the
174− -- stream remains unchanged after the transformation.
176 -- ** Scanning
228 -- ** Scanning
177− -- | Scan is a transformation by continuously folding the result with the
229+ -- | Scans stream all the intermediate reduction steps of the corresponding
178− -- next element of the stream. This is the generalized way to transform a
230+ -- folds. The following equations hold for lists:
179− -- stream carrying state from previous transformation steps, other forms of
231+ --
180− -- transformation like map can be expressed in terms of this.
232+ -- > scanl f z xs == map (foldl f z) $ inits xs
233+ -- > scanr f z xs == map (foldr f z) $ tails
235+ -- We do not provide a right associative scan, it can be recovered from a
236+ -- 'scanl'' as follows:
238+ -- > scanr f z xs == reverse $ scanl' (flip f) z (reverse xs)
240+ -- Scan is like a stateful map. If we discard the state, we get the map:
242+ -- > S.drop 1 $ S.scanl' (\_ x -> f x) z xs == map f xs
244+ -- > S.postscanl' (\_ x -> f x) z xs == map f xs
182 , scanlM'
247 , scanlM'
183− , scanlM
248+ -- , postscanl'
184− , scanl
249+ -- , postscanlM'
185− , scanl1M'
250+ -- , prescanl'
186 , scanl1'
252 , scanl1'
187− , scanl1M
253+ , scanl1M'
191 -- ** Mapping
256 -- ** Mapping
192− -- | Map is a special form of scan where no state is carried from one step
257+ -- | Map is a strictly one-to-one transformation of stream elements. It
193− -- to the next.
258+ -- cannot add or remove elements from the stream, just transforms them.
194 , Serial.map
259 , Serial.map
197 -- ** Flattening
261 -- ** Flattening
198 , sequence
262 , sequence
200− -- * Filtering and Insertion
201− -- | Adding or removing elements from the stream thus changing the length
204 -- ** Filtering
265 -- ** Filtering
266+ -- | Filtering may remove some elements from the stream.
206 , filterM
269 , filterM
209 , takeWhile
271 , takeWhile
210 , takeWhileM
272 , takeWhileM
212 , dropWhile
274 , dropWhile
213 , dropWhileM
275 , dropWhileM
215− -- ** Inserting
279+ -- ** Insertion
216− , intersperseM
280+ -- | Insertion adds more elements to the stream.
219− -- ** Deleting
282+ , insertBy
220− , deleteBy
283+ , intersperseM
222− -- * Reordering
285+ -- ** Reordering
223 , reverse
286 , reverse
225 -- * Hybrid Operations
288 -- * Hybrid Operations
@@ -234,20 +297,51 @@ module Streamly.Prelude
234 , findIndices
297 , findIndices
235 , elemIndices
298 , elemIndices
300+ -- * Multi-Stream Operations
301+ -- | New streams can be constructed by appending, merging or zipping
302+ -- existing streams.
305+ -- | Streams form a 'Semigroup' and a 'Monoid' under the append
309+ -- >> S.toList $ S.fromList [1,2] \<> S.fromList [3,4]
311+ -- >> S.toList $ fold $ [S.fromList [1,2], S.fromList [3,4]]
316+ -- | Streams form a commutative semigroup under the merge
237− -- * Zipping
325+ -- ** Zipping
240 , zipWith
326 , zipWith
241 , zipWithM
327 , zipWithM
242 , Z.zipAsyncWith
328 , Z.zipAsyncWith
243 , Z.zipAsyncWithM
329 , Z.zipAsyncWithM
245− -- * Merging
331+ -- Special zips
246− , mergeBy
332+ , indexed
247− , merge
333+ , indexedR
249− -- * Transformation comprehensions
335+ -- ** Flattening
250− , the
336+ , concatMapM
252 -- * Deprecated
346 -- * Deprecated
@@ -270,6 +364,7 @@ import qualified Prelude
270import qualified System.IO as IO
364import qualified System.IO as IO
272import Streamly.SVar (MonadAsync, defState, rstState)
366import Streamly.SVar (MonadAsync, defState, rstState)
367+import Streamly.Streams.Async (mkAsync')
273import Streamly.Streams.Combinators (maxYields)
368import Streamly.Streams.Combinators (maxYields)
274import Streamly.Streams.Prelude (fromStreamS, toStreamS)
369import Streamly.Streams.Prelude (fromStreamS, toStreamS)
275import Streamly.Streams.StreamD (fromStreamD, toStreamD)
370import Streamly.Streams.StreamD (fromStreamD, toStreamD)
@@ -306,10 +401,18 @@ uncons m = K.uncons (K.adapt m)
306-- Generation by Unfolding
401-- Generation by Unfolding
307------------------------------------------------------------------------------
402------------------------------------------------------------------------------
408+-- Nothing -> 'K.nil'
409+-- Just (a, b) -> a \`cons` unfoldr step b
309−-- | Build a stream by unfolding a /pure/ step function starting from a seed.
412+-- Build a stream by unfolding a /pure/ step function @step@ starting from a
310−-- The step function returns the next element in the stream and the next seed
413+-- seed @s@. The step function returns the next element in the stream and the
311−-- value. When it is done it returns 'Nothing' and the stream ends. For
414+-- next seed value. When it is done it returns 'Nothing' and the stream ends.
312−-- example,
415+-- For example,
315-- let f b =
418-- let f b =
@@ -322,15 +425,6 @@ uncons m = K.uncons (K.adapt m)
322-- [0,1,2,3]
425-- [0,1,2,3]
325−-- unfoldr can be expressed in terms of 'yield' and '<>' as follows:
331−-- Just (a, b) -> 'yield' a '<>' (unfoldr step b)
334-- @since 0.1.0
428-- @since 0.1.0
335{-# INLINE_EARLY unfoldr #-}
429{-# INLINE_EARLY unfoldr #-}
336unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
430unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
@@ -384,20 +478,36 @@ unfoldrMSerial step seed = fromStreamS (S.unfoldrM step seed)
385-- Faster than yieldM because there is no bind.
479-- Faster than yieldM because there is no bind.
483+-- yield a = a \`cons` nil
387−-- | Create a singleton stream from a pure value. Same as @a `cons` nil@ but
486+-- Create a singleton stream from a pure value.
388−-- slighly more efficient. Note that in monadic streams, 'yield' is the same
487+--
488+-- The following holds in monadic streams, but not in Zip streams:
492+-- yield = yieldM . pure
389−-- as 'pure' or 'return', however, in Zip applicative streams it is not the
495+-- In Zip applicative streams 'yield' is not the same as 'pure' because in that
390−-- same as 'pure' because in that case 'pure' is equivalent to 'repeat'
496+-- case 'pure' is equivalent to 'repeat' instead. 'yield' and 'pure' are
391−-- instead. In all other stream types, 'yield' is the same as @yieldM . pure@
497+-- equally efficient, in other cases 'yield' may be slightly more efficient
392−-- but more efficient.
498+-- than the other equivalent definitions.
394-- @since 0.4.0
500-- @since 0.4.0
395{-# INLINE yield #-}
501{-# INLINE yield #-}
396yield :: IsStream t => a -> t m a
502yield :: IsStream t => a -> t m a
397yield = K.yield
503yield = K.yield
507+-- yieldM m = m \`consM` nil
399−-- | Create a singleton stream from a monadic action. Same as @m \`consM` nil@
510+-- Create a singleton stream from a monadic action.
400−-- but more efficient.
403-- > toList $ yieldM getLine
513-- > toList $ yieldM getLine
@@ -410,24 +520,231 @@ yield = K.yield
410yieldM :: (Monad m, IsStream t) => m a -> t m a
520yieldM :: (Monad m, IsStream t) => m a -> t m a
411yieldM = K.yieldM
521yieldM = K.yieldM
413−-- | Generate a stream of given length, from a given pure
523+-- | @intFromStep from step@ generates an infinite stream whose first element
414−-- function of its indices.
524+-- is @from@ and the successive elements are in increments of @step@. This does
415−{-# INLINE generate #-}
525+-- not check for overflow or underflow if the 'Integral' type is bounded.
528+-- > S.toList $ S.take 4 $ S.intFromStep 0 2
530+-- > S.toList $ S.take 3 $ S.intFromStep 0 (-2)
535+{-# INLINE intFromStep #-}
416−generate :: (IsStream t, Monad m) => Int -> (Int -> a) -> t m a
536+intFromStep :: (IsStream t, Monad m, Integral a) => a -> a -> t m a
417−generate n = fromStreamD . D.generate n
537+intFromStep from stride = fromStreamD $ D.intFromStep from stride
419−-- | Generate a stream of given length, from a given monadic
539+-- | Enumerate an 'Integral' type. @intFrom from@ generates a stream whose
420−-- function of its indices.
540+-- first element is @from@ and the successive elements are in increments of
541+-- @1@. The stream is bounded by the size of the 'Integral' type.
544+-- > S.toList $ S.take 4 $ S.intFrom (0 :: Int)
422−{-# INLINE generateM #-}
549+{-# INLINE intFrom #-}
423−generateM :: (IsStream t, Monad m) => Int -> (Int -> m a) -> t m a
550+intFrom :: (IsStream t, Monad m, Integral a, Bounded a) => a -> t m a
424−generateM n = fromStreamD . D.generateM n
551+intFrom from = fromStreamD $ D.intFrom from
426−-- | Generate a stream by performing a monadic action @n@ times. Can be
553+-- | Enumerate an 'Integral' type in steps. @intFromThen from then@ generates a
427−-- expressed as @stimes n (yieldM m)@.
554+-- stream whose first element is @from@, the second element is @then@ and the
555+-- successive elements are in increments of @then - from@. The stream is
556+-- bounded by the size of the 'Integral' type.
559+-- > S.toList $ S.take 4 $ S.intFromThen (0 :: Int) 2
561+-- > S.toList $ S.take 4 $ S.intFromThen (0 :: Int) (-2)
566+{-# INLINE intFromThen #-}
567+intFromThen :: (IsStream t, Monad m, Integral a, Bounded a) => a -> a -> t m a
568+intFromThen from next = fromStreamD $ D.intFromThen from next
570+-- | Enumerate an 'Integral' type up to a given limit. @intFromTo from to@
571+-- generates a finite stream whose first element is @from@ and successive
572+-- elements are in increments of @1@ up to @to@.
575+-- > S.toList $ S.intFromTo 0 4
580+{-# INLINE intFromTo #-}
581+intFromTo :: (IsStream t, Monad m, Integral a) => a -> a -> t m a
582+intFromTo from to = fromStreamD $ D.intFromTo from to
584+-- | Enumerate an 'Integral' type in steps up to a given limit. @intFromThenTo
585+-- from then to@ generates a finite stream whose first element is @from@, the
586+-- second element is @then@ and the successive elements are in increments of
587+-- @then - from@ up to @to@.
590+-- > S.toList $ S.intFromThenTo 0 2 6
592+-- > S.toList $ S.intFromThenTo 0 (-2) (-6)
597+{-# INLINE intFromThenTo #-}
598+intFromThenTo :: (IsStream t, Monad m, Integral a) => a -> a -> a -> t m a
599+intFromThenTo from next to = fromStreamD $ D.intFromThenTo from next to
601+-- | @numFromStep from step@ generates an infinite stream whose first element
602+-- is @from@ and the successive elements are in increments of @step@. This is
603+-- numerically stable but does not check for overflow or underflow for bounded
604+-- types. Note, for 'Integral' types 'intFromStep' is faster.
608+-- > S.toList $ S.take 4 $ S.numFromStep 0.1 2
610+-- > S.toList $ S.take 3 $ S.numFromStep 0.1 (-2)
611+-- [0.1,-1.9,-3.9,-5.9]
615+{-# INLINE numFromStep #-}
616+numFromStep :: (IsStream t, Monad m, Num a) => a -> a -> t m a
617+numFromStep from stride = fromStreamD $ D.numFromStep from stride
619+-- Even though the underlying implementation of fracFrom and fracFromThen works
620+-- for any 'Num' we have restricted these to 'Fractional' because these do not
621+-- perform any bounds check, in contrast to int versions and are therefore not
622+-- equivalent substitutes for those.
624+-- | Numerically stable enumeration from a 'Fractional' number in steps of size
625+-- @1@. @fracFrom from@ generates a stream whose first element is @from@ and
626+-- the successive elements are in increments of @1@. No overflow or underflow
627+-- checks are performed.
629+-- This is the equivalent to 'enumFrom' for 'Fractional' types. For example:
632+-- > S.toList $ S.take 4 $ S.fracFrom 1.1
638+{-# INLINE fracFrom #-}
639+fracFrom :: (IsStream t, Monad m, Fractional a) => a -> t m a
640+fracFrom from = fromStreamD $ D.numFrom from
642+-- | Numerically stable enumeration from a 'Fractional' number in steps.
643+-- @fracFromThen from then@ generates a stream whose first element is @from@,
644+-- the second element is @then@ and the successive elements are in increments
645+-- of @then - from@. No overflow or underflow checks are performed.
647+-- This is the equivalent of 'enumFromThen' for 'Fractional' types. For
651+-- > S.toList $ S.take 4 $ S.fracFromThen 1.1 2.1
653+-- > S.toList $ S.take 4 $ S.fracFromThen 1.1 (-2.1)
654+-- [1.1,-2.1,-5.300000000000001,-8.500000000000002]
658+{-# INLINE fracFromThen #-}
659+fracFromThen :: (IsStream t, Monad m, Fractional a) => a -> a -> t m a
660+fracFromThen from next = fromStreamD $ D.numFromThen from next
662+-- | Numerically stable enumeration from a 'Fractional' number to a given
663+-- limit. @fracFromTo from to@ generates a finite stream whose first element
664+-- is @from@ and successive elements are in increments of @1@ up to @to@.
666+-- This is the equivalent of 'enumFromTo' for 'Fractional' types. For
670+-- > S.toList $ S.fracFromTo 1.1 4
672+-- > S.toList $ S.fracFromTo 1.1 4.6
673+-- [1.1,2.1,3.1,4.1,5.1]
676+-- Notice that the last element is equal to the specified @to@ value after
677+-- rounding to the nearest integer.
680+{-# INLINE fracFromTo #-}
681+fracFromTo :: (IsStream t, Monad m, Fractional a, Ord a) => a -> a -> t m a
682+fracFromTo from to = fromStreamD $ D.fracFromTo from to
684+-- | Numerically stable enumeration from a 'Fractional' number in steps up to a
685+-- given limit. @fracFromThenTo from then to@ generates a finite stream whose
686+-- first element is @from@, the second element is @then@ and the successive
687+-- elements are in increments of @then - from@ up to @to@.
689+-- This is the equivalent of 'enumFromThenTo' for 'Fractional' types. For
693+-- > S.toList $ S.fracFromThenTo 0.1 2 6
694+-- [0.1,2.0,3.9,5.799999999999999]
695+-- > S.toList $ S.fracFromThenTo 0.1 (-2) (-6)
696+-- [0.1,-2.0,-4.1000000000000005,-6.200000000000001]
701+{-# INLINE fracFromThenTo #-}
703+ :: (IsStream t, Monad m, Fractional a, Ord a)
704+ => a -> a -> a -> t m a
705+fracFromThenTo from next to = fromStreamD $ D.fracFromThenTo from next to
709+-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0
712+-- Generate an infinite stream, whose values are the output of a function @f@
713+-- applied on the corresponding index. Index starts at 0.
716+-- > S.toList $ S.take 5 $ S.fromIndices id
721+{-# INLINE fromIndices #-}
722+fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
723+fromIndices = fromStreamD . D.fromIndices
725+-- XXX this needs to be concurrent
729+-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0
732+-- Generate an infinite stream, whose values are the output of a monadic
733+-- function @f@ applied on the corresponding index. Index starts at 0.
736+{-# INLINE fromIndicesM #-}
737+fromIndicesM :: (IsStream t, Monad m) => (Int -> m a) -> t m a
738+fromIndicesM = fromStreamD . D.fromIndicesM
742+-- replicateM = take n . repeatM
745+-- Generate a stream by performing a monadic action @n@ times. Same as:
431-- runStream $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
748-- runStream $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
432-- runStream $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
749-- runStream $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
@@ -435,20 +752,33 @@ generateM n = fromStreamD . D.generateM n
435-- /Concurrent/
752-- /Concurrent/
437-- @since 0.1.1
754-- @since 0.1.1
755+{-# INLINE_EARLY replicateM #-}
438replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
756replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
439−replicateM n m = go n
757+replicateM = K.replicateM
441− go cnt = if cnt <= 0 then K.nil else m |: go (cnt - 1)
759+{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
760+{-# INLINE replicateMSerial #-}
761+replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
762+replicateMSerial n = fromStreamS . S.replicateM n
443−-- | Generate a stream by replicating the value @n@ times.
764+-- |
766+-- replicate = take n . repeat
769+-- Generate a stream of length @n@ by repeating a value @n@ times.
445−replicate :: IsStream t => Int -> a -> t m a
772+replicate :: (IsStream t, Monad m) => Int -> a -> t m a
446−replicate n a = go n
773+replicate n = fromStreamS . S.replicate n
448− go cnt = if cnt <= 0 then K.nil else a `K.cons` go (cnt - 1)
777+-- repeatM = fix . consM
778+-- repeatM = cycle1 . yieldM
450−-- | Generate a stream by repeatedly executing a monadic action forever. Can be
781+-- Generate a stream by repeatedly executing a monadic action forever.
451−-- expressed as @cycle1 . yieldM@.
454-- runStream $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
784-- runStream $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
@@ -462,7 +792,19 @@ repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
462repeatM = go
792repeatM = go
463 where go m = m |: go m
793 where go m = m |: go m
465−-- | Iterate a pure function from a seed value, streaming the results forever.
795+-- |
797+-- iterate f x = x \`cons` iterate f x
800+-- Generate an infinite stream with @x@ as the first element and each
801+-- successive element derived by applying the function @f@ on the previous
805+-- > S.toList $ S.take 5 $ S.iterate (+1) 1
467-- @since 0.1.2
809-- @since 0.1.2
468iterate :: IsStream t => (a -> a) -> a -> t m a
810iterate :: IsStream t => (a -> a) -> a -> t m a
@@ -470,8 +812,14 @@ iterate step = fromStream . go
471 go s = K.cons s (go (step s))
813 go s = K.cons s (go (step s))
473−-- | Iterate a monadic function from a seed value, streaming the results
815+-- |
817+-- iterateM f m = m \`consM` iterateM f m
820+-- Generate an infinite stream with the first element generated by the action
821+-- @m@ and each successive element derived by applying the monadic function
822+-- @f@ on the previous element.
476-- When run concurrently, the next iteration can run concurrently with the
824-- When run concurrently, the next iteration can run concurrently with the
477-- processing of the previous iteration. Note that more than one iteration
825-- processing of the previous iteration. Note that more than one iteration
@@ -500,9 +848,13 @@ iterateM step = go
500-- Conversions
848-- Conversions
501------------------------------------------------------------------------------
849------------------------------------------------------------------------------
853+-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'
503−-- | Construct a stream from a list containing monadic actions. More efficient
856+-- Construct a stream from a list of monadic actions. This is more efficient
504−-- list specific implementation of 'fromFoldableM' especially for serial
857+-- than 'fromFoldableM' for serial streams.
505−-- streams as it works well with fusion optimization.
507-- @since 0.4.0
859-- @since 0.4.0
508{-# INLINE_EARLY fromListM #-}
860{-# INLINE_EARLY fromListM #-}
@@ -511,8 +863,12 @@ fromListM = fromStreamD . D.fromListM
511{-# RULES "fromListM fallback to StreamK" [1]
863{-# RULES "fromListM fallback to StreamK" [1]
512 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
864 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
868+-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'
514−-- | Construct a stream from a 'Foldable' containing monadic actions. Same as
871+-- Construct a stream from a 'Foldable' containing monadic actions.
515−-- @'Prelude.foldr' 'consM' 'K.nil'@.
518-- runStream $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
874-- runStream $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
@@ -556,7 +912,7 @@ fromHandle h = fromStream go
556-- stream into a list:
912-- stream into a list:
559−-- >> runIdentity $ foldrM (\\x xs -> return (x : xs)) [] (serially $ fromFoldable [1,2,3])
915+-- >> S.foldrM (\\x xs -> return (x : xs)) [] $ fromList [1,2,3]
560-- [1,2,3]
916-- [1,2,3]
@@ -565,19 +921,74 @@ fromHandle h = fromStream go
565foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b
921foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b
566foldrM = P.foldrM
922foldrM = P.foldrM
568−-- | Lazy right associative fold. For example, to fold a stream into a list:
924+-- | Lazy right associative fold.
926+-- @foldr f z xs@ deconstructs @xs@ one element @x@ at a time, applying the
927+-- function @f@ to @x@ and the tail of the output. The tail recurses until the
928+-- input finishes and @z@ is used as the tail end.
571−-- >> runIdentity $ foldr (:) [] (serially $ fromFoldable [1,2,3])
931+-- foldr f z xs == x1 \`f` tail1
572−-- [1,2,3]
932+-- tail1 == x2 \`f` tail2
933+-- tail2 == x3 \`f` tail3
935+-- tailn == xn \`f` z
938+-- Resulting in a right associated expression:
941+-- foldr f z xs == x1 \`f` (x2 \`f` ...(xn \`f` z))
944+-- When the outermost operation in the fold function is a constructor, foldr
945+-- generates a structure that can be consumed lazily. For example:
948+-- > S.foldr (:) [] $ S.fromList [1,2,3,4]
955+-- 1 : (2 : (3 : tail))
956+-- 1 : (2 : (3 : (4 : [])))
959+-- When the outermost operation is a function @foldr@ results in a right
960+-- associated expression which cannot be reduced until the whole expression has
961+-- been built. Therefore, it ends up consuming the whole input, buffering the
962+-- whole expression in memory before reduction can start. For example:
965+-- > S.foldr (+) 0 $ S.fromList [1,2,3,4]
972+-- 1 + (2 + (3 + tail))
973+-- 1 + (2 + (3 + (4 + 0)))
976+-- In @foldr@, the output tail is the source of recursion, we can stop
977+-- recursion and terminate the fold by yielding a terminal value as tail:
979+-- >> S.foldr (\x rest -> if x == 3 then [] else x : rest) [] $ S.fromList [4,1,3,undefined]
982+-- The arguments to the folding function (@a -> b -> b@) are in the head and
983+-- tail order of the output, @a@ is the head and @b@ is the tail. Remember, in
984+-- a right fold the zero is on the right, it is the tail end.
575-- @since 0.1.0
986-- @since 0.1.0
576{-# INLINE foldr #-}
987{-# INLINE foldr #-}
577foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
988foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
578foldr = P.foldr
989foldr = P.foldr
580−-- | Right fold, for non-empty streams, using first element as the starting
991+-- | Lazy right fold for non-empty streams, using first element as the starting
581-- value. Returns 'Nothing' if the stream is empty.
992-- value. Returns 'Nothing' if the stream is empty.
583-- @since 0.5.0
994-- @since 0.5.0
@@ -603,6 +1014,85 @@ foldl = foldx
604-- | Strict left associative fold.
1015-- | Strict left associative fold.
1017+-- @foldl' f z xs@ deconstructs @xs@ one element @x@ at a time, applying the
1018+-- function @f@ to the output accumulated till now (the head of the expression)
1019+-- and @x@. The accumulator starts with the initial value @z@ and keeps
1020+-- accumulating elements until the input finishes.
1023+-- head1 == z \`f` x1
1024+-- head2 == head1 \`f` x2
1025+-- head3 == head2 \`f` x3
1027+-- foldl' f z xs == head(n-1) \`f` xn
1030+-- Recursively building a left associated expression:
1033+-- foldl' f z xs == (((z \`f` x1) \`f` x2) ...) \`f` xn
1036+-- When the outermost operation in the fold function is a (left associated)
1037+-- constructor, foldl' consumes the whole input to construct the new structure
1038+-- buffered in memory in the reverse order of the input. For example:
1041+-- > S.foldl' (flip (:)) [] $ S.fromList [1,2,3,4]
1045+-- When the outermost operation is a function, @foldl'@ results in a left
1046+-- associated expression which is incrementally reduced at each step of the
1047+-- fold, thus never building the whole expression in memory. For example:
1050+-- > S.foldl' (+) 0 $ S.fromList [1,2,3,4]
1057+-- ((0 + 1) + 2) + 3
1058+-- (((0 + 1) + 2) + 3) + 4
1061+-- 0 + 1 => reduce to 1
1062+-- 1 + 2 => reduce to 3
1063+-- 3 + 3 => reduce to 6
1064+-- 6 + 4 => reduce to 10
1067+-- We can stop recursion and terminate the fold by yielding an expression that
1068+-- is independent of the input element:
1070+-- >> S.foldl' (\acc x -> if acc >= 8 then acc else x + acc) 0 $ S.fromList [4,1,3,undefined]
1073+-- To stop on encountering the number 3 we will have to store the previous
1074+-- number in the accumulator so that we do not depend on the current input:
1077+-- > S.foldl' (\(acc, prev) x -> if prev == 3 then (acc,prev) else (x + acc,x))
1078+-- (0,0) $ S.fromList \[4,1,3,undefined]
1082+-- To map the fold operation to stateful or event-driven programming, we can
1083+-- consider @z@ as the initial state and the stream being folded as a stream of
1084+-- events, thus @foldl'@ processes all the events in the stream updating the
1085+-- state on each event and then ultimately returning the final state.
1087+-- The arguments to the folding function (@b -> a -> b@) are in the head and
1088+-- tail order of the output expression, @b@ is the head and @a@ is the tail.
1089+-- Remember, in a left fold the zero is on the left, at the head of the
1092+-- IMPORTANT: 'foldl'' evaluates the accumulator to WHNF. To avoid building
1093+-- lazy expressions inside the accumulator, and to help GHC optimize better, it
1094+-- is recommended that a strict data structure is used for accumulator.
606-- @since 0.2.0
1096-- @since 0.2.0
607{-# INLINE foldl' #-}
1097{-# INLINE foldl' #-}
608foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
1098foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
@@ -622,7 +1112,6 @@ foldl1' step m = do
622 res <- foldl' step h t
1112 res <- foldl' step h t
623 return $ Just res
1113 return $ Just res
625−-- XXX replace the recursive "go" with explicit continuations.
626-- | Like 'foldx', but with a monadic step function.
1115-- | Like 'foldx', but with a monadic step function.
628-- @since 0.2.0
1117-- @since 0.2.0
@@ -654,6 +1143,8 @@ null = K.null
655-- | Extract the first element of the stream, if any.
1144-- | Extract the first element of the stream, if any.
657-- @since 0.1.0
1148-- @since 0.1.0
658{-# INLINE head #-}
1149{-# INLINE head #-}
659head :: Monad m => SerialT m a -> m (Maybe a)
1150head :: Monad m => SerialT m a -> m (Maybe a)
@@ -675,6 +1166,8 @@ init m = K.init (K.adapt m)
676-- | Extract the last element of the stream, if any.
1167-- | Extract the last element of the stream, if any.
1169+-- > last xs = xs !! (length xs - 1)
678-- @since 0.1.1
1171-- @since 0.1.1
679{-# INLINE last #-}
1172{-# INLINE last #-}
680last :: Monad m => SerialT m a -> m (Maybe a)
1173last :: Monad m => SerialT m a -> m (Maybe a)
@@ -722,75 +1215,99 @@ any p m = S.any p (toStreamS m)
722and :: Monad m => SerialT m Bool -> m Bool
1215and :: Monad m => SerialT m Bool -> m Bool
723and = all (==True)
1216and = all (==True)
725−-- | Determines wheter at least one element of a boolean stream is True.
1218+-- | Determines whether at least one element of a boolean stream is True.
727-- @since 0.5.0
1220-- @since 0.5.0
728{-# INLINE or #-}
1221{-# INLINE or #-}
729or :: Monad m => SerialT m Bool -> m Bool
1222or :: Monad m => SerialT m Bool -> m Bool
730or = any (==True)
1223or = any (==True)
732−-- | Determine the sum of all elements of a stream of numbers
1225+-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
1226+-- the stream is empty. Note that this is not numerically stable for floating
734-- @since 0.1.0
1229-- @since 0.1.0
735{-# INLINE sum #-}
1230{-# INLINE sum #-}
736sum :: (Monad m, Num a) => SerialT m a -> m a
1231sum :: (Monad m, Num a) => SerialT m a -> m a
737sum = foldl' (+) 0
1232sum = foldl' (+) 0
739−-- | Determine the product of all elements of a stream of numbers
1234+-- | Determine the product of all elements of a stream of numbers. Returns @1@
1235+-- when the stream is empty.
741-- @since 0.1.1
1237-- @since 0.1.1
742{-# INLINE product #-}
1238{-# INLINE product #-}
743product :: (Monad m, Num a) => SerialT m a -> m a
1239product :: (Monad m, Num a) => SerialT m a -> m a
744product = foldl' (*) 1
1240product = foldl' (*) 1
1244+-- minimum = 'minimumBy' compare
746−-- | Determine the minimum element in a stream.
1247+-- Determine the minimum element in a stream.
748-- @since 0.1.0
1249-- @since 0.1.0
749{-# INLINE minimum #-}
1250{-# INLINE minimum #-}
750minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1251minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
751minimum m = S.minimum (toStreamS m)
1252minimum m = S.minimum (toStreamS m)
753−-- | Non-overloaded version of minimum.
1254+-- | Determine the minimum element in a stream using the supplied comparison
755{-# INLINE minimumBy #-}
1258{-# INLINE minimumBy #-}
756minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1259minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
757minimumBy cmp m = S.minimumBy cmp (toStreamS m)
1260minimumBy cmp m = S.minimumBy cmp (toStreamS m)
1264+-- maximum = 'maximumBy' compare
759−-- | Determine the maximum element in a stream.
1267+-- Determine the maximum element in a stream.
761{-# INLINE maximum #-}
1270{-# INLINE maximum #-}
762maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1271maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
763maximum m = S.maximum (toStreamS m)
1272maximum m = S.maximum (toStreamS m)
765−-- | Non-overloaded version of maximum.
1274+-- | Determine the maximum element in a stream using the supplied comparison
767{-# INLINE maximumBy #-}
1278{-# INLINE maximumBy #-}
768maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1279maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
769maximumBy cmp m = S.maximumBy cmp (toStreamS m)
1280maximumBy cmp m = S.maximumBy cmp (toStreamS m)
771−-- | Looks the element at the given index.
1282+-- | Lookup the element at the given index.
773{-# INLINE (!!) #-}
1285{-# INLINE (!!) #-}
774(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
1286(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
775−m !! i = (toStreamS m) S.!! i
1287+m !! i = toStreamS m S.!! i
777−-- | Looks the given key up, treating the given stream as an association list.
1289+-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1290+-- first pair where the key equals the given value @a@.
1292+-- > lookup = snd <$> find ((==) . fst)
779-- @since 0.5.0
1294-- @since 0.5.0
780{-# INLINE lookup #-}
1295{-# INLINE lookup #-}
781lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
1296lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
782lookup a m = S.lookup a (toStreamS m)
1297lookup a m = S.lookup a (toStreamS m)
784−-- | Returns the first element of the stream satisfying the given predicate,
1299+-- | Like 'findM' but with a non-monadic predicate.
1301+-- > find p = findM (return . p)
787-- @since 0.5.0
1303-- @since 0.5.0
788{-# INLINE find #-}
1304{-# INLINE find #-}
789find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
1305find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
790find p m = S.find p (toStreamS m)
1306find p m = S.find p (toStreamS m)
792−-- | Like 'find' but with a monadic predicate.
1308+-- | Returns the first element that satisfies the given predicate.
794{-# INLINE findM #-}
1311{-# INLINE findM #-}
795findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
1312findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
796findM p m = S.findM p (toStreamS m)
1313findM p m = S.findM p (toStreamS m)
@@ -803,8 +1320,7 @@ findM p m = S.findM p (toStreamS m)
803findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
1320findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
804findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
1321findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
806−-- | Gives the index of the first stream element satisfying the given
1323+-- | Returns the first index that satisfies the given predicate.
809-- @since 0.5.0
1325-- @since 0.5.0
810{-# INLINE findIndex #-}
1326{-# INLINE findIndex #-}
@@ -819,46 +1335,74 @@ findIndex p = head . findIndices p
819elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1335elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
820elemIndices a = findIndices (==a)
1336elemIndices a = findIndices (==a)
822−-- | Gives the first index of an element in the stream, which equals the given.
1338+-- | Returns the first index where a given value is found in the stream.
1340+-- > elemIndex a = findIndex (== a)
824-- @since 0.5.0
1342-- @since 0.5.0
825{-# INLINE elemIndex #-}
1343{-# INLINE elemIndex #-}
826elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
1344elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
827−elemIndex a = findIndex (==a)
1345+elemIndex a = findIndex (== a)
1347+-- | Map each element to a stream and then flatten the results into a single
1350+-- > concatMap f = concatMapM (return . f)
829{-# INLINE concatMap #-}
1353{-# INLINE concatMap #-}
830−concatMap :: Monad m => (a -> SerialT m b) -> SerialT m a -> SerialT m b
1354+concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
831concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
1355concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
1357+-- | Map each element to a stream using a monadic function and then flatten the
1358+-- results into a single stream.
833{-# INLINE concatMapM #-}
1361{-# INLINE concatMapM #-}
834−concatMapM :: Monad m => (a -> m (SerialT m b)) -> SerialT m a -> SerialT m b
1362+concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
835concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
1363concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
837------------------------------------------------------------------------------
1365------------------------------------------------------------------------------
838-- Substreams
1366-- Substreams
839------------------------------------------------------------------------------
1367------------------------------------------------------------------------------
841−-- | Takes two streams and returns 'True' iff the first stream is a prefix
1369+-- | Returns 'True' if the first stream is the same as or a prefix of the
842−-- of the second.
1370+-- second.
1373+-- > S.isPrefixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
844{-# INLINE isPrefixOf #-}
1378{-# INLINE isPrefixOf #-}
845isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1379isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
846isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
1380isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
848−-- | Takes two streams and returns 'True' if all the elements of the
1382+-- | Returns 'True' if all the elements of the first stream occur, in order, in
849−-- first stream occur, in order, in the second stream. The elements do
850−-- not have to occur consecutively.
1383+-- the second stream. The elements do not have to occur consecutively. A stream
1384+-- is treated as a subsequence of itself.
1387+-- > S.isSubsequenceOf (S.fromList "hlo") (S.fromList "hello" :: SerialT IO Char)
852{-# INLINE isSubsequenceOf #-}
1392{-# INLINE isSubsequenceOf #-}
853isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1393isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
854isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
1394isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
856−-- | Drops the given prefix from a stream. Returns Nothing if the stream
1396+-- | Drops the given prefix from a stream. Returns 'Nothing' if the stream does
857−-- doesn't start with the prefix given.
1397+-- not start with the given prefix. Returns @Just nil@ when the prefix is the
1398+-- same as the stream.
859{-# INLINE stripPrefix #-}
1401{-# INLINE stripPrefix #-}
860−stripPrefix :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m (Maybe (t m a))
1403+ :: (Eq a, IsStream t, Monad m)
1404+ => t m a -> t m a -> m (Maybe (t m a))
861−stripPrefix m1 m2 = fmap (fmap fromStreamD) $
1405+stripPrefix m1 m2 = fmap fromStreamD <$>
862 D.stripPrefix (toStreamD m1) (toStreamD m2)
1406 D.stripPrefix (toStreamD m1) (toStreamD m2)
864------------------------------------------------------------------------------
1408------------------------------------------------------------------------------
@@ -878,14 +1422,24 @@ mapM_ f m = S.mapM_ f $ toStreamS m
878-- Conversions
1422-- Conversions
879------------------------------------------------------------------------------
1423------------------------------------------------------------------------------
1427+-- toList = S.foldr (:) []
881−-- | Convert a stream into a list in the underlying monad.
1430+-- Convert a stream into a list in the underlying monad. Same as:
883-- @since 0.1.0
1432-- @since 0.1.0
884{-# INLINE toList #-}
1433{-# INLINE toList #-}
885toList :: Monad m => SerialT m a -> m [a]
1434toList :: Monad m => SerialT m a -> m [a]
886toList = P.toList
1435toList = P.toList
1439+-- toHandle h = S.mapM_ $ hPutStrLn h
888−-- | Write a stream of Strings to an IO Handle.
1442+-- Write a stream of Strings to an IO Handle.
890-- @since 0.1.0
1444-- @since 0.1.0
891toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
1445toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
@@ -917,56 +1471,116 @@ scanx = K.scanx
917scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
1471scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
918scan = scanx
1472scan = scanx
1474+-- XXX this needs to be concurrent
920−-- | Like 'scanl'' but with a monadic step function.
1475+-- | Like 'scanl'' but with a monadic fold function.
922-- @since 0.4.0
1477-- @since 0.4.0
923{-# INLINE scanlM' #-}
1478{-# INLINE scanlM' #-}
924scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1479scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
925scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
1480scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
927−-- | Strict left scan. Like 'foldl'', but returns the folded value at each
1482+-- | Strict left scan.
928−-- step, generating a stream of all intermediate fold results. The first
1483+--
929−-- element of the stream is the user supplied initial value, and the last
1484+-- @
930−-- element of the stream is the same as the result of 'foldl''.
1485+-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
1490+-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
1491+-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
1494+-- The output of 'scanl'' is the initial value of the accumulator followed by
1495+-- all the intermediate steps and the final result of 'foldl''.
1497+-- By streaming the accumulated state after each fold step, we can share the
1498+-- state across multiple stages of stream composition. Each stage can modify or
1499+-- extend the state, do some processing with it and emit it for the next stage,
1500+-- thus modularizing the stream processing. This can be useful in
1501+-- stateful or event-driven programming.
1503+-- Consider the following example, computing the sum and the product of the
1504+-- elements in a stream in one go using a @foldl'@:
1507+-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
1511+-- Using @scanl'@ we can compute the sum in the first stage and pass it down to
1512+-- the next stage for computing the product:
1515+-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
1516+-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
1517+-- $ S.fromList \[1,2,3,4]
1521+-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
1522+-- lazy expressions inside the accumulator, and to help GHC optimize better, it
1523+-- is recommended that a strict data structure is used for accumulator.
932-- @since 0.2.0
1525-- @since 0.2.0
933{-# INLINE scanl' #-}
1526{-# INLINE scanl' #-}
934scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1527scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
935scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
1528scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
937−-- | Like 'scanlM'' but with a non-strict accumulator.
1530+-- XXX enable once the signature (monadic zero) change is settled
1531+-- | Like scanl' but does not stream the initial value of the accumulator.
1533+-- > postscanl' f z xs = S.drop 1 $ scanl' f z xs
1536+{-# INLINE _postscanl' #-}
1537+_postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1538+_postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
1540+-- XXX this needs to be concurrent
1541+-- | Like postscanl' but with a monadic step function.
939−{-# INLINE scanlM #-}
1544+{-# INLINE _postscanlM' #-}
940−scanlM :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1545+_postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
941−scanlM step begin m = fromStreamD $ D.scanlM step begin $ toStreamD m
1546+_postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
1548+-- XXX prescanl does not sound very useful, enable only if there is a
1549+-- compelling use case.
943−-- | Like 'scanl'' but with a non-strict accumulator.
1551+-- | Like scanl' but does not stream the final value of the accumulator.
944−{-# INLINE scanl #-}
1554+{-# INLINE _prescanl' #-}
945−scanl :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1555+_prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
946−scanl step z m = fromStreamD $ D.scanl step z $ toStreamD m
1556+_prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
1558+-- XXX this needs to be concurrent
948−-- | Left scan with a monadic step function. Uses the first element of
1559+-- | Like postscanl' but with a monadic step function.
949−-- the stream as the starting value. Returns an empty stream if the
950−-- given stream is empty.
952−{-# INLINE scanl1M #-}
1562+{-# INLINE _prescanlM' #-}
953−scanl1M :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
1563+_prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
954−scanl1M step m = fromStreamD $ D.scanl1M step $ toStreamD m
1564+_prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
956−-- | Like 'scanl1M' but with a strict accumulator.
1566+-- XXX this needs to be concurrent
1567+-- | Like 'scanl1'' but with a monadic step function.
958{-# INLINE scanl1M' #-}
1570{-# INLINE scanl1M' #-}
959scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
1571scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
960scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
1572scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
962−-- | Left scan, uses the first element of the stream as the starting
1574+-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
963−-- value. Returns an empty stream if the given stream is empty.
1575+-- is used as the initial value of the accumulator. Does nothing if the stream
964−{-# INLINE scanl1 #-}
1576+-- is empty.
965−scanl1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
966−scanl1 step m = fromStreamD $ D.scanl1 step $ toStreamD m
968−-- | Like 'scanl1'' but with a strict accumulator.
1579+-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
970{-# INLINE scanl1' #-}
1584{-# INLINE scanl1' #-}
971scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
1585scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
972scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
1586scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
@@ -997,10 +1611,19 @@ filterM p m = fromStreamD $ D.filterM p $ toStreamD m
998-- | Drop repeated elements that are adjacent to each other.
1612-- | Drop repeated elements that are adjacent to each other.
1000{-# INLINE uniq #-}
1615{-# INLINE uniq #-}
1001uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
1616uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
1002uniq = fromStreamD . D.uniq . toStreamD
1617uniq = fromStreamD . D.uniq . toStreamD
1619+-- | Ensures that all the elements of the stream are identical and then returns
1620+-- that unique element.
1624+the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
1625+the m = S.the (toStreamS m)
1004-- | Take first 'n' elements from the stream and discard the rest.
1627-- | Take first 'n' elements from the stream and discard the rest.
1006-- @since 0.1.0
1629-- @since 0.1.0
@@ -1049,10 +1672,18 @@ dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
1049-- Transformation by Mapping
1672-- Transformation by Mapping
1050------------------------------------------------------------------------------
1673------------------------------------------------------------------------------
1677+-- mapM f = sequence . map f
1052−-- | Replace each element of the stream with the result of a monadic action
1680+-- Apply a monadic function to each element of the stream and replace it with
1053−-- applied on the element.
1681+-- the output of the resulting action.
1684+-- > runStream $ S.mapM putStr $ S.fromList ["a", "b", "c"]
1056-- runStream $ S.replicateM 10 (return 1)
1687-- runStream $ S.replicateM 10 (return 1)
1057-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
1688-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
@@ -1072,10 +1703,18 @@ mapM = K.mapM
1072mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
1703mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
1073mapMSerial = Serial.mapM
1704mapMSerial = Serial.mapM
1708+-- sequence = mapM id
1075−-- | Reduce a stream of monadic actions to a stream of the output of those
1711+-- Replace the elements of a stream of monadic actions with the outputs of
1076−-- actions.
1712+-- those actions.
1715+-- > runStream $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
1079-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
1718-- runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
1080-- & (serially . S.sequence)
1719-- & (serially . S.sequence)
@@ -1142,17 +1781,34 @@ reverse m = fromStream $ go K.nil (toStream m)
1142-- Transformation by Inserting
1781-- Transformation by Inserting
1143------------------------------------------------------------------------------
1782------------------------------------------------------------------------------
1145−-- | Generate a stream by performing the monadic action inbetween all elements
1784+-- | Generate a stream by performing a monadic action between consecutive
1146−-- of the given stream.
1785+-- elements of the given stream.
1787+-- /Concurrent (do not use with 'parallely' on infinite streams)/
1790+-- > S.toList $ S.intersperseM (putChar \'a' >> return ',') $ S.fromList "hello"
1148-- @since 0.5.0
1794-- @since 0.5.0
1149{-# INLINE intersperseM #-}
1795{-# INLINE intersperseM #-}
1150intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1796intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1151intersperseM = K.intersperseM
1797intersperseM = K.intersperseM
1153−-- | Inserts @x@ before the first element in the Stream that is
1799+-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
1154−-- less than @x@, based on the user supplied comparison function.
1800+-- @stream@ that is less than @elem@ when compared using @cmp@.
1803+-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
1807+-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
1156{-# INLINE insertBy #-}
1812{-# INLINE insertBy #-}
1157insertBy ::
1813insertBy ::
1158 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
1814 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
@@ -1162,9 +1818,15 @@ insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
1162-- Deleting
1818-- Deleting
1163------------------------------------------------------------------------------
1819------------------------------------------------------------------------------
1165−-- | Deletes the first occurence of @x@ in the Stream that satisfies the
1821+-- | Deletes the first occurence of the element in the stream that satisfies
1166−-- equality predicate.
1822+-- the given equality predicate.
1825+-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
1168{-# INLINE deleteBy #-}
1830{-# INLINE deleteBy #-}
1169deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
1831deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
1170deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
1832deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
@@ -1173,20 +1835,38 @@ deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
1173-- Zipping
1835-- Zipping
1174------------------------------------------------------------------------------
1836------------------------------------------------------------------------------
1839+-- > indexed = S.zipWith (,) (S.intFrom 0)
1176−-- | Pair each element in a stream with its index.
1841+-- Pair each element in a stream with its index.
1844+-- > S.toList $ S.indexed $ S.fromList "hello"
1845+-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1178{-# INLINE indexed #-}
1849{-# INLINE indexed #-}
1179indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1850indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1180indexed = fromStreamD . D.indexed . toStreamD
1851indexed = fromStreamD . D.indexed . toStreamD
1854+-- > indexedR n = S.zipWith (,) (S.intFromThen n (n - 1))
1182−-- | Pair each element in a stream with its index, starting from the
1856+-- Pair each element in a stream with its index, starting from the
1183-- given index @n@ and counting down.
1857-- given index @n@ and counting down.
1860+-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
1861+-- [(9,'h'),(8,'e'),(7,'l'),(6,'l'),(5,'o')]
1185{-# INLINE indexedR #-}
1865{-# INLINE indexedR #-}
1186indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1866indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1187indexedR n = fromStreamD . D.indexedR n . toStreamD
1867indexedR n = fromStreamD . D.indexedR n . toStreamD
1189−-- | Zip two streams serially using a monadic zipping function.
1869+-- | Like 'zipWith' but using a monadic zipping function.
1191-- @since 0.4.0
1871-- @since 0.4.0
1192{-# INLINABLE zipWithM #-}
1872{-# INLINABLE zipWithM #-}
@@ -1195,6 +1875,11 @@ zipWithM f m1 m2 = fromStreamS $ S.zipWithM f (toStreamS m1) (toStreamS m2)
1196-- | Zip two streams serially using a pure zipping function.
1876-- | Zip two streams serially using a pure zipping function.
1879+-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
1198-- @since 0.1.0
1883-- @since 0.1.0
1199{-# INLINABLE zipWith #-}
1884{-# INLINABLE zipWith #-}
1200zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
1885zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
@@ -1204,36 +1889,115 @@ zipWith f m1 m2 = fromStreamS $ S.zipWith f (toStreamS m1) (toStreamS m2)
1204-- Comparison
1889-- Comparison
1205------------------------------------------------------------------------------
1890------------------------------------------------------------------------------
1207−-- | Compare two streams for equality using a equality function.
1892+-- | Compare two streams for equality using an equality function.
1209−-- @since 0.5.3
1894+-- @since 0.6.0
1210{-# INLINABLE eqBy #-}
1895{-# INLINABLE eqBy #-}
1211−eqBy :: Monad m => (a -> b -> Bool) -> SerialT m a -> SerialT m b -> m Bool
1896+eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
1212eqBy = P.eqBy
1897eqBy = P.eqBy
1214−-- | Compare two streams using a comparison function.
1899+-- | Compare two streams lexicographically using a comparison function.
1216−-- @since 0.5.3
1901+-- @since 0.6.0
1217{-# INLINABLE cmpBy #-}
1902{-# INLINABLE cmpBy #-}
1219− :: Monad m
1904+ :: (IsStream t, Monad m)
1220− => (a -> b -> Ordering) -> SerialT m a -> SerialT m b -> m Ordering
1905+ => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
1221cmpBy = P.cmpBy
1906cmpBy = P.cmpBy
1223−-- | Non-overloaded version of merge.
1908+------------------------------------------------------------------------------
1910+------------------------------------------------------------------------------
1912+-- | Merge two streams using a comparison function. The head elements of both
1913+-- the streams are compared and the smaller of the two elements is emitted, if
1914+-- both elements are equal then the element from the first stream is used
1917+-- If the streams are sorted in ascending order, the resulting stream would
1918+-- also remain sorted in ascending order.
1921+-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
1224{-# INLINABLE mergeBy #-}
1926{-# INLINABLE mergeBy #-}
1225mergeBy ::
1927mergeBy ::
1226 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
1928 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
1227mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
1929mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
1229−-- | Merge two sorted streams.
1931+-- | Like 'mergeBy' but with a monadic comparison function.
1933+-- Merge two streams randomly:
1936+-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
1937+-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
1938+-- [2,1,2,2,2,1,1,1]
1941+-- Merge two streams in a proportion of 2:1:
1944+-- proportionately m n = do
1945+-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
1946+-- return $ \\_ _ -> do
1947+-- r <- readIORef ref
1948+-- writeIORef ref $ tail r
1952+-- f <- proportionately 2 1
1953+-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
1957+-- [1,1,2,1,1,2,1,1,2]
1961+{-# INLINABLE mergeByM #-}
1963+ :: (IsStream t, Monad m)
1964+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1965+mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
1967+-- Holding this back for now, we may want to use the name "merge" differently
1969+-- | Same as @'mergeBy' 'compare'@.
1972+-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
1230{-# INLINABLE merge #-}
1977{-# INLINABLE merge #-}
1232 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
1979 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
1233merge = mergeBy compare
1980merge = mergeBy compare
1235−-- | Ensures all the elements are identical and then returns
1982+
1236−-- that unique element.
1983+-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
1237−{-# INLINE the #-}
1984+-- merged are generated concurrently).
1238−the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
1985+--
1239−the m = S.the (toStreamS m)
1986+-- @since 0.6.0
1987+mergeAsyncBy :: (IsStream t, MonadAsync m)
1988+ => (a -> a -> Ordering) -> t m a -> t m a -> t m a
1989+mergeAsyncBy f m1 m2 = K.fromStream $ K.Stream $ \st stp sng yld -> do
1990+ ma <- mkAsync' (rstState st) m1
1991+ mb <- mkAsync' (rstState st) m2
1992+ K.unStream (K.toStream (K.mergeBy f ma mb)) (rstState st) stp sng yld
1994+-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
1995+-- merged are generated concurrently).
1998+mergeAsyncByM :: (IsStream t, MonadAsync m)
1999+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2000+mergeAsyncByM f m1 m2 = K.fromStream $ K.Stream $ \st stp sng yld -> do
2001+ ma <- mkAsync' (rstState st) m1
2002+ mb <- mkAsync' (rstState st) m2
2003+ K.unStream (K.toStream (K.mergeByM f ma mb)) (rstState st) stp sng yld