@@ -17,26 +17,15 @@
18module Streamly.Internal.Data.Stream.StreamK.Type
18module Streamly.Internal.Data.Stream.StreamK.Type
20− -- * A class for streams
24 -- * The stream type
20 -- * The stream type
25− , Stream (..)
21+ Stream (..)
27− -- * Construction
25+ -- * foldr/build
28 , mkStream
26 , mkStream
34 , foldStream
27 , foldStream
35 , foldStreamShared
28 , foldStreamShared
42 , foldrSShared
31 , foldrSShared
@@ -45,27 +34,51 @@ module Streamly.Internal.Data.Stream.StreamK.Type
49 , augmentS
37 , augmentS
50 , augmentSM
38 , augmentSM
52− -- instances
40+ -- * Construction
55− , consMStream
46+ , consM
57 , fromEffect
52 , fromEffect
58 , fromPure
53 , fromPure
60− , nil
65+ -- * Elimination
66 , mapMSerial
79 , mapMSerial
68− , concatMapBy
81+ , concatMapWith
69 , concatMap
82 , concatMap
70 , bindWith
83 , bindWith
71 , concatPairsWith
84 , concatPairsWith
@@ -73,23 +86,25 @@ module Streamly.Internal.Data.Stream.StreamK.Type
73 , apSerial
86 , apSerial
74 , apSerialDiscardFst
87 , apSerialDiscardFst
75 , apSerialDiscardSnd
88 , apSerialDiscardSnd
77− , Streaming -- deprecated
92+ -- * Reader
81import Control.Monad (ap, (>=>))
97import Control.Monad (ap, (>=>))
98+import Control.Monad.Reader.Class (MonadReader(..))
82import Control.Monad.Trans.Class (MonadTrans(lift))
99import Control.Monad.Trans.Class (MonadTrans(lift))
83−import Data.Kind (Type)
100+import Data.Function (fix)
84−#if __GLASGOW_HASKELL__ < 808
101+#if __GLASGOW_HASKELL__ < 804
85import Data.Semigroup (Semigroup(..))
102import Data.Semigroup (Semigroup(..))
87−import Streamly.Internal.Control.Concurrent (MonadAsync)
88import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)
104import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)
89−import Prelude hiding (map, mapM, concatMap, foldr)
105+import Prelude hiding
90−
106+ (map, mapM, concatMap, foldr, repeat, null, reverse, tail, init)
91−-- $setup
107+import qualified Prelude
92−-- >>> import Streamly.Prelude as Stream
94------------------------------------------------------------------------------
109------------------------------------------------------------------------------
95-- Basic stream type
110-- Basic stream type
@@ -124,140 +139,22 @@ newtype Stream m a =
127−------------------------------------------------------------------------------
142+{-# INLINE fromStreamK #-}
128−-- Types that can behave as a Stream
143+fromStreamK :: Stream m a -> Stream m a
129−------------------------------------------------------------------------------
144+fromStreamK = id
134−-- XXX Use a different SVar based on the stream type. But we need to make sure
135−-- that we do not lose performance due to polymorphism.
137−-- | Class of types that can represent a stream of elements of some type 'a' in
140−-- /Since: 0.2.0 ("Streamly")/
144−#if __GLASGOW_HASKELL__ >= 806
145− ( forall m a. MonadAsync m => Semigroup (t m a)
146− , forall m a. MonadAsync m => Monoid (t m a)
147− , forall m. Monad m => Functor (t m)
148− , forall m. MonadAsync m => Applicative (t m)
152− toStream :: t m a -> Stream m a
153− fromStream :: Stream m a -> t m a
154− -- | Constructs a stream by adding a monadic action at the head of an
155− -- existing stream. For example:
158− -- > toList $ getLine \`consM` getLine \`consM` nil
161− -- ["hello","world"]
164− -- /Concurrent (do not use 'fromParallel' to construct infinite streams)/
167− consM :: MonadAsync m => m a -> t m a -> t m a
168− -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
169− -- to remember that @|@ comes before ':'.
172− -- > toList $ getLine |: getLine |: nil
175− -- ["hello","world"]
179− -- let delay = threadDelay 1000000 >> print 1
180− -- drain $ fromSerial $ delay |: delay |: delay |: nil
181− -- drain $ fromParallel $ delay |: delay |: delay |: nil
184− -- /Concurrent (do not use 'fromParallel' to construct infinite streams)/
187− (|:) :: MonadAsync m => m a -> t m a -> t m a
188− -- We can define (|:) just as 'consM' but it is defined explicitly for each
189− -- type because we want to use SPECIALIZE pragma on the definition.
191−-- | Same as 'IsStream'.
194−{-# DEPRECATED Streaming "Please use IsStream instead." #-}
195−type Streaming = IsStream
197−-------------------------------------------------------------------------------
198−-- Type adapting combinators
199−-------------------------------------------------------------------------------
201−-- XXX Move/reset the State here by reconstructing the stream with cleared
202−-- state. Can we make sure we do not do that when t1 = t2? If we do this then
203−-- we do not need to do that explicitly using svarStyle. It would act as
204−-- unShare when the stream type is the same.
206−-- | Adapt any specific stream type to any other specific stream type.
208−-- /Since: 0.1.0 ("Streamly")/
211−adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
212−adapt = fromStream . toStream
214−------------------------------------------------------------------------------
216−------------------------------------------------------------------------------
218−-- XXX The State is always parameterized by "Stream" which means State is not
146+{-# INLINE toStreamK #-}
219−-- different for different stream types. So we have to manually make sure that
147+toStreamK :: Stream m a -> Stream m a
220−-- when converting from one stream to another we migrate the state correctly.
148+toStreamK = id
221−-- This can be fixed if we use a different SVar type for different streams.
222−-- Currently we always use "SVar Stream" and therefore a different State type
223−-- parameterized by that stream.
225−-- XXX Since t is coercible we should be able to coerce k
226−-- mkStream k = fromStream $ MkStream $ coerce k
228−-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
229−-- continuation and a yield continuation.
230−{-# INLINE_EARLY mkStream #-}
231−mkStream :: IsStream t
232− => (forall r. State Stream m a
233− -> (a -> t m a -> m r)
238−mkStream k = fromStream $ MkStream $ \st yld sng stp ->
239− let yieldk a r = yld a (toStream r)
240− in k st yieldk sng stp
242−{-# RULES "mkStream from stream" mkStream = mkStreamFromStream #-}
243−mkStreamFromStream :: IsStream t
244− => (forall r. State Stream m a
245− -> (a -> Stream m a -> m r)
250−mkStreamFromStream k = fromStream $ MkStream k
252−{-# RULES "mkStream stream" mkStream = mkStreamStream #-}
150+mkStream
254 :: (forall r. State Stream m a
151 :: (forall r. State Stream m a
255 -> (a -> Stream m a -> m r)
152 -> (a -> Stream m a -> m r)
256 -> (a -> m r)
153 -> (a -> m r)
259 -> Stream m a
156 -> Stream m a
260−mkStreamStream = MkStream
157+mkStream = MkStream
262-- | A terminal function that has no continuation to follow.
159-- | A terminal function that has no continuation to follow.
263type StopK m = forall r. m r -> m r
160type StopK m = forall r. m r -> m r
@@ -272,17 +169,17 @@ _wrapM :: Monad m => m a -> YieldK m a
272_wrapM m = (m >>=)
169_wrapM m = (m >>=)
274-- | Make an empty stream from a stop function.
171-- | Make an empty stream from a stop function.
275−fromStopK :: IsStream t => StopK m -> t m a
172+fromStopK :: StopK m -> Stream m a
276fromStopK k = mkStream $ \_ _ _ stp -> k stp
173fromStopK k = mkStream $ \_ _ _ stp -> k stp
278-- | Make a singleton stream from a callback function. The callback function
175-- | Make a singleton stream from a callback function. The callback function
279-- calls the one-shot yield continuation to yield an element.
176-- calls the one-shot yield continuation to yield an element.
280−fromYieldK :: IsStream t => YieldK m a -> t m a
177+fromYieldK :: YieldK m a -> Stream m a
281fromYieldK k = mkStream $ \_ _ sng _ -> k sng
178fromYieldK k = mkStream $ \_ _ sng _ -> k sng
283-- | Add a yield function at the head of the stream.
180-- | Add a yield function at the head of the stream.
284−consK :: IsStream t => YieldK m a -> t m a -> t m a
181+consK :: YieldK m a -> Stream m a -> Stream m a
285−consK k r = mkStream $ \_ yld _ _ -> k (`yld` r)
182+consK k r = mkStream $ \_ yld _ _ -> k (\x -> yld x r)
287-- XXX Build a stream from a repeating callback function.
184-- XXX Build a stream from a repeating callback function.
@@ -305,8 +202,8 @@ infixr 5 `cons`
306-- @since 0.1.0
203-- @since 0.1.0
307{-# INLINE_NORMAL cons #-}
204{-# INLINE_NORMAL cons #-}
308−cons :: IsStream t => a -> t m a -> t m a
205+cons :: a -> Stream m a -> Stream m a
309−cons a r = mkStream $ \_ yld _ _ -> yld a r
206+cons a r = mkStream $ \_ yield _ _ -> yield a r
311infixr 5 .:
208infixr 5 .:
@@ -319,7 +216,7 @@ infixr 5 .:
320-- @since 0.1.1
217-- @since 0.1.1
321{-# INLINE (.:) #-}
218{-# INLINE (.:) #-}
322−(.:) :: IsStream t => a -> t m a -> t m a
219+(.:) :: a -> Stream m a -> Stream m a
323(.:) = cons
220(.:) = cons
325-- | An empty stream.
222-- | An empty stream.
@@ -331,7 +228,7 @@ infixr 5 .:
332-- @since 0.1.0
229-- @since 0.1.0
333{-# INLINE_NORMAL nil #-}
230{-# INLINE_NORMAL nil #-}
334−nil :: IsStream t => t m a
231+nil :: Stream m a
335nil = mkStream $ \_ _ _ stp -> stp
232nil = mkStream $ \_ _ _ stp -> stp
337-- | An empty stream producing a side effect.
234-- | An empty stream producing a side effect.
@@ -344,22 +241,32 @@ nil = mkStream $ \_ _ _ stp -> stp
345-- /Pre-release/
242-- /Pre-release/
346{-# INLINE_NORMAL nilM #-}
243{-# INLINE_NORMAL nilM #-}
347−nilM :: (IsStream t, Monad m) => m b -> t m a
244+nilM :: Monad m => m b -> Stream m a
348nilM m = mkStream $ \_ _ _ stp -> m >> stp
245nilM m = mkStream $ \_ _ _ stp -> m >> stp
350{-# INLINE_NORMAL fromPure #-}
247{-# INLINE_NORMAL fromPure #-}
351−fromPure :: IsStream t => a -> t m a
248+fromPure :: a -> Stream m a
352fromPure a = mkStream $ \_ _ single _ -> single a
249fromPure a = mkStream $ \_ _ single _ -> single a
354{-# INLINE_NORMAL fromEffect #-}
251{-# INLINE_NORMAL fromEffect #-}
355−fromEffect :: (Monad m, IsStream t) => m a -> t m a
252+fromEffect :: Monad m => m a -> Stream m a
356−fromEffect m = fromStream $ mkStream $ \_ _ single _ -> m >>= single
253+fromEffect m = mkStream $ \_ _ single _ -> m >>= single
257+-- NOTE: specializing the function outside the instance definition seems to
258+-- improve performance quite a bit at times, even if we have the same
259+-- SPECIALIZE in the instance definition.
261+{-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
262+consM :: (Monad m) => m a -> Stream m a -> Stream m a
263+consM m r = MkStream $ \_ yld _ _ -> m >>= \a -> yld a r
358-- XXX specialize to IO?
265-- XXX specialize to IO?
359{-# INLINE consMBy #-}
266{-# INLINE consMBy #-}
360−consMBy :: (IsStream t, MonadAsync m) => (t m a -> t m a -> t m a)
267+consMBy :: Monad m =>
361− -> m a -> t m a -> t m a
268+ (Stream m a -> Stream m a -> Stream m a) -> m a -> Stream m a -> Stream m a
362−consMBy f m r = fromStream (fromEffect m) `f` r
269+consMBy f m r = fromEffect m `f` r
364------------------------------------------------------------------------------
271------------------------------------------------------------------------------
365-- Folding a stream
272-- Folding a stream
@@ -370,78 +277,27 @@ consMBy f m r = fromStream (fromEffect m) `f` r
370-- SVar passed via the State.
277-- SVar passed via the State.
371{-# INLINE_EARLY foldStreamShared #-}
278{-# INLINE_EARLY foldStreamShared #-}
372foldStreamShared
279foldStreamShared
375− -> (a -> t m a -> m r)
380−foldStreamShared st yld sng stp m =
381− let yieldk a x = yld a (fromStream x)
382− MkStream k = toStream m
383− in k st yieldk sng stp
385−-- XXX write a similar rule for foldStream as well?
386−{-# RULES "foldStreamShared from stream"
387− foldStreamShared = foldStreamSharedStream #-}
388−foldStreamSharedStream
389 :: State Stream m a
280 :: State Stream m a
390 -> (a -> Stream m a -> m r)
281 -> (a -> Stream m a -> m r)
391 -> (a -> m r)
282 -> (a -> m r)
393 -> Stream m a
284 -> Stream m a
395−foldStreamSharedStream st yld sng stp m =
286+foldStreamShared s yield single stop (MkStream k) = k s yield single stop
396− let MkStream k = toStream m
399-- | Fold a stream by providing a State, stop continuation, a singleton
288-- | Fold a stream by providing a State, stop continuation, a singleton
400-- continuation and a yield continuation. The stream will not use the SVar
289-- continuation and a yield continuation. The stream will not use the SVar
401-- passed via State.
290-- passed via State.
402{-# INLINE foldStream #-}
291{-# INLINE foldStream #-}
403foldStream
292foldStream
405− => State Stream m a
293+ :: State Stream m a
406− -> (a -> t m a -> m r)
294+ -> (a -> Stream m a -> m r)
407 -> (a -> m r)
295 -> (a -> m r)
409− -> t m a
297+ -> Stream m a
411−foldStream st yld sng stp m =
299+foldStream s yield single stop (MkStream k) =
412− let yieldk a x = yld a (fromStream x)
413− MkStream k = toStream m
414− in k (adaptState st) yieldk sng stp
300+ k (adaptState s) yield single stop
416−-------------------------------------------------------------------------------
418−-------------------------------------------------------------------------------
420−-- NOTE: specializing the function outside the instance definition seems to
421−-- improve performance quite a bit at times, even if we have the same
422−-- SPECIALIZE in the instance definition.
423−{-# INLINE consMStream #-}
424−{-# SPECIALIZE consMStream :: IO a -> Stream IO a -> Stream IO a #-}
425−consMStream :: (Monad m) => m a -> Stream m a -> Stream m a
426−consMStream m r = MkStream $ \_ yld _ _ -> m >>= \a -> yld a r
428−-------------------------------------------------------------------------------
430−-------------------------------------------------------------------------------
432−instance IsStream Stream where
436− {-# INLINE consM #-}
437− {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
438− consM :: Monad m => m a -> Stream m a -> Stream m a
442− {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
443− (|:) :: Monad m => m a -> Stream m a -> Stream m a
446-------------------------------------------------------------------------------
302-------------------------------------------------------------------------------
447-- foldr/build fusion
303-- foldr/build fusion
@@ -454,15 +310,18 @@ instance IsStream Stream where
454-- reconstruct using a shared state (SVar) or without sharing the state.
310-- reconstruct using a shared state (SVar) or without sharing the state.
456{-# INLINE foldrSWith #-}
312{-# INLINE foldrSWith #-}
457−foldrSWith :: IsStream t
313+foldrSWith ::
458− => (forall r. State Stream m b
314+ (forall r. State Stream m b
459− -> (b -> t m b -> m r)
315+ -> (b -> Stream m b -> m r)
460 -> (b -> m r)
316 -> (b -> m r)
462− -> t m b
318+ -> Stream m b
464− -> (a -> t m b -> t m b) -> t m b -> t m a -> t m b
320+ -> (a -> Stream m b -> Stream m b)
465−foldrSWith f step final = go
324+foldrSWith f step final m = go m
467 go m1 = mkStream $ \st yld sng stp ->
326 go m1 = mkStream $ \st yld sng stp ->
468 let run x = f st yld sng stp x
327 let run x = f st yld sng stp x
@@ -480,7 +339,11 @@ foldrSWith f step final = go
481-- | Fold sharing the SVar state within the reconstructed stream
340-- | Fold sharing the SVar state within the reconstructed stream
482{-# INLINE_NORMAL foldrSShared #-}
341{-# INLINE_NORMAL foldrSShared #-}
483−foldrSShared :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
342+foldrSShared ::
343+ (a -> Stream m b -> Stream m b)
484foldrSShared = foldrSWith foldStreamShared
347foldrSShared = foldrSWith foldStreamShared
486-- XXX consM is a typeclass method, therefore rewritten already. Instead maybe
349-- XXX consM is a typeclass method, therefore rewritten already. Instead maybe
@@ -495,7 +358,11 @@ foldrSShared = foldrSWith foldStreamShared
496-- | Lazy right associative fold to a stream.
359-- | Lazy right associative fold to a stream.
497{-# INLINE_NORMAL foldrS #-}
360{-# INLINE_NORMAL foldrS #-}
498−foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
361+foldrS ::
362+ (a -> Stream m b -> Stream m b)
499foldrS = foldrSWith foldStream
366foldrS = foldrSWith foldStream
501{-# RULES "foldrS/id" foldrS cons nil = \x -> x #-}
368{-# RULES "foldrS/id" foldrS cons nil = \x -> x #-}
@@ -512,15 +379,18 @@ foldrS = foldrSWith foldStream
512-------------------------------------------------------------------------------
379-------------------------------------------------------------------------------
514{-# INLINE foldrSMWith #-}
381{-# INLINE foldrSMWith #-}
515−foldrSMWith :: (IsStream t, Monad m)
382+foldrSMWith :: Monad m
516 => (forall r. State Stream m b
383 => (forall r. State Stream m b
517− -> (b -> t m b -> m r)
384+ -> (b -> Stream m b -> m r)
518 -> (b -> m r)
385 -> (b -> m r)
520− -> t m b
387+ -> Stream m b
522− -> (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
389+ -> (m a -> Stream m b -> Stream m b)
523−foldrSMWith f step final = go
393+foldrSMWith f step final m = go m
525 go m1 = mkStream $ \st yld sng stp ->
395 go m1 = mkStream $ \st yld sng stp ->
526 let run x = f st yld sng stp x
396 let run x = f st yld sng stp x
@@ -530,8 +400,11 @@ foldrSMWith f step final = go
530 in foldStream (adaptState st) yieldk single stop m1
400 in foldStream (adaptState st) yieldk single stop m1
532{-# INLINE_NORMAL foldrSM #-}
402{-# INLINE_NORMAL foldrSM #-}
533−foldrSM :: (IsStream t, Monad m)
403+foldrSM :: Monad m
534− => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
404+ => (m a -> Stream m b -> Stream m b)
535foldrSM = foldrSMWith foldStream
408foldrSM = foldrSMWith foldStream
537-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
410-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
@@ -542,8 +415,11 @@ foldrSM = foldrSMWith foldStream
543-- Like foldrSM but sharing the SVar state within the recostructed stream.
416-- Like foldrSM but sharing the SVar state within the recostructed stream.
544{-# INLINE_NORMAL foldrSMShared #-}
417{-# INLINE_NORMAL foldrSMShared #-}
545−foldrSMShared :: (IsStream t, Monad m)
418+foldrSMShared :: Monad m
546− => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
419+ => (m a -> Stream m b -> Stream m b)
547foldrSMShared = foldrSMWith foldStreamShared
423foldrSMShared = foldrSMWith foldStreamShared
549-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
425-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
@@ -559,7 +435,7 @@ foldrSMShared = foldrSMWith foldStreamShared
559-------------------------------------------------------------------------------
435-------------------------------------------------------------------------------
561{-# INLINE_NORMAL build #-}
437{-# INLINE_NORMAL build #-}
562−build :: IsStream t => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a
438+build :: forall m a. (forall b. (a -> b -> b) -> b -> b) -> Stream m a
563build g = g cons nil
439build g = g cons nil
565{-# RULES "foldrM/build"
441{-# RULES "foldrM/build"
@@ -584,37 +460,46 @@ build g = g cons nil
585-- build a stream by applying cons and nil to a build function
461-- build a stream by applying cons and nil to a build function
586{-# INLINE_NORMAL buildS #-}
462{-# INLINE_NORMAL buildS #-}
587−buildS :: IsStream t => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a
463+buildS ::
464+ ((a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
588buildS g = g cons nil
466buildS g = g cons nil
590{-# RULES "foldrS/buildS"
468{-# RULES "foldrS/buildS"
591− forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
469+ forall k z
470+ (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
592 foldrS k z (buildS g) = g k z #-}
471 foldrS k z (buildS g) = g k z #-}
594{-# RULES "foldrS/cons/buildS"
473{-# RULES "foldrS/cons/buildS"
595− forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
474+ forall k z x
475+ (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
596 foldrS k z (x `cons` buildS g) = k x (g k z) #-}
476 foldrS k z (x `cons` buildS g) = k x (g k z) #-}
598{-# RULES "foldrSShared/buildS"
478{-# RULES "foldrSShared/buildS"
599− forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
479+ forall k z
480+ (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
600 foldrSShared k z (buildS g) = g k z #-}
481 foldrSShared k z (buildS g) = g k z #-}
602{-# RULES "foldrSShared/cons/buildS"
483{-# RULES "foldrSShared/cons/buildS"
603− forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
484+ forall k z x
485+ (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
604 foldrSShared k z (x `cons` buildS g) = k x (g k z) #-}
486 foldrSShared k z (x `cons` buildS g) = k x (g k z) #-}
606-- build a stream by applying consM and nil to a build function
488-- build a stream by applying consM and nil to a build function
607{-# INLINE_NORMAL buildSM #-}
489{-# INLINE_NORMAL buildSM #-}
608−buildSM :: (IsStream t, MonadAsync m)
490+buildSM :: Monad m
609− => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a
491+ => ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
610buildSM g = g consM nil
493buildSM g = g consM nil
612{-# RULES "foldrSM/buildSM"
495{-# RULES "foldrSM/buildSM"
613− forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
496+ forall k z
497+ (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
614 foldrSM k z (buildSM g) = g k z #-}
498 foldrSM k z (buildSM g) = g k z #-}
616{-# RULES "foldrSMShared/buildSM"
500{-# RULES "foldrSMShared/buildSM"
617− forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
501+ forall k z
502+ (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
618 foldrSMShared k z (buildSM g) = g k z #-}
503 foldrSMShared k z (buildSM g) = g k z #-}
620-- Disabled because this may not fire as consM is a class Op
505-- Disabled because this may not fire as consM is a class Op
@@ -629,71 +514,79 @@ buildSM g = g consM nil
629-- Build using monadic build functions (continuations) instead of
514-- Build using monadic build functions (continuations) instead of
630-- reconstructing a stream.
515-- reconstructing a stream.
631{-# INLINE_NORMAL buildM #-}
516{-# INLINE_NORMAL buildM #-}
632−buildM :: (IsStream t, MonadAsync m)
517+buildM :: Monad m
633− => (forall r. (a -> t m a -> m r)
518+ => (forall r. (a -> Stream m a -> m r)
634 -> (a -> m r)
519 -> (a -> m r)
638− -> t m a
523+ -> Stream m a
639buildM g = mkStream $ \st yld sng stp ->
524buildM g = mkStream $ \st yld sng stp ->
640 g (\a r -> foldStream st yld sng stp (return a `consM` r)) sng stp
525 g (\a r -> foldStream st yld sng stp (return a `consM` r)) sng stp
642-- | Like 'buildM' but shares the SVar state across computations.
527-- | Like 'buildM' but shares the SVar state across computations.
643−{-# INLINE_NORMAL sharedM #-}
528+{-# INLINE_NORMAL sharedMWith #-}
644−sharedM :: (IsStream t, MonadAsync m)
529+sharedMWith :: Monad m
530+ => (m a -> Stream m a -> Stream m a)
645− => (forall r. (a -> t m a -> m r)
531+ -> (forall r. (a -> Stream m a -> m r)
646 -> (a -> m r)
532 -> (a -> m r)
650− -> t m a
536+ -> Stream m a
651−sharedM g = mkStream $ \st yld sng stp ->
537+sharedMWith cns g = mkStream $ \st yld sng stp ->
652− g (\a r -> foldStreamShared st yld sng stp (return a `consM` r)) sng stp
538+ g (\a r -> foldStreamShared st yld sng stp (return a `cns` r)) sng stp
654-------------------------------------------------------------------------------
540-------------------------------------------------------------------------------
655-- augment
541-- augment
656-------------------------------------------------------------------------------
542-------------------------------------------------------------------------------
658{-# INLINE_NORMAL augmentS #-}
544{-# INLINE_NORMAL augmentS #-}
659−augmentS :: IsStream t
545+augmentS ::
660− => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
546+ ((a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
661−augmentS g = g cons
549+augmentS g xs = g cons xs
663{-# RULES "augmentS/nil"
551{-# RULES "augmentS/nil"
664− forall (g :: (a -> t m a -> t m a) -> t m a -> t m a).
552+ forall (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
665 augmentS g nil = buildS g
553 augmentS g nil = buildS g
668{-# RULES "foldrS/augmentS"
556{-# RULES "foldrS/augmentS"
669− forall k z xs (g :: (a -> t m a -> t m a) -> t m a -> t m a).
557+ forall k z xs
558+ (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
670 foldrS k z (augmentS g xs) = g k (foldrS k z xs)
559 foldrS k z (augmentS g xs) = g k (foldrS k z xs)
673{-# RULES "augmentS/buildS"
562{-# RULES "augmentS/buildS"
674− forall (g :: (a -> t m a -> t m a) -> t m a -> t m a)
563+ forall (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
675− (h :: (a -> t m a -> t m a) -> t m a -> t m a).
564+ (h :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
676 augmentS g (buildS h) = buildS (\c n -> g c (h c n))
565 augmentS g (buildS h) = buildS (\c n -> g c (h c n))
679{-# INLINE_NORMAL augmentSM #-}
568{-# INLINE_NORMAL augmentSM #-}
680−augmentSM :: (IsStream t, MonadAsync m)
569+augmentSM :: Monad m =>
681− => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
570+ ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
571+ -> Stream m a -> Stream m a
682−augmentSM g = g consM
572+augmentSM g xs = g consM xs
684{-# RULES "augmentSM/nil"
574{-# RULES "augmentSM/nil"
685− forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
575+ forall
576+ (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
686 augmentSM g nil = buildSM g
577 augmentSM g nil = buildSM g
689{-# RULES "foldrSM/augmentSM"
580{-# RULES "foldrSM/augmentSM"
690− forall k z xs (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
581+ forall k z xs
582+ (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
691 foldrSM k z (augmentSM g xs) = g k (foldrSM k z xs)
583 foldrSM k z (augmentSM g xs) = g k (foldrSM k z xs)
694{-# RULES "augmentSM/buildSM"
586{-# RULES "augmentSM/buildSM"
695− forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
587+ forall
588+ (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
696− (h :: (m a -> t m a -> t m a) -> t m a -> t m a).
589+ (h :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
697 augmentSM g (buildSM h) = buildSM (\c n -> g c (h c n))
590 augmentSM g (buildSM h) = buildSM (\c n -> g c (h c n))
@@ -703,8 +596,8 @@ augmentSM g = g consM
704-- | Lazy right fold with a monadic step function.
597-- | Lazy right fold with a monadic step function.
705{-# INLINE_NORMAL foldrM #-}
598{-# INLINE_NORMAL foldrM #-}
706−foldrM :: IsStream t => (a -> m b -> m b) -> m b -> t m a -> m b
599+foldrM :: (a -> m b -> m b) -> m b -> Stream m a -> m b
707−foldrM step acc = go
600+foldrM step acc m = go m
710 let stop = acc
603 let stop = acc
@@ -715,14 +608,14 @@ foldrM step acc = go
715{-# INLINE_NORMAL foldrMKWith #-}
608{-# INLINE_NORMAL foldrMKWith #-}
716foldrMKWith
609foldrMKWith
717 :: (State Stream m a
610 :: (State Stream m a
718− -> (a -> t m a -> m b)
611+ -> (a -> Stream m a -> m b)
719 -> (a -> m b)
612 -> (a -> m b)
721− -> t m a
614+ -> Stream m a
723 -> (a -> m b -> m b)
616 -> (a -> m b -> m b)
725− -> ((a -> t m a -> m b) -> (a -> m b) -> m b -> m b)
618+ -> ((a -> Stream m a -> m b) -> (a -> m b) -> m b -> m b)
727foldrMKWith f step acc = go
620foldrMKWith f step acc = go
@@ -742,7 +635,7 @@ foldrMKWith f step acc = go
742-- XXX in which case will foldrM/buildM fusion be useful?
635-- XXX in which case will foldrM/buildM fusion be useful?
743{-# RULES "foldrM/buildM"
636{-# RULES "foldrM/buildM"
744 forall step acc (g :: (forall r.
637 forall step acc (g :: (forall r.
745− (a -> t m a -> m r)
638+ (a -> Stream m a -> m r)
746 -> (a -> m r)
639 -> (a -> m r)
@@ -750,15 +643,17 @@ foldrMKWith f step acc = go
750 foldrM step acc (buildM g) = foldrMKWith foldStream step acc g
643 foldrM step acc (buildM g) = foldrMKWith foldStream step acc g
753{-# RULES "foldrM/sharedM"
647{-# RULES "foldrM/sharedM"
754 forall step acc (g :: (forall r.
648 forall step acc (g :: (forall r.
755− (a -> t m a -> m r)
649+ (a -> Stream m a -> m r)
756 -> (a -> m r)
650 -> (a -> m r)
760 foldrM step acc (sharedM g) = foldrMKWith foldStreamShared step acc g
654 foldrM step acc (sharedM g) = foldrMKWith foldStreamShared step acc g
763------------------------------------------------------------------------------
658------------------------------------------------------------------------------
764-- Left fold
659-- Left fold
@@ -771,12 +666,12 @@ foldrMKWith f step acc = go
772-- Note that the accumulator is always evaluated including the initial value.
667-- Note that the accumulator is always evaluated including the initial value.
773{-# INLINE foldlx' #-}
668{-# INLINE foldlx' #-}
774−foldlx' :: forall t m a b x. (IsStream t, Monad m)
669+foldlx' :: forall m a b x. Monad m
775− => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
670+ => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> m b
776foldlx' step begin done m = get $ go m begin
671foldlx' step begin done m = get $ go m begin
778 {-# NOINLINE get #-}
673 {-# NOINLINE get #-}
779− get :: t m x -> m b
674+ get :: Stream m x -> m b
781 -- XXX we are not strictly evaluating the accumulator here. Is this
676 -- XXX we are not strictly evaluating the accumulator here. Is this
@@ -788,7 +683,7 @@ foldlx' step begin done m = get $ go m begin
788 -- however that is more expensive because of unnecessary recursion
683 -- however that is more expensive because of unnecessary recursion
789 -- that cannot be tail call optimized. Unfolding recursion explicitly via
684 -- that cannot be tail call optimized. Unfolding recursion explicitly via
790 -- continuations is much more efficient.
685 -- continuations is much more efficient.
791− go :: t m a -> x -> t m x
686+ go :: Stream m a -> x -> Stream m x
792 go m1 !acc = mkStream $ \_ yld sng _ ->
687 go m1 !acc = mkStream $ \_ yld sng _ ->
793 let stop = sng acc
688 let stop = sng acc
794 single a = sng $ step acc a
689 single a = sng $ step acc a
@@ -799,9 +694,44 @@ foldlx' step begin done m = get $ go m begin
800-- | Strict left associative fold.
695-- | Strict left associative fold.
801{-# INLINE foldl' #-}
696{-# INLINE foldl' #-}
802−foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
697+foldl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> m b
803foldl' step begin = foldlx' step begin id
698foldl' step begin = foldlx' step begin id
700+------------------------------------------------------------------------------
702+------------------------------------------------------------------------------
704+-- XXX use foldrM to implement folds where possible
705+-- XXX This (commented) definition of drain and mapM_ perform much better on
706+-- some benchmarks but worse on others. Need to investigate why, may there is
707+-- an optimization opportunity that we can exploit.
708+-- drain = foldrM (\_ xs -> return () >> xs) (return ())
711+-- > drain = foldl' (\_ _ -> ()) ()
712+-- > drain = mapM_ (\_ -> return ())
714+drain :: Monad m => Stream m a -> m ()
715+drain = foldrM (\_ xs -> xs) (return ())
720+ let stop = return ()
721+ single _ = return ()
723+ in foldStream defState yieldk single stop m1
727+null :: Monad m => Stream m a -> m Bool
728+-- null = foldrM (\_ _ -> return True) (return False)
730+ let stop = return True
731+ single _ = return False
732+ yieldk _ _ = return False
733+ in foldStream defState yieldk single stop m
805------------------------------------------------------------------------------
735------------------------------------------------------------------------------
806-- Semigroup
736-- Semigroup
807------------------------------------------------------------------------------
737------------------------------------------------------------------------------
@@ -811,19 +741,8 @@ infixr 6 `serial`
811-- | Appends two streams sequentially, yielding all elements from the first
741-- | Appends two streams sequentially, yielding all elements from the first
812-- stream, and then all elements from the second stream.
742-- stream, and then all elements from the second stream.
814−-- >>> import Streamly.Prelude (serial)
815−-- >>> stream1 = Stream.fromList [1,2]
816−-- >>> stream2 = Stream.fromList [3,4]
817−-- >>> Stream.toList $ stream1 `serial` stream2
820−-- This operation can be used to fold an infinite lazy container of streams.
822−-- /Since: 0.2.0 ("Streamly")/
825{-# INLINE serial #-}
744{-# INLINE serial #-}
826−serial :: IsStream t => t m a -> t m a -> t m a
745+serial :: Stream m a -> Stream m a -> Stream m a
827-- XXX This doubles the time of toNullAp benchmark, may not be fusing properly
746-- XXX This doubles the time of toNullAp benchmark, may not be fusing properly
828-- serial xs ys = augmentS (\c n -> foldrS c n xs) ys
747-- serial xs ys = augmentS (\c n -> foldrS c n xs) ys
829serial m1 m2 = go m1
748serial m1 m2 = go m1
@@ -836,7 +755,7 @@ serial m1 m2 = go m1
837-- join/merge/append streams depending on consM
756-- join/merge/append streams depending on consM
838{-# INLINE conjoin #-}
757{-# INLINE conjoin #-}
839−conjoin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
758+conjoin :: Monad m => Stream m a -> Stream m a -> Stream m a
840conjoin xs = augmentSM (\c n -> foldrSM c n xs)
759conjoin xs = augmentSM (\c n -> foldrSM c n xs)
842instance Semigroup (Stream m a) where
761instance Semigroup (Stream m a) where
@@ -856,10 +775,13 @@ instance Monoid (Stream m a) where
857-- Note eta expanded
776-- Note eta expanded
858{-# INLINE_LATE mapFB #-}
777{-# INLINE_LATE mapFB #-}
859−mapFB :: forall (t :: (Type -> Type) -> Type -> Type) b m a.
778+mapFB :: forall b m a.
860− (b -> t m b -> t m b) -> (a -> b) -> a -> t m b -> t m b
779+ (b -> Stream m b -> Stream m b)
861−mapFB c f x ys = c (f x) ys
784+mapFB c f = \x ys -> c (f x) ys
865"mapFB/mapFB" forall c f g. mapFB (mapFB c f) g = mapFB c (f . g)
787"mapFB/mapFB" forall c f g. mapFB (mapFB c f) g = mapFB c (f . g)
@@ -867,7 +789,7 @@ mapFB c f x ys = c (f x) ys
869{-# INLINE map #-}
791{-# INLINE map #-}
870−map :: IsStream t => (a -> b) -> t m a -> t m b
792+map :: (a -> b) -> Stream m a -> Stream m b
871map f xs = buildS (\c n -> foldrS (mapFB c f) n xs)
793map f xs = buildS (\c n -> foldrS (mapFB c f) n xs)
873-- XXX This definition might potentially be more efficient, but the cost in the
795-- XXX This definition might potentially be more efficient, but the cost in the
@@ -898,21 +820,6 @@ mapMFB c f x = c (x >>= f)
898"mapMFB/return" forall c. mapMFB c (\x -> return x) = c
820"mapMFB/return" forall c. mapMFB c (\x -> return x) = c
901−-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
902−-- other stream types to overload it.
904−mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
905−mapM f = foldrSShared (\x xs -> f x `consM` xs) nil
906−-- See note under map definition above.
910− go m1 = mkStream $ \st yld sng stp ->
911− let single a = f a >>= sng
912− yieldk a r = foldStreamShared st yld sng stp $ f a |: go r
913− in foldStream (adaptState st) yieldk single stp m1
916-- This is experimental serial version supporting fusion.
823-- This is experimental serial version supporting fusion.
918-- XXX what if we do not want to fuse two concurrent mapMs?
825-- XXX what if we do not want to fuse two concurrent mapMs?
@@ -921,9 +828,27 @@ mapM f m = go m
921-- XXX fusion would be easier for monomoprhic stream types.
828-- XXX fusion would be easier for monomoprhic stream types.
922-- {-# RULES "mapM serial" mapM = mapMSerial #-}
829-- {-# RULES "mapM serial" mapM = mapMSerial #-}
923{-# INLINE mapMSerial #-}
830{-# INLINE mapMSerial #-}
924−mapMSerial :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
831+mapMSerial :: Monad m => (a -> m b) -> Stream m a -> Stream m b
925mapMSerial f xs = buildSM (\c n -> foldrSMShared (mapMFB c f) n xs)
832mapMSerial f xs = buildSM (\c n -> foldrSMShared (mapMFB c f) n xs)
834+{-# INLINE mapMWith #-}
836+ (m b -> Stream m b -> Stream m b)
840+mapMWith cns f = foldrSShared (\x xs -> f x `cns` xs) nil
843+-- See note under map definition above.
846+ go m1 = mkStream $ \st yld sng stp ->
847+ let single a = f a >>= sng
848+ yieldk a r = foldStreamShared st yld sng stp $ f a `cns` go r
849+ in foldStream (adaptState st) yieldk single stp m1
927-- XXX in fact use the Stream type everywhere and only use polymorphism in the
852-- XXX in fact use the Stream type everywhere and only use polymorphism in the
928-- high level modules/prelude.
853-- high level modules/prelude.
929instance Monad m => Functor (Stream m) where
854instance Monad m => Functor (Stream m) where
@@ -943,18 +868,17 @@ instance MonadTrans Stream where
944-- | Detach a stream from an SVar
869-- | Detach a stream from an SVar
945{-# INLINE unShare #-}
870{-# INLINE unShare #-}
946−unShare :: IsStream t => t m a -> t m a
871+unShare :: Stream m a -> Stream m a
947unShare x = mkStream $ \st yld sng stp ->
872unShare x = mkStream $ \st yld sng stp ->
948 foldStream st yld sng stp x
873 foldStream st yld sng stp x
950-- XXX the function stream and value stream can run in parallel
875-- XXX the function stream and value stream can run in parallel
951{-# INLINE apWith #-}
876{-# INLINE apWith #-}
953− :: IsStream t
878+ (Stream m b -> Stream m b -> Stream m b)
954− => (t m b -> t m b -> t m b)
955− -> t m (a -> b)
879+ -> Stream m (a -> b)
956− -> t m a
880+ -> Stream m a
957− -> t m b
881+ -> Stream m b
958apWith par fstream stream = go1 fstream
882apWith par fstream stream = go1 fstream
@@ -973,11 +897,10 @@ apWith par fstream stream = go1 fstream
973 in foldStream (adaptState st) yieldk single stp m
897 in foldStream (adaptState st) yieldk single stp m
975{-# INLINE apSerial #-}
899{-# INLINE apSerial #-}
976−apSerial
900+apSerial ::
978− => t m (a -> b)
901+ Stream m (a -> b)
979− -> t m a
902+ -> Stream m a
980− -> t m b
903+ -> Stream m b
981apSerial fstream stream = go1 fstream
904apSerial fstream stream = go1 fstream
@@ -1004,11 +927,10 @@ apSerial fstream stream = go1 fstream
1004 in foldStream (adaptState st) yieldk single stp m
927 in foldStream (adaptState st) yieldk single stp m
1006{-# INLINE apSerialDiscardFst #-}
929{-# INLINE apSerialDiscardFst #-}
1007−apSerialDiscardFst
930+apSerialDiscardFst ::
1008− :: IsStream t
931+ Stream m a
1009− => t m a
932+ -> Stream m b
1010− -> t m b
933+ -> Stream m b
1012apSerialDiscardFst fstream stream = go1 fstream
934apSerialDiscardFst fstream stream = go1 fstream
@@ -1029,11 +951,10 @@ apSerialDiscardFst fstream stream = go1 fstream
1029 in foldStream st yieldk single stop m
951 in foldStream st yieldk single stop m
1031{-# INLINE apSerialDiscardSnd #-}
953{-# INLINE apSerialDiscardSnd #-}
1032−apSerialDiscardSnd
954+apSerialDiscardSnd ::
1033− :: IsStream t
955+ Stream m a
1034− => t m a
956+ -> Stream m b
1035− -> t m b
957+ -> Stream m a
1037apSerialDiscardSnd fstream stream = go1 fstream
958apSerialDiscardSnd fstream stream = go1 fstream
@@ -1059,17 +980,16 @@ apSerialDiscardSnd fstream stream = go1 fstream
1059 yieldk _ r = yld f (go3 f r)
980 yieldk _ r = yld f (go3 f r)
1060 in foldStream (adaptState st) yieldk single stp m
981 in foldStream (adaptState st) yieldk single stp m
1062−-- XXX This is just concatMapBy with arguments flipped. We need to keep this
983+-- XXX This is just concatMapWith with arguments flipped. We need to keep this
1063-- instead of using a concatMap style definition because the bind
984-- instead of using a concatMap style definition because the bind
1064-- implementation in Async and WAsync streams show significant perf degradation
985-- implementation in Async and WAsync streams show significant perf degradation
1065-- if the argument order is changed.
986-- if the argument order is changed.
1066{-# INLINE bindWith #-}
987{-# INLINE bindWith #-}
1067−bindWith
988+bindWith ::
1068− :: IsStream t
989+ (Stream m b -> Stream m b -> Stream m b)
1069− => (t m b -> t m b -> t m b)
990+ -> Stream m a
1071− -> (a -> t m b)
991+ -> (a -> Stream m b)
1072− -> t m b
992+ -> Stream m b
1073bindWith par m1 f = go m1
993bindWith par m1 f = go m1
@@ -1094,21 +1014,18 @@ bindWith par m1 f = go m1
1094-- function.
1014-- function.
1096-- @since 0.7.0
1016-- @since 0.7.0
1097−{-# INLINE concatMapBy #-}
1017+{-# INLINE concatMapWith #-}
1098−concatMapBy
1018+concatMapWith
1099− :: IsStream t
1019+ ::
1100− => (t m b -> t m b -> t m b)
1020+ (Stream m b -> Stream m b -> Stream m b)
1101− -> (a -> t m b)
1021+ -> (a -> Stream m b)
1102− -> t m a
1022+ -> Stream m a
1103− -> t m b
1023+ -> Stream m b
1104−concatMapBy par f xs = bindWith par xs f
1024+concatMapWith par f xs = bindWith par xs f
1106{-# INLINE concatMap #-}
1026{-# INLINE concatMap #-}
1107−concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
1027+concatMap :: (a -> Stream m b) -> Stream m a -> Stream m b
1108−concatMap f m = fromStream $
1028+concatMap = concatMapWith serial
1110− (adapt . toStream . f)
1111− (adapt $ toStream m)
1114-- Fused version.
1031-- Fused version.
@@ -1133,11 +1050,11 @@ concatMap_ f xs = buildS
1134{-# INLINE concatPairsWith #-}
1051{-# INLINE concatPairsWith #-}
1135concatPairsWith
1052concatPairsWith
1136− :: IsStream t
1053+ ::
1137− => (t m b -> t m b -> t m b)
1054+ (Stream m b -> Stream m b -> Stream m b)
1138− -> (a -> t m b)
1055+ -> (a -> Stream m b)
1139− -> t m a
1056+ -> Stream m a
1140− -> t m b
1057+ -> Stream m b
1141concatPairsWith combine f = go Nothing
1058concatPairsWith combine f = go Nothing
@@ -1195,3 +1112,194 @@ concatUnfoldr :: IsStream t
1195 => (b -> t m (Maybe (a, b))) -> t m b -> t m a
1112 => (b -> t m (Maybe (a, b))) -> t m b -> t m a
1196concatUnfoldr = undefined
1113concatUnfoldr = undefined
1116+------------------------------------------------------------------------------
1118+------------------------------------------------------------------------------
1120+{-# INLINABLE withLocal #-}
1121+withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1123+ mkStream $ \st yld sng stp ->
1124+ let single = local f . sng
1125+ yieldk a r = local f $ yld a (withLocal f r)
1126+ in foldStream st yieldk single (local f stp) m
1128+-------------------------------------------------------------------------------
1130+-------------------------------------------------------------------------------
1132+{-# INLINE unfoldr #-}
1133+unfoldr :: (b -> Maybe (a, b)) -> b -> Stream m a
1134+unfoldr next s0 = build $ \yld stp ->
1137+ Just (a, b) -> yld a (go b)
1141+{-# INLINE unfoldrMWith #-}
1142+unfoldrMWith :: Monad m =>
1143+ (m a -> Stream m a -> Stream m a)
1144+ -> (b -> m (Maybe (a, b)))
1147+unfoldrMWith cns step = go
1151+ go s = sharedMWith cns $ \yld _ stp -> do
1154+ Just (a, b) -> yld a (go b)
1157+-- | Generate an infinite stream by repeating a pure value.
1160+{-# INLINE repeat #-}
1161+repeat :: a -> Stream m a
1162+repeat a = let x = cons a x in x
1164+-- | Like 'repeatM' but takes a stream 'cons' operation to combine the actions
1165+-- in a stream specific manner. A serial cons would repeat the values serially
1166+-- while an async cons would repeat concurrently.
1169+repeatMWith :: (m a -> t m a -> t m a) -> m a -> t m a
1170+repeatMWith cns = go
1174+ go m = m `cns` go m
1176+{-# INLINE replicateMWith #-}
1177+replicateMWith :: (m a -> Stream m a -> Stream m a) -> Int -> m a -> Stream m a
1178+replicateMWith cns n m = go n
1182+ go cnt = if cnt <= 0 then nil else m `cns` go (cnt - 1)
1184+{-# INLINE fromIndicesMWith #-}
1186+ (m a -> Stream m a -> Stream m a) -> (Int -> m a) -> Stream m a
1187+fromIndicesMWith cns gen = go 0
1191+ go i = mkStream $ \st stp sng yld -> do
1192+ foldStreamShared st stp sng yld (gen i `cns` go (i + 1))
1194+{-# INLINE iterateMWith #-}
1195+iterateMWith :: Monad m =>
1196+ (m a -> Stream m a -> Stream m a) -> (a -> m a) -> m a -> Stream m a
1197+iterateMWith cns step = go
1201+ go s = mkStream $ \st stp sng yld -> do
1203+ foldStreamShared st stp sng yld (return next `cns` go (step next))
1205+{-# INLINE headPartial #-}
1206+headPartial :: Monad m => Stream m a -> m a
1207+headPartial = foldrM (\x _ -> return x) (error "head of nil")
1209+{-# INLINE tailPartial #-}
1210+tailPartial :: Stream m a -> Stream m a
1211+tailPartial m = mkStream $ \st yld sng stp ->
1212+ let stop = error "tail of nil"
1214+ yieldk _ r = foldStream st yld sng stp r
1215+ in foldStream st yieldk single stop m
1218+mfix :: Monad m => (m a -> Stream m a) -> Stream m a
1219+mfix f = mkStream $ \st yld sng stp ->
1220+ let single a = foldStream st yld sng stp $ a `cons` ys
1221+ yieldk a _ = foldStream st yld sng stp $ a `cons` ys
1222+ in foldStream st yieldk single stp xs
1226+ -- fix the head element of the stream
1227+ xs = fix (f . headPartial)
1229+ -- now fix the tail recursively
1230+ ys = mfix (tailPartial . f)
1232+-------------------------------------------------------------------------------
1234+-------------------------------------------------------------------------------
1238+-- fromFoldable = 'Prelude.foldr' 'cons' 'nil'
1241+-- Construct a stream from a 'Foldable' containing pure values:
1244+{-# INLINE fromFoldable #-}
1245+fromFoldable :: Foldable f => f a -> Stream m a
1246+fromFoldable = Prelude.foldr cons nil
1248+{-# INLINE fromFoldableM #-}
1249+fromFoldableM :: (Foldable f, Monad m) => f (m a) -> Stream m a
1250+fromFoldableM = Prelude.foldr consM nil
1252+-------------------------------------------------------------------------------
1254+-------------------------------------------------------------------------------
1256+{-# INLINE uncons #-}
1257+uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
1259+ let stop = return Nothing
1260+ single a = return (Just (a, nil))
1261+ yieldk a r = return (Just (a, r))
1262+ in foldStream defState yieldk single stop m
1265+tail :: Monad m => Stream m a -> m (Maybe (Stream m a))
1267+ let stop = return Nothing
1268+ single _ = return $ Just nil
1269+ yieldk _ r = return $ Just r
1270+ in foldStream defState yieldk single stop
1273+init :: Monad m => Stream m a -> m (Maybe (Stream m a))
1279+ Nothing -> return Nothing
1280+ Just (h, t) -> return . Just $ go h t
1281+ go p m1 = mkStream $ \_ yld sng stp ->
1282+ let single _ = sng p
1283+ yieldk a x = yld p $ go a x
1284+ in foldStream defState yieldk single stp m1
1286+------------------------------------------------------------------------------
1288+------------------------------------------------------------------------------
1290+-- | Lazy left fold to a stream.
1291+{-# INLINE foldlS #-}
1293+ (Stream m b -> a -> Stream m b) -> Stream m b -> Stream m a -> Stream m b
1296+ go acc rest = mkStream $ \st yld sng stp ->
1297+ let run x = foldStream st yld sng stp x
1299+ single a = run $ step acc a
1300+ yieldk a r = run $ go (step acc a) r
1301+ in foldStream (adaptState st) yieldk single stop rest
1303+{-# INLINE reverse #-}
1304+reverse :: Stream m a -> Stream m a
1305+reverse = foldlS (flip cons) nil