@@ -1,4 +1,4 @@
1−#include "inline.hs"
1+{-# LANGUAGE UndecidableInstances #-}
4-- Module : Streamly.Internal.Data.Stream.StreamD.Type
4-- Module : Streamly.Internal.Data.Stream.StreamD.Type
@@ -17,127 +17,172 @@ module Streamly.Internal.Data.Stream.StreamD.Type
18 -- * The stream type
18 -- * The stream type
20− -- XXX UnStream is exported to avoid a performance issue in concatMap if we
20+ -- XXX UnStream is exported to avoid a performance issue in some
21− -- use the pattern synonym "Stream".
21+ -- combinators if we use the pattern synonym "Stream".
22 , Stream (Stream, UnStream)
22 , Stream (Stream, UnStream)
24− -- * Primitives
24+ -- * CrossStream type wrapper
25− , nilM
25+ , CrossStream
27− , uncons
27+ , mkCross
29+ -- * Conversion to StreamK
29 -- * From Unfold
33 -- * From Unfold
32− -- * From Values
41+ -- ** From Values
33 , fromPure
42 , fromPure
34 , fromEffect
43 , fromEffect
36− -- * From Containers
45+ -- ** From Containers
37− , fromList
46+ , Streamly.Internal.Data.Stream.StreamD.Type.fromList
39− -- * Conversions From/To
48+ -- * Elimination
40− , fromStreamK
49+ -- ** Primitives
41− , toStreamK
50+ , uncons
43− -- * Running a 'Fold'
52+ -- ** Strict Left Folds
44− , fold
53+ , Streamly.Internal.Data.Stream.StreamD.Type.fold
45 , foldBreak
54 , foldBreak
46− , foldContinue
55+ , foldAddLazy
47 , foldEither
57 , foldEither
49− -- * Right Folds
59+ , Streamly.Internal.Data.Stream.StreamD.Type.foldl'
59 , foldlMx'
62 , foldlMx'
61− -- * Special Folds
64+ -- ** Lazy Right Folds
64− -- * To Containers
67+ , Streamly.Internal.Data.Stream.StreamD.Type.foldr
67− -- * Multi-stream folds
70+ -- ** Specific Folds
69− , cmpBy
72+ , Streamly.Internal.Data.Stream.StreamD.Type.toList
71− -- * Transformations
74+ -- * Mapping
78+ -- * Stateful Filters
75 , takeWhile
80 , takeWhile
76 , takeWhileM
81 , takeWhileM
77 , takeEndBy
82 , takeEndBy
78 , takeEndByM
83 , takeEndByM
80− -- * Nesting
85+ -- * Combining Two Streams
81 , crossApply
91 , crossApply
82 , crossApplyFst
92 , crossApplyFst
83 , crossApplySnd
93 , crossApplySnd
85 , ConcatMapUState (..)
98 , ConcatMapUState (..)
86 , unfoldMany
99 , unfoldMany
88 , concatMap
103 , concatMap
89 , concatMapM
104 , concatMapM
90− , concatEffect
105+ , concat
92− -- * Expanding trees top down
107+ -- * Unfold Iterate
93 , unfoldIterateDfs
108 , unfoldIterateDfs
94 , unfoldIterateBfs
109 , unfoldIterateBfs
95 , unfoldIterateBfsRev
110 , unfoldIterateBfsRev
97 , concatIterateScan
113 , concatIterateScan
98 , concatIterateDfs
114 , concatIterateDfs
99 , concatIterateBfs
115 , concatIterateBfs
100 , concatIterateBfsRev
116 , concatIterateBfsRev
102 , FoldMany (..) -- for inspection testing
119 , FoldMany (..) -- for inspection testing
103 , FoldManyPost (..)
120 , FoldManyPost (..)
104 , foldMany
121 , foldMany
105 , foldManyPost
122 , foldManyPost
106 , chunksOf
123 , chunksOf
108 , refoldMany
124 , refoldMany
110− -- * Folding trees bottom up
126+ -- * Fold Iterate
111 , reduceIterateBfs
127 , reduceIterateBfs
112 , foldIterateBfs
128 , foldIterateBfs
130+ -- * Multi-stream folds
116import Control.Applicative (liftA2)
138import Control.Applicative (liftA2)
139+import Control.Monad.Catch (MonadThrow, throwM)
140+import Control.Monad.Trans.Class (MonadTrans(lift))
141+import Control.Monad.IO.Class (MonadIO(..))
142+import Data.Foldable (Foldable(foldl'), fold, foldr)
117import Data.Functor (($>))
143import Data.Functor (($>))
118import Data.Functor.Identity (Identity(..))
144import Data.Functor.Identity (Identity(..))
145+import Data.Maybe (fromMaybe)
146+import Data.Semigroup (Endo(..))
119import Fusion.Plugin.Types (Fuse(..))
147import Fusion.Plugin.Types (Fuse(..))
120import GHC.Base (build)
148import GHC.Base (build)
149+import GHC.Exts (IsList(..), IsString(..), oneShot)
121import GHC.Types (SPEC(..))
150import GHC.Types (SPEC(..))
122−import Prelude hiding (map, mapM, foldr, take, concatMap, takeWhile)
151+import Prelude hiding (map, mapM, take, concatMap, takeWhile, zipWith, concat)
153+ ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
154+ , readListPrecDefault)
156+import Streamly.Internal.BaseCompat ((#.))
124import Streamly.Internal.Data.Fold.Type (Fold(..))
157import Streamly.Internal.Data.Fold.Type (Fold(..))
158+import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
125import Streamly.Internal.Data.Refold.Type (Refold(..))
159import Streamly.Internal.Data.Refold.Type (Refold(..))
126import Streamly.Internal.Data.Stream.StreamD.Step (Step (..))
160import Streamly.Internal.Data.Stream.StreamD.Step (Step (..))
127import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)
161import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)
128import Streamly.Internal.Data.Unfold.Type (Unfold(..))
162import Streamly.Internal.Data.Unfold.Type (Unfold(..))
130−import qualified Streamly.Internal.Data.Fold.Type as FL
164+import qualified Streamly.Internal.Data.Fold.Type as FL hiding (foldr)
131import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
165import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
132#ifdef USE_UNFOLDS_EVERYWHERE
166#ifdef USE_UNFOLDS_EVERYWHERE
133import qualified Streamly.Internal.Data.Unfold.Type as Unfold
167import qualified Streamly.Internal.Data.Unfold.Type as Unfold
171+-- >>> import Streamly.Internal.Data.Stream (CrossStream(..))
172+-- >>> import qualified Streamly.Data.Fold as Fold
173+-- >>> import qualified Streamly.Data.Parser as Parser
174+-- >>> import qualified Streamly.Data.Stream as Stream
175+-- >>> import qualified Streamly.Data.Unfold as Unfold
176+-- >>> import qualified Streamly.Internal.Data.Stream as Stream
177+-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir
178+-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
136------------------------------------------------------------------------------
180------------------------------------------------------------------------------
137-- The direct style stream type
181-- The direct style stream type
138------------------------------------------------------------------------------
182------------------------------------------------------------------------------
140-- gst = global state
184-- gst = global state
141-- | A stream consists of a step function that generates the next step given a
186-- | A stream consists of a step function that generates the next step given a
142-- current state, and the current state.
187-- current state, and the current state.
143data Stream m a =
188data Stream m a =
@@ -160,11 +205,19 @@ pattern Stream step state <- (unShare -> UnStream step state)
160-- Primitives
205-- Primitives
161------------------------------------------------------------------------------
206------------------------------------------------------------------------------
163−-- | An empty 'Stream' with a side effect.
208+-- | A stream that terminates without producing any output, but produces a side
211+-- >>> Stream.fold Fold.toList (Stream.nilM (print "nil"))
164{-# INLINE_NORMAL nilM #-}
216{-# INLINE_NORMAL nilM #-}
165nilM :: Applicative m => m b -> Stream m a
217nilM :: Applicative m => m b -> Stream m a
166nilM m = Stream (\_ _ -> m $> Stop) ()
218nilM m = Stream (\_ _ -> m $> Stop) ()
220+-- | Like 'cons' but fuses an effect instead of a pure value.
168{-# INLINE_NORMAL consM #-}
221{-# INLINE_NORMAL consM #-}
169consM :: Applicative m => m a -> Stream m a -> Stream m a
222consM :: Applicative m => m a -> Stream m a -> Stream m a
170consM m (Stream step state) = Stream step1 Nothing
223consM m (Stream step state) = Stream step1 Nothing
@@ -179,7 +232,35 @@ consM m (Stream step state) = Stream step1 Nothing
179 Skip s -> Skip (Just s)
232 Skip s -> Skip (Just s)
180 Stop -> Stop) <$> step gst st
233 Stop -> Stop) <$> step gst st
182−-- | Does not fuse, has the same performance as the StreamK version.
235+-- | Decompose a stream into its head and tail. If the stream is empty, returns
236+-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
237+-- the head of the stream and @ma@ its tail.
241+-- >>> Nothing <- Stream.uncons Stream.nil
242+-- >>> Just ("a", t) <- Stream.uncons (Stream.cons "a" Stream.nil)
244+-- This can be used to consume the stream in an imperative manner one element
245+-- at a time, as it just breaks down the stream into individual elements and we
246+-- can loop over them as we deem fit. For example, this can be used to convert
247+-- a streamly stream into other stream types.
249+-- All the folds in this module can be expressed in terms of 'uncons', however,
250+-- this is generally less efficient than specific folds because it takes apart
251+-- the stream one element at a time, therefore, does not take adavantage of
254+-- 'foldBreak' is a more general way of consuming a stream piecemeal.
258+-- r <- Stream.foldBreak Fold.one xs
259+-- return $ case r of
260+-- (Nothing, _) -> Nothing
261+-- (Just h, t) -> Just (h, t)
183{-# INLINE_NORMAL uncons #-}
264{-# INLINE_NORMAL uncons #-}
184uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
265uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
185uncons (UnStream step state) = go SPEC state
266uncons (UnStream step state) = go SPEC state
@@ -197,7 +278,13 @@ uncons (UnStream step state) = go SPEC state
198data UnfoldState s = UnfoldNothing | UnfoldJust s
279data UnfoldState s = UnfoldNothing | UnfoldJust s
200−-- | Convert an 'Unfold' into a 'Stream' by supplying it a seed.
281+-- | Convert an 'Unfold' into a stream by supplying it an input seed.
283+-- >>> s = Stream.unfold Unfold.replicateM (3, putStrLn "hello")
284+-- >>> Stream.fold Fold.drain s
202{-# INLINE_NORMAL unfold #-}
289{-# INLINE_NORMAL unfold #-}
203unfold :: Applicative m => Unfold m a b -> a -> Stream m b
290unfold :: Applicative m => Unfold m a b -> a -> Stream m b
@@ -217,7 +304,12 @@ unfold (Unfold ustep inject) seed = Stream step UnfoldNothing
217-- From Values
304-- From Values
218------------------------------------------------------------------------------
305------------------------------------------------------------------------------
220−-- | Create a singleton 'Stream' from a pure value.
307+-- | Create a singleton stream from a pure value.
309+-- >>> fromPure a = a `Stream.cons` Stream.nil
310+-- >>> fromPure = pure
311+-- >>> fromPure = fromEffect . pure
221{-# INLINE_NORMAL fromPure #-}
313{-# INLINE_NORMAL fromPure #-}
222fromPure :: Applicative m => a -> Stream m a
314fromPure :: Applicative m => a -> Stream m a
223fromPure x = Stream (\_ s -> pure $ step undefined s) True
315fromPure x = Stream (\_ s -> pure $ step undefined s) True
@@ -226,7 +318,14 @@ fromPure x = Stream (\_ s -> pure $ step undefined s) True
226 step _ True = Yield x False
318 step _ True = Yield x False
227 step _ False = Stop
319 step _ False = Stop
229−-- | Create a singleton 'Stream' from a monadic action.
321+-- | Create a singleton stream from a monadic action.
323+-- >>> fromEffect m = m `consM` Stream.nil
324+-- >>> fromEffect = Stream.sequence . Stream.fromPure
326+-- >>> Stream.fold Fold.drain $ Stream.fromEffect (putStrLn "hello")
230{-# INLINE_NORMAL fromEffect #-}
329{-# INLINE_NORMAL fromEffect #-}
231fromEffect :: Applicative m => m a -> Stream m a
330fromEffect :: Applicative m => m a -> Stream m a
232fromEffect m = Stream step True
331fromEffect m = Stream step True
@@ -242,7 +341,8 @@ fromEffect m = Stream step True
242------------------------------------------------------------------------------
341------------------------------------------------------------------------------
244-- Adapted from the vector package.
343-- Adapted from the vector package.
245−-- | Convert a list of pure values to a 'Stream'
345+-- | Construct a stream from a list of pure values.
246{-# INLINE_LATE fromList #-}
346{-# INLINE_LATE fromList #-}
247fromList :: Applicative m => [a] -> Stream m a
347fromList :: Applicative m => [a] -> Stream m a
248#ifdef USE_UNFOLDS_EVERYWHERE
348#ifdef USE_UNFOLDS_EVERYWHERE
@@ -295,12 +395,38 @@ toStreamK (Stream step state) = go state
295-- Running a 'Fold'
395-- Running a 'Fold'
296------------------------------------------------------------------------------
396------------------------------------------------------------------------------
398+-- >>> fold f = Fold.extractM . Stream.foldAddLazy f
399+-- >>> fold f = Stream.fold Fold.one . Stream.foldManyPost f
400+-- >>> fold f = Fold.extractM <=< Stream.foldAdd f
402+-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
403+-- expression strictly at each step. The behavior is similar to 'foldl''. A
404+-- 'Fold' can terminate early without consuming the full stream. See the
405+-- documentation of individual 'Fold's for termination behavior.
409+-- >>> fold f = fmap fst . Stream.foldBreak f
410+-- >>> fold f = Stream.parse (Parser.fromFold f)
414+-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
298{-# INLINE_NORMAL fold #-}
417{-# INLINE_NORMAL fold #-}
299fold :: Monad m => Fold m a b -> Stream m a -> m b
418fold :: Monad m => Fold m a b -> Stream m a -> m b
300fold fld strm = do
419fold fld strm = do
301 (b, _) <- foldBreak fld strm
420 (b, _) <- foldBreak fld strm
423+-- | Fold resulting in either breaking the stream or continuation of the fold.
424+-- Instead of supplying the input stream in one go we can run the fold multiple
425+-- times, each time supplying the next segment of the input stream. If the fold
426+-- has not yet finished it returns a fold that can be run again otherwise it
427+-- returns the fold result and the residual stream.
304{-# INLINE_NORMAL foldEither #-}
430{-# INLINE_NORMAL foldEither #-}
305foldEither :: Monad m =>
431foldEither :: Monad m =>
306 Fold m a b -> Stream m a -> m (Either (Fold m a b) (b, Stream m a))
432 Fold m a b -> Stream m a -> m (Either (Fold m a b) (b, Stream m a))
@@ -324,6 +450,9 @@ foldEither (Fold fstep begin done) (UnStream step state) = do
324 Skip s -> go SPEC fs s
450 Skip s -> go SPEC fs s
325 Stop -> return $! Left (Fold fstep (return $ FL.Partial fs) done)
451 Stop -> return $! Left (Fold fstep (return $ FL.Partial fs) done)
453+-- | Like 'fold' but also returns the remaining stream. The resulting stream
454+-- would be 'Stream.nil' if the stream finished before the fold.
327{-# INLINE_NORMAL foldBreak #-}
456{-# INLINE_NORMAL foldBreak #-}
328foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
457foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
329foldBreak fld strm = do
458foldBreak fld strm = do
@@ -342,9 +471,18 @@ foldBreak fld strm = do
343 nil = Stream (\_ _ -> return Stop) ()
472 nil = Stream (\_ _ -> return Stop) ()
474+-- | Append a stream to a fold lazily to build an accumulator incrementally.
476+-- Example, to continue folding a list of streams on the same sum fold:
478+-- >>> streams = [Stream.fromList [1..5], Stream.fromList [6..10]]
479+-- >>> f = Prelude.foldl Stream.foldAddLazy Fold.sum streams
480+-- >>> Stream.fold f Stream.nil
345−{-# INLINE_NORMAL foldContinue #-}
483+{-# INLINE_NORMAL foldAddLazy #-}
346−foldContinue :: Monad m => Fold m a b -> Stream m a -> Fold m a b
484+foldAddLazy :: Monad m => Fold m a b -> Stream m a -> Fold m a b
347−foldContinue (Fold fstep finitial fextract) (Stream sstep state) =
485+foldAddLazy (Fold fstep finitial fextract) (Stream sstep state) =
348 Fold fstep initial fextract
486 Fold fstep initial fextract
@@ -367,6 +505,15 @@ foldContinue (Fold fstep finitial fextract) (Stream sstep state) =
367 Skip s -> go SPEC fs s
505 Skip s -> go SPEC fs s
368 Stop -> return $ FL.Partial fs
506 Stop -> return $ FL.Partial fs
508+-- >>> foldAdd f = Stream.foldAddLazy f >=> Fold.reduce
511+-- >>> foldAdd = flip Fold.addStream
513+foldAdd :: Monad m => Fold m a b -> Stream m a -> m (Fold m a b)
515+ Streamly.Internal.Data.Stream.StreamD.Type.fold (FL.duplicate f)
370------------------------------------------------------------------------------
517------------------------------------------------------------------------------
371-- Right Folds
518-- Right Folds
372------------------------------------------------------------------------------
519------------------------------------------------------------------------------
@@ -391,6 +538,23 @@ foldContinue (Fold fstep finitial fextract) (Stream sstep state) =
392-- S.foldrM (\x t -> if x then return t else return False) (return True)
539-- S.foldrM (\x t -> if x then return t else return False) (return True)
393-- (S.fromList [False,undefined] :: Stream IO Bool)
540-- (S.fromList [False,undefined] :: Stream IO Bool)
542+-- | Right associative/lazy pull fold. @foldrM build final stream@ constructs
543+-- an output structure using the step function @build@. @build@ is invoked with
544+-- the next input element and the remaining (lazy) tail of the output
545+-- structure. It builds a lazy output expression using the two. When the "tail
546+-- structure" in the output expression is evaluated it calls @build@ again thus
547+-- lazily consuming the input @stream@ until either the output expression built
548+-- by @build@ is free of the "tail" or the input is exhausted in which case
549+-- @final@ is used as the terminating case for the output structure. For more
550+-- details see the description in the previous section.
552+-- Example, determine if any element is 'odd' in a stream:
554+-- >>> s = Stream.fromList (2:4:5:undefined)
555+-- >>> step x xs = if odd x then return True else xs
556+-- >>> Stream.foldrM step (return False) s
395{-# INLINE_NORMAL foldrM #-}
559{-# INLINE_NORMAL foldrM #-}
396foldrM :: Monad m => (a -> m b -> m b) -> m b -> Stream m a -> m b
560foldrM :: Monad m => (a -> m b -> m b) -> m b -> Stream m a -> m b
@@ -425,6 +589,18 @@ foldrMx fstep final convert (Stream step state) = convert $ go SPEC state
425-- monads. For example, if "any" is implemented using "foldr" instead of
589-- monads. For example, if "any" is implemented using "foldr" instead of
426-- "foldrM" it performs the same with Identity monad but performs 1000x slower
590-- "foldrM" it performs the same with Identity monad but performs 1000x slower
427-- with IO monad.
591-- with IO monad.
593+-- | Right fold, lazy for lazy monads and pure streams, and strict for strict
596+-- Please avoid using this routine in strict monads like IO unless you need a
597+-- strict right fold. This is provided only for use in lazy monads (e.g.
598+-- Identity) or pure streams. Note that with this signature it is not possible
599+-- to implement a lazy foldr when the monad @m@ is strict. In that case it
600+-- would be strict in its accumulator and therefore would necessarily consume
603+-- >>> foldr f z = Stream.foldrM (\a b -> f a <$> b) (return z)
429{-# INLINE_NORMAL foldr #-}
605{-# INLINE_NORMAL foldr #-}
430foldr :: Monad m => (a -> b -> b) -> b -> Stream m a -> m b
606foldr :: Monad m => (a -> b -> b) -> b -> Stream m a -> m b
@@ -520,7 +696,7 @@ drain (Stream step state) = go SPEC state
521{-# INLINE_NORMAL toList #-}
697{-# INLINE_NORMAL toList #-}
522toList :: Monad m => Stream m a -> m [a]
698toList :: Monad m => Stream m a -> m [a]
523−toList = foldr (:) []
699+toList = Streamly.Internal.Data.Stream.StreamD.Type.foldr (:) []
525-- Use foldr/build fusion to fuse with list consumers
701-- Use foldr/build fusion to fuse with list consumers
526-- This can be useful when using the IsList instance
702-- This can be useful when using the IsList instance
@@ -533,7 +709,7 @@ toListFB c n (Stream step state) = go state
533 Skip s -> go s
709 Skip s -> go s
534 Stop -> n
710 Stop -> n
536−{-# RULES "toList Identity" toList = toListId #-}
712+{-# RULES "toList Identity" Streamly.Internal.Data.Stream.StreamD.Type.toList = toListId #-}
537{-# INLINE_EARLY toListId #-}
713{-# INLINE_EARLY toListId #-}
538toListId :: Stream Identity a -> Identity [a]
714toListId :: Stream Identity a -> Identity [a]
539toListId s = Identity $ build (\c n -> toListFB c n s)
715toListId s = Identity $ build (\c n -> toListFB c n s)
@@ -543,6 +719,8 @@ toListId s = Identity $ build (\c n -> toListFB c n s)
543------------------------------------------------------------------------------
719------------------------------------------------------------------------------
545-- Adapted from the vector package.
721-- Adapted from the vector package.
723+-- | Compare two streams for equality
546{-# INLINE_NORMAL eqBy #-}
724{-# INLINE_NORMAL eqBy #-}
547eqBy :: Monad m => (a -> b -> Bool) -> Stream m a -> Stream m b -> m Bool
725eqBy :: Monad m => (a -> b -> Bool) -> Stream m a -> Stream m b -> m Bool
548eqBy eq (Stream step1 t1) (Stream step2 t2) = eq_loop0 SPEC t1 t2
726eqBy eq (Stream step1 t1) (Stream step2 t2) = eq_loop0 SPEC t1 t2
@@ -571,7 +749,8 @@ eqBy eq (Stream step1 t1) (Stream step2 t2) = eq_loop0 SPEC t1 t2
571 Stop -> return True
749 Stop -> return True
573-- Adapted from the vector package.
751-- Adapted from the vector package.
574−-- | Compare two streams lexicographically
753+-- | Compare two streams lexicographically.
575{-# INLINE_NORMAL cmpBy #-}
754{-# INLINE_NORMAL cmpBy #-}
577 :: Monad m
756 :: Monad m
@@ -606,7 +785,17 @@ cmpBy cmp (Stream step1 t1) (Stream step2 t2) = cmp_loop0 SPEC t1 t2
606------------------------------------------------------------------------------
785------------------------------------------------------------------------------
608-- Adapted from the vector package.
787-- Adapted from the vector package.
609−-- | Map a monadic function over a 'Stream'
788+
790+-- >>> mapM f = Stream.sequence . fmap f
792+-- Apply a monadic function to each element of the stream and replace it with
793+-- the output of the resulting action.
795+-- >>> s = Stream.fromList ["a", "b", "c"]
796+-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
610{-# INLINE_NORMAL mapM #-}
799{-# INLINE_NORMAL mapM #-}
611mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
800mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
612mapM f (Stream step state) = Stream step' state
801mapM f (Stream step state) = Stream step' state
@@ -633,11 +822,149 @@ instance Functor m => Functor (Stream m) where
633 {-# INLINE (<$) #-}
822 {-# INLINE (<$) #-}
634 (<$) = fmap . const
823 (<$) = fmap . const
825+------------------------------------------------------------------------------
827+------------------------------------------------------------------------------
829+-- XXX Show instance is 10x slower compared to read, we can do much better.
830+-- The list show instance itself is really slow.
832+-- XXX The default definitions of "<" in the Ord instance etc. do not perform
833+-- well, because they do not get inlined. Need to add INLINE in Ord class in
836+instance IsList (Stream Identity a) where
837+ type (Item (Stream Identity a)) = a
839+ {-# INLINE fromList #-}
840+ fromList = Streamly.Internal.Data.Stream.StreamD.Type.fromList
842+ {-# INLINE toList #-}
843+ toList = runIdentity . Streamly.Internal.Data.Stream.StreamD.Type.toList
845+instance Eq a => Eq (Stream Identity a) where
847+ (==) xs ys = runIdentity $ eqBy (==) xs ys
849+instance Ord a => Ord (Stream Identity a) where
850+ {-# INLINE compare #-}
851+ compare xs ys = runIdentity $ cmpBy compare xs ys
878+ max x y = if x <= y then y else x
881+ min x y = if x <= y then x else y
883+instance Show a => Show (Stream Identity a) where
884+ showsPrec p dl = showParen (p > 10) $
885+ showString "fromList " . shows (GHC.Exts.toList dl)
887+instance Read a => Read (Stream Identity a) where
888+ readPrec = parens $ prec 10 $ do
889+ Ident "fromList" <- lexP
890+ Streamly.Internal.Data.Stream.StreamD.Type.fromList <$> readPrec
892+ readListPrec = readListPrecDefault
894+instance (a ~ Char) => IsString (Stream Identity a) where
895+ {-# INLINE fromString #-}
896+ fromString = Streamly.Internal.Data.Stream.StreamD.Type.fromList
898+-------------------------------------------------------------------------------
900+-------------------------------------------------------------------------------
902+-- The default Foldable instance has several issues:
903+-- 1) several definitions do not have INLINE on them, so we provide
904+-- re-implementations with INLINE pragmas.
905+-- 2) the definitions of sum/product/maximum/minimum are inefficient as they
906+-- use right folds, they cannot run in constant memory. We provide
907+-- implementations using strict left folds here.
909+-- There is no Traversable instance because, there is no scalable cons for
910+-- StreamD, use toList and fromList instead.
912+instance (Foldable m, Monad m) => Foldable (Stream m) where
914+ {-# INLINE foldMap #-}
917+ . Streamly.Internal.Data.Stream.StreamD.Type.foldr (mappend . f) mempty
919+ {-# INLINE foldr #-}
920+ foldr f z t = appEndo (foldMap (Endo #. f) t) z
922+ {-# INLINE foldl' #-}
923+ foldl' f z0 xs = Data.Foldable.foldr f' id xs z0
924+ where f' x k = oneShot $ \z -> k $! f z x
926+ {-# INLINE length #-}
927+ length = Data.Foldable.foldl' (\n _ -> n + 1) 0
932+ {-# INLINE maximum #-}
934+ fromMaybe (errorWithoutStackTrace "maximum: empty stream")
936+ . Data.Foldable.foldl' getMax Nothing'
940+ getMax Nothing' x = Just' x
941+ getMax (Just' mx) x = Just' $! max mx x
943+ {-# INLINE minimum #-}
945+ fromMaybe (errorWithoutStackTrace "minimum: empty stream")
947+ . Data.Foldable.foldl' getMin Nothing'
951+ getMin Nothing' x = Just' x
952+ getMin (Just' mn) x = Just' $! min mn x
955+ sum = Data.Foldable.foldl' (+) 0
957+ {-# INLINE product #-}
958+ product = Data.Foldable.foldl' (*) 1
636-------------------------------------------------------------------------------
960-------------------------------------------------------------------------------
637-- Filtering
961-- Filtering
638-------------------------------------------------------------------------------
962-------------------------------------------------------------------------------
640-- Adapted from the vector package.
964-- Adapted from the vector package.
966+-- | Take first 'n' elements from the stream and discard the rest.
641{-# INLINE_NORMAL take #-}
968{-# INLINE_NORMAL take #-}
642take :: Applicative m => Int -> Stream m a -> Stream m a
969take :: Applicative m => Int -> Stream m a -> Stream m a
643take n (Stream step state) = n `seq` Stream step' (state, 0)
970take n (Stream step state) = n `seq` Stream step' (state, 0)
@@ -653,8 +980,12 @@ take n (Stream step state) = n `seq` Stream step' (state, 0)
653 step' _ (_, _) = pure Stop
980 step' _ (_, _) = pure Stop
655-- Adapted from the vector package.
982-- Adapted from the vector package.
984+-- | Same as 'takeWhile' but with a monadic predicate.
656{-# INLINE_NORMAL takeWhileM #-}
986{-# INLINE_NORMAL takeWhileM #-}
657takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
987takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
988+-- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))
658takeWhileM f (Stream step state) = Stream step' state
989takeWhileM f (Stream step state) = Stream step' state
660 {-# INLINE_LATE step' #-}
991 {-# INLINE_LATE step' #-}
@@ -667,6 +998,8 @@ takeWhileM f (Stream step state) = Stream step' state
667 Skip s -> return $ Skip s
998 Skip s -> return $ Skip s
668 Stop -> return Stop
999 Stop -> return Stop
1001+-- | End the stream as soon as the predicate fails on an element.
670{-# INLINE takeWhile #-}
1003{-# INLINE takeWhile #-}
671takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
1004takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
672takeWhile f = takeWhileM (return . f)
1005takeWhile f = takeWhileM (return . f)
@@ -697,10 +1030,64 @@ takeEndByM f (Stream step state) = Stream step' (Just state)
697takeEndBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
1030takeEndBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
698takeEndBy f = takeEndByM (return . f)
1031takeEndBy f = takeEndByM (return . f)
1033+------------------------------------------------------------------------------
1035+------------------------------------------------------------------------------
1037+-- | Like 'zipWith' but using a monadic zipping function.
1039+{-# INLINE_NORMAL zipWithM #-}
1041+ => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
1042+zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)
1044+ {-# INLINE_LATE step #-}
1045+ step gst (sa, sb, Nothing) = do
1046+ r <- stepa (adaptState gst) sa
1049+ Yield x sa' -> Skip (sa', sb, Just x)
1050+ Skip sa' -> Skip (sa', sb, Nothing)
1053+ step gst (sa, sb, Just x) = do
1054+ r <- stepb (adaptState gst) sb
1058+ return $ Yield z (sa, sb', Nothing)
1059+ Skip sb' -> return $ Skip (sa, sb', Just x)
1060+ Stop -> return Stop
1062+{-# RULES "zipWithM xs xs"
1063+ forall f xs. zipWithM @Identity f xs xs = mapM (\x -> f x x) xs #-}
1065+-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
1066+-- elements @a@ and @b@ are then zipped using the supplied zip function and the
1067+-- result @c@ is yielded to the consumer.
1069+-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
1070+-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
1072+-- >>> s1 = Stream.fromList [1,2,3]
1073+-- >>> s2 = Stream.fromList [4,5,6]
1074+-- >>> Stream.fold Fold.toList $ Stream.zipWith (+) s1 s2
1077+{-# INLINE zipWith #-}
1078+zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1079+zipWith f = zipWithM (\a b -> return (f a b))
700------------------------------------------------------------------------------
1081------------------------------------------------------------------------------
701-- Combine N Streams - concatAp
1082-- Combine N Streams - concatAp
702------------------------------------------------------------------------------
1083------------------------------------------------------------------------------
1085+-- | Apply a stream of functions to a stream of values and flatten the results.
1087+-- Note that the second stream is evaluated multiple times.
1089+-- >>> crossApply = Stream.crossWith id
704{-# INLINE_NORMAL crossApply #-}
1091{-# INLINE_NORMAL crossApply #-}
705crossApply :: Functor f => Stream f (a -> b) -> Stream f a -> Stream f b
1092crossApply :: Functor f => Stream f (a -> b) -> Stream f a -> Stream f b
706crossApply (Stream stepa statea) (Stream stepb stateb) =
1093crossApply (Stream stepa statea) (Stream stepb stateb) =
@@ -786,6 +1173,39 @@ instance Applicative f => Applicative (Stream f) where
786 (<*) = crossApplyFst
1173 (<*) = crossApplyFst
1179+-- >>> crossWith f m1 m2 = fmap f m1 `Stream.crossApply` m2
1181+-- Note that the second stream is evaluated multiple times.
1183+{-# INLINE crossWith #-}
1184+crossWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1185+crossWith f m1 m2 = fmap f m1 `crossApply` m2
1187+-- | Given a @Stream m a@ and @Stream m b@ generate a stream with all possible
1188+-- combinations of the tuple @(a, b)@.
1192+-- >>> cross = Stream.crossWith (,)
1194+-- The second stream is evaluated multiple times. If that is not desired it can
1195+-- be cached in an 'Data.Array.Array' and then generated from the array before
1196+-- calling this function. Caching may also improve performance if the stream is
1197+-- expensive to evaluate.
1199+-- See 'Streamly.Internal.Data.Unfold.cross' for a much faster fused
1205+{-# INLINE cross #-}
1206+cross :: Monad m => Stream m a -> Stream m b -> Stream m (a, b)
1207+cross = crossWith (,)
789------------------------------------------------------------------------------
1209------------------------------------------------------------------------------
790-- Combine N Streams - unfoldMany
1210-- Combine N Streams - unfoldMany
791------------------------------------------------------------------------------
1211------------------------------------------------------------------------------
@@ -804,6 +1224,10 @@ data ConcatMapUState o i =
804-- optimization via fusion. This can be many times more efficient than
1224-- optimization via fusion. This can be many times more efficient than
805-- 'concatMap'.
1225-- 'concatMap'.
1227+-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
1228+-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
1229+-- therefore provide many times better performance.
807{-# INLINE_NORMAL unfoldMany #-}
1231{-# INLINE_NORMAL unfoldMany #-}
808unfoldMany :: Monad m => Unfold m a b -> Stream m a -> Stream m b
1232unfoldMany :: Monad m => Unfold m a b -> Stream m a -> Stream m b
809unfoldMany (Unfold istep inject) (Stream ostep ost) =
1233unfoldMany (Unfold istep inject) (Stream ostep ost) =
@@ -831,6 +1255,14 @@ unfoldMany (Unfold istep inject) (Stream ostep ost) =
831------------------------------------------------------------------------------
1255------------------------------------------------------------------------------
833-- Adapted from the vector package.
1257-- Adapted from the vector package.
1259+-- | Map a stream producing monadic function on each element of the stream
1260+-- and then flatten the results into a single stream. Since the stream
1261+-- generation function is monadic, unlike 'concatMap', it can produce an
1262+-- effect at the beginning of each iteration of the inner loop.
1264+-- See 'unfoldMany' for a fusible alternative.
834{-# INLINE_NORMAL concatMapM #-}
1266{-# INLINE_NORMAL concatMapM #-}
835concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
1267concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
836concatMapM f (Stream step state) = Stream step' (Left state)
1268concatMapM f (Stream step state) = Stream step' (Left state)
@@ -861,10 +1293,28 @@ concatMapM f (Stream step state) = Stream step' (Left state)
861 return $ Skip (Right (Stream inner_step inner_s, st))
1293 return $ Skip (Right (Stream inner_step inner_s, st))
862 Stop -> return $ Skip (Left st)
1294 Stop -> return $ Skip (Left st)
1296+-- | Map a stream producing function on each element of the stream and then
1297+-- flatten the results into a single stream.
1299+-- >>> concatMap f = Stream.concatMapM (return . f)
1300+-- >>> concatMap f = Stream.concat . fmap f
1301+-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
1303+-- See 'unfoldMany' for a fusible alternative.
864{-# INLINE concatMap #-}
1305{-# INLINE concatMap #-}
865concatMap :: Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
1306concatMap :: Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
866concatMap f = concatMapM (return . f)
1307concatMap f = concatMapM (return . f)
1309+-- | Flatten a stream of streams to a single stream.
1311+-- >>> concat = Stream.concatMap id
1314+{-# INLINE concat #-}
1315+concat :: Monad m => Stream m (Stream m a) -> Stream m a
1316+concat = concatMap id
868-- XXX The idea behind this rule is to rewrite any calls to "concatMap
1318-- XXX The idea behind this rule is to rewrite any calls to "concatMap
869-- fromArray" automatically to flattenArrays which is much faster. However, we
1319-- fromArray" automatically to flattenArrays which is much faster. However, we
870-- need an INLINE_EARLY on concatMap for this rule to fire. But if we use
1320-- need an INLINE_EARLY on concatMap for this rule to fire. But if we use
@@ -876,10 +1326,16 @@ concatMap f = concatMapM (return . f)
876-- {-# RULES "concatMap Array.toStreamD"
1326-- {-# RULES "concatMap Array.toStreamD"
877-- concatMap Array.toStreamD = Array.flattenArray #-}
1327-- concatMap Array.toStreamD = Array.flattenArray #-}
879−-- |
1329+-- >>> concatEffect = Stream.concat . lift -- requires (MonadTrans t)
880−-- Definition:
1330+-- >>> concatEffect = join . lift -- requires (MonadTrans t, Monad (Stream m))
1332+-- | Given a stream value in the underlying monad, lift and join the underlying
1333+-- monad with the stream monad.
1335+-- >>> concatEffect = Stream.concat . Stream.fromEffect
1336+-- >>> concatEffect eff = Stream.concatMapM (\() -> eff) (fromPure ())
882−-- > concatEffect generator = concatMapM (\() -> generator) (fromPure ())
1338+-- See also: 'concat', 'sequence'
884{-# INLINE concatEffect #-}
1340{-# INLINE concatEffect #-}
885concatEffect :: Monad m => m (Stream m a) -> Stream m a
1341concatEffect :: Monad m => m (Stream m a) -> Stream m a
@@ -903,6 +1359,13 @@ instance Monad m => Monad (Stream m) where
903-- Traversing a tree top down
1359-- Traversing a tree top down
904------------------------------------------------------------------------------
1360------------------------------------------------------------------------------
1362+-- Next stream is to be generated by the return value of the previous stream. A
1363+-- general intuitive way of doing that could be to use an appending monad
1364+-- instance for streams where the result of the previous stream is used to
1365+-- generate the next one. In the first pass we can just emit the values in the
1366+-- stream and keep building a buffered list/stream, once done we can then
1367+-- process the buffered stream.
906-- | Generate a stream from an initial state, scan and concat the stream,
1369-- | Generate a stream from an initial state, scan and concat the stream,
907-- generate a stream again from the final state of the previous scan and repeat
1370-- generate a stream again from the final state of the previous scan and repeat
908-- the process.
1371-- the process.
@@ -938,9 +1401,11 @@ concatIterateScan scanner generate initial = Stream step (Left initial)
938-- have to store any state. This makes the stored state proportional to the
1401-- have to store any state. This makes the stored state proportional to the
939-- number of non-leaf nodes rather than total number of nodes.
1402-- number of non-leaf nodes rather than total number of nodes.
1404+-- | Same as 'concatIterateBfs' except that the traversal of the last
1405+-- element on a level is emitted first and then going backwards up to the first
941−-- | This function may be slightly faster than concatIterateBfs because it
1406+-- element (reversed ordering). This may be slightly faster than
942−-- traverses the elements on a level in reverse order, therefore, does not have
1407+-- 'concatIterateBfs'.
943−-- to reverse the list stroing those.
1408+--
944{-# INLINE_NORMAL concatIterateBfsRev #-}
1409{-# INLINE_NORMAL concatIterateBfsRev #-}
945concatIterateBfsRev :: Monad m =>
1410concatIterateBfsRev :: Monad m =>
946 (a -> Maybe (Stream m a))
1411 (a -> Maybe (Stream m a))
@@ -966,6 +1431,17 @@ concatIterateBfsRev f stream = Stream step (stream, [])
966 (y:ys) -> return $ Skip (y, ys)
1431 (y:ys) -> return $ Skip (y, ys)
967 [] -> return Stop
1432 [] -> return Stop
1434+-- | Similar to 'concatIterateDfs' except that it traverses the stream in
1435+-- breadth first style (BFS). First, all the elements in the input stream are
1436+-- emitted, and then their traversals are emitted.
1438+-- Example, list a directory tree using BFS:
1440+-- >>> f = either (Just . Dir.readEitherPaths) (const Nothing)
1441+-- >>> input = Stream.fromPure (Left ".")
1442+-- >>> ls = Stream.concatIterateBfs f input
969{-# INLINE_NORMAL concatIterateBfs #-}
1445{-# INLINE_NORMAL concatIterateBfs #-}
970concatIterateBfs :: Monad m =>
1446concatIterateBfs :: Monad m =>
971 (a -> Maybe (Stream m a))
1447 (a -> Maybe (Stream m a))
@@ -994,6 +1470,19 @@ concatIterateBfs f stream = Stream step (stream, [], [])
994 (x:xs1) -> return $ Skip (x, xs1, [])
1470 (x:xs1) -> return $ Skip (x, xs1, [])
995 [] -> return Stop
1471 [] -> return Stop
1473+-- | Traverse the stream in depth first style (DFS). Map each element in the
1474+-- input stream to a stream and flatten, recursively map the resulting elements
1475+-- as well to a stream and flatten until no more streams are generated.
1477+-- Example, list a directory tree using DFS:
1479+-- >>> f = either (Just . Dir.readEitherPaths) (const Nothing)
1480+-- >>> input = Stream.fromPure (Left ".")
1481+-- >>> ls = Stream.concatIterateDfs f input
1483+-- This is equivalent to using @concatIterateWith StreamK.append@.
997{-# INLINE_NORMAL concatIterateDfs #-}
1486{-# INLINE_NORMAL concatIterateDfs #-}
998concatIterateDfs :: Monad m =>
1487concatIterateDfs :: Monad m =>
999 (a -> Maybe (Stream m a))
1488 (a -> Maybe (Stream m a))
@@ -1024,6 +1513,15 @@ data IterateUnfoldState o i =
1024 IterateUnfoldOuter o
1513 IterateUnfoldOuter o
1025 | IterateUnfoldInner o i [i]
1514 | IterateUnfoldInner o i [i]
1516+-- | Same as @concatIterateDfs@ but more efficient due to stream fusion.
1518+-- Example, list a directory tree using DFS:
1520+-- >>> f = Unfold.either Dir.eitherReaderPaths Unfold.nil
1521+-- >>> input = Stream.fromPure (Left ".")
1522+-- >>> ls = Stream.unfoldIterateDfs f input
1027{-# INLINE_NORMAL unfoldIterateDfs #-}
1525{-# INLINE_NORMAL unfoldIterateDfs #-}
1028unfoldIterateDfs :: Monad m =>
1526unfoldIterateDfs :: Monad m =>
1029 Unfold m a a
1527 Unfold m a a
@@ -1061,6 +1559,10 @@ data IterateUnfoldBFSRevState o i =
1061 IterateUnfoldBFSRevOuter o [i]
1559 IterateUnfoldBFSRevOuter o [i]
1062 | IterateUnfoldBFSRevInner i [i]
1560 | IterateUnfoldBFSRevInner i [i]
1562+-- | Like 'unfoldIterateBfs' but processes the children in reverse order,
1563+-- therefore, may be slightly faster.
1064{-# INLINE_NORMAL unfoldIterateBfsRev #-}
1566{-# INLINE_NORMAL unfoldIterateBfsRev #-}
1065unfoldIterateBfsRev :: Monad m =>
1567unfoldIterateBfsRev :: Monad m =>
1066 Unfold m a a
1568 Unfold m a a
@@ -1101,6 +1603,9 @@ data IterateUnfoldBFSState o i =
1101 IterateUnfoldBFSOuter o [i]
1603 IterateUnfoldBFSOuter o [i]
1102 | IterateUnfoldBFSInner i [i] [i]
1604 | IterateUnfoldBFSInner i [i] [i]
1606+-- | Like 'unfoldIterateDfs' but uses breadth first style traversal.
1104{-# INLINE_NORMAL unfoldIterateBfs #-}
1609{-# INLINE_NORMAL unfoldIterateBfs #-}
1105unfoldIterateBfs :: Monad m =>
1610unfoldIterateBfs :: Monad m =>
1106 Unfold m a a
1611 Unfold m a a
@@ -1143,9 +1648,10 @@ unfoldIterateBfs (Unfold istep inject) (Stream ostep ost) =
1143-- Folding a tree bottom up
1648-- Folding a tree bottom up
1144------------------------------------------------------------------------------
1649------------------------------------------------------------------------------
1146−-- | Binary BFS style reduce, folds a level entirely, before starting to fold
1651+-- | Binary BFS style reduce, folds a level entirely using the supplied fold
1652+-- function, collecting the outputs as next level of the tree, then repeats the
1147−-- the next level. The last elements of a previously folded level are folded
1653+-- same process on the next level. The last elements of a previously folded
1148−-- first.
1654+-- level are folded first.
1149{-# INLINE_NORMAL reduceIterateBfs #-}
1655{-# INLINE_NORMAL reduceIterateBfs #-}
1150reduceIterateBfs :: Monad m =>
1656reduceIterateBfs :: Monad m =>
1151 (a -> a -> m a) -> Stream m a -> m (Maybe a)
1657 (a -> a -> m a) -> Stream m a -> m (Maybe a)
@@ -1206,7 +1712,35 @@ data FoldManyPost s fs b a
1206 | FoldManyPostYield b (FoldManyPost s fs b a)
1712 | FoldManyPostYield b (FoldManyPost s fs b a)
1207 | FoldManyPostDone
1713 | FoldManyPostDone
1715+-- XXX Need a more intuitive name, and need to reconcile the names
1716+-- foldMany/fold/parse/parseMany/parseManyPost etc.
1718+-- | Like 'foldMany' but evaluates the fold before the stream, and yields its
1719+-- output even if the stream is empty, therefore, always results in a non-empty
1720+-- output even on an empty stream (default result of the fold).
1722+-- Example, empty stream:
1724+-- >>> f = Fold.take 2 Fold.sum
1209−-- | 'Streamly.Internal.Data.Stream.foldManyPost'.
1725+-- >>> fmany = Stream.fold Fold.toList . Stream.foldManyPost f
1726+-- >>> fmany $ Stream.fromList []
1729+-- Example, last fold empty:
1731+-- >>> fmany $ Stream.fromList [1..4]
1734+-- Example, last fold non-empty:
1736+-- >>> fmany $ Stream.fromList [1..5]
1739+-- Note that using a closed fold e.g. @Fold.take 0@, would result in an
1740+-- infinite stream without consuming the input.
1210{-# INLINE_NORMAL foldManyPost #-}
1744{-# INLINE_NORMAL foldManyPost #-}
1211foldManyPost :: Monad m => Fold m a b -> Stream m a -> Stream m b
1745foldManyPost :: Monad m => Fold m a b -> Stream m a -> Stream m b
1212foldManyPost (Fold fstep initial extract) (Stream step state) =
1746foldManyPost (Fold fstep initial extract) (Stream step state) =
@@ -1252,7 +1786,34 @@ data FoldMany s fs b a
1253-- XXX Nested foldMany does not fuse.
1787-- XXX Nested foldMany does not fuse.
1255−-- | 'Streamly.Internal.Data.Stream.foldMany'.
1789+-- | Apply a 'Fold' repeatedly on a stream and emit the results in the
1790+-- output stream. Unlike 'foldManyPost' it evaluates the fold after the stream,
1791+-- therefore, an empty input stream results in an empty output stream.
1795+-- >>> foldMany f = Stream.parseMany (Parser.fromFold f)
1797+-- Example, empty stream:
1799+-- >>> f = Fold.take 2 Fold.sum
1800+-- >>> fmany = Stream.fold Fold.toList . Stream.foldMany f
1801+-- >>> fmany $ Stream.fromList []
1804+-- Example, last fold empty:
1806+-- >>> fmany $ Stream.fromList [1..4]
1809+-- Example, last fold non-empty:
1811+-- >>> fmany $ Stream.fromList [1..5]
1814+-- Note that using a closed fold e.g. @Fold.take 0@, would result in an
1815+-- infinite stream on a non-empty input stream.
1256{-# INLINE_NORMAL foldMany #-}
1817{-# INLINE_NORMAL foldMany #-}
1257foldMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
1818foldMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
1258foldMany (Fold fstep initial extract) (Stream step state) =
1819foldMany (Fold fstep initial extract) (Stream step state) =
@@ -1340,3 +1901,122 @@ refoldMany (Refold fstep inject extract) action (Stream step state) =
1340 return $ Skip (FoldManyYield b FoldManyDone)
1901 return $ Skip (FoldManyYield b FoldManyDone)
1341 step' _ (FoldManyYield b next) = return $ Yield b next
1902 step' _ (FoldManyYield b next) = return $ Yield b next
1342 step' _ FoldManyDone = return Stop
1903 step' _ FoldManyDone = return Stop
1905+------------------------------------------------------------------------------
1906+-- Stream with a cross product style monad instance
1907+------------------------------------------------------------------------------
1909+-- XXX CrossStream performs better than the CrossStreamK when nesting two
1910+-- loops, however, CrossStreamK seems to be better for more than two nestings,
1911+-- need to do more perf investigation.
1913+-- | A newtype wrapper for the 'Stream' type with a cross product style monad
1916+-- A 'Monad' bind behaves like a @for@ loop:
1919+-- Stream.fold Fold.toList $ Stream.unCross $ do
1920+-- x <- Stream.mkCross $ Stream.fromList [1,2]
1921+-- -- Perform the following actions for each x in the stream
1926+-- Nested monad binds behave like nested @for@ loops:
1929+-- Stream.fold Fold.toList $ Stream.unCross $ do
1930+-- x <- Stream.mkCross $ Stream.fromList [1,2]
1931+-- y <- Stream.mkCross $ Stream.fromList [3,4]
1932+-- -- Perform the following actions for each x, for each y
1935+-- [(1,3),(1,4),(2,3),(2,4)]
1937+newtype CrossStream m a = CrossStream {unCrossStream :: Stream m a}
1938+ deriving (Functor, Foldable)
1940+{-# INLINE mkCross #-}
1941+mkCross :: Stream m a -> CrossStream m a
1942+mkCross = CrossStream
1944+{-# INLINE unCross #-}
1945+unCross :: CrossStream m a -> Stream m a
1946+unCross = unCrossStream
1948+-- Pure (Identity monad) stream instances
1949+deriving instance IsList (CrossStream Identity a)
1950+deriving instance (a ~ Char) => IsString (CrossStream Identity a)
1951+deriving instance Eq a => Eq (CrossStream Identity a)
1952+deriving instance Ord a => Ord (CrossStream Identity a)
1954+-- Do not use automatic derivation for this to show as "fromList" rather than
1955+-- "fromList Identity".
1956+instance Show a => Show (CrossStream Identity a) where
1957+ {-# INLINE show #-}
1958+ show (CrossStream xs) = show xs
1960+instance Read a => Read (CrossStream Identity a) where
1961+ {-# INLINE readPrec #-}
1962+ readPrec = fmap CrossStream readPrec
1964+------------------------------------------------------------------------------
1966+------------------------------------------------------------------------------
1968+-- Note: we need to define all the typeclass operations because we want to
1970+instance Monad m => Applicative (CrossStream m) where
1971+ {-# INLINE pure #-}
1972+ pure x = CrossStream (fromPure x)
1974+ {-# INLINE (<*>) #-}
1975+ (CrossStream s1) <*> (CrossStream s2) =
1976+ CrossStream (crossApply s1 s2)
1978+ {-# INLINE liftA2 #-}
1979+ liftA2 f x = (<*>) (fmap f x)
1981+ {-# INLINE (*>) #-}
1982+ (CrossStream s1) *> (CrossStream s2) =
1983+ CrossStream (crossApplySnd s1 s2)
1985+ {-# INLINE (<*) #-}
1986+ (CrossStream s1) <* (CrossStream s2) =
1987+ CrossStream (crossApplyFst s1 s2)
1989+------------------------------------------------------------------------------
1991+------------------------------------------------------------------------------
1993+instance Monad m => Monad (CrossStream m) where
1996+ -- Benchmarks better with StreamD bind and pure:
1997+ -- toList, filterAllout, *>, *<, >> (~2x)
2000+ -- Benchmarks better with CPS bind and pure:
2001+ -- Prime sieve (25x)
2002+ -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)
2004+ {-# INLINE (>>=) #-}
2005+ (>>=) (CrossStream m) f = CrossStream (concatMap (unCrossStream . f) m)
2007+ {-# INLINE (>>) #-}
2010+------------------------------------------------------------------------------
2012+------------------------------------------------------------------------------
2014+instance (MonadIO m) => MonadIO (CrossStream m) where
2015+ liftIO x = CrossStream (fromEffect $ liftIO x)
2017+instance MonadTrans CrossStream where
2018+ {-# INLINE lift #-}
2019+ lift x = CrossStream (fromEffect x)
2021+instance (MonadThrow m) => MonadThrow (CrossStream m) where
2022+ throwM = lift . throwM