@@ -17,26 +17,15 @@
1717--
1818module Streamly.Internal.Data.Stream.StreamK.Type
1919 (
20− -- * A class for streams
21− IsStream (..)
22− , adapt
23−
2420 -- * The stream type
25− , Stream (..)
21+ Stream (..)
22+ , toStreamK
23+ , fromStreamK
2624
27− -- * Construction
25+ -- * foldr/build
2826 , mkStream
29− , fromStopK
30− , fromYieldK
31− , consK
32−
33− -- * Elimination
3427 , foldStream
3528 , foldStreamShared
36− , foldl'
37− , foldlx'
38−
39− -- * foldr/build
4029 , foldrM
4130 , foldrS
4231 , foldrSShared
@@ -45,27 +34,51 @@ module Streamly.Internal.Data.Stream.StreamK.Type
4534 , buildS
4635 , buildM
4736 , buildSM
48− , sharedM
4937 , augmentS
5038 , augmentSM
5139
52− -- instances
40+ -- * Construction
41+ , fromStopK
42+ , fromYieldK
43+ , consK
5344 , cons
5445 , (.:)
55− , consMStream
46+ , consM
5647 , consMBy
48+ , nil
49+ , nilM
50+
51+ -- * Generation
5752 , fromEffect
5853 , fromPure
54+ , unfoldr
55+ , unfoldrMWith
56+ , repeat
57+ , repeatMWith
58+ , replicateMWith
59+ , fromIndicesMWith
60+ , iterateMWith
61+ , fromFoldable
62+ , fromFoldableM
63+ , mfix
5964
60− , nil
61− , nilM
65+ -- * Elimination
66+ , uncons
67+ , foldl'
68+ , foldlx'
69+ , drain
70+ , null
71+ , tail
72+ , init
73+
74+ -- * Transformation
6275 , conjoin
6376 , serial
6477 , map
65− , mapM
78+ , mapMWith
6679 , mapMSerial
6780 , unShare
68− , concatMapBy
81+ , concatMapWith
6982 , concatMap
7083 , bindWith
7184 , concatPairsWith
@@ -73,23 +86,25 @@ module Streamly.Internal.Data.Stream.StreamK.Type
7386 , apSerial
7487 , apSerialDiscardFst
7588 , apSerialDiscardSnd
89+ , foldlS
90+ , reverse
7691
77− , Streaming -- deprecated
92+ -- * Reader
93+ , withLocal
7894 )
7995where
8096
8197import Control.Monad (ap, (>=>))
98+import Control.Monad.Reader.Class (MonadReader(..))
8299import Control.Monad.Trans.Class (MonadTrans(lift))
83−import Data.Kind (Type)
84−#if __GLASGOW_HASKELL__ < 808
100+import Data.Function (fix)
101+#if __GLASGOW_HASKELL__ < 804
85102import Data.Semigroup (Semigroup(..))
86103#endif
87−import Streamly.Internal.Control.Concurrent (MonadAsync)
88104import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)
89−import Prelude hiding (map, mapM, concatMap, foldr)
90−
91−-- $setup
92−-- >>> import Streamly.Prelude as Stream
105+import Prelude hiding
106+ (map, mapM, concatMap, foldr, repeat, null, reverse, tail, init)
107+import qualified Prelude
93108
94109------------------------------------------------------------------------------
95110-- Basic stream type
@@ -124,140 +139,22 @@ newtype Stream m a =
124139 -> m r
125140 )
126141
127−------------------------------------------------------------------------------
128−-- Types that can behave as a Stream
129−------------------------------------------------------------------------------
130−
131−infixr 5 `consM`
132−infixr 5 |:
133−
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.
136−--
137−-- | Class of types that can represent a stream of elements of some type 'a' in
138−-- some monad 'm'.
139−--
140−-- /Since: 0.2.0 ("Streamly")/
141−--
142−-- @since 0.8.0
143−class
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)
149− ) =>
150−#endif
151− IsStream t where
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:
156− --
157− -- @
158− -- > toList $ getLine \`consM` getLine \`consM` nil
159− -- hello
160− -- world
161− -- ["hello","world"]
162− -- @
163− --
164− -- /Concurrent (do not use 'fromParallel' to construct infinite streams)/
165− --
166− -- @since 0.2.0
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 ':'.
170− --
171− -- @
172− -- > toList $ getLine |: getLine |: nil
173− -- hello
174− -- world
175− -- ["hello","world"]
176− -- @
177− --
178− -- @
179− -- let delay = threadDelay 1000000 >> print 1
180− -- drain $ fromSerial $ delay |: delay |: delay |: nil
181− -- drain $ fromParallel $ delay |: delay |: delay |: nil
182− -- @
183− --
184− -- /Concurrent (do not use 'fromParallel' to construct infinite streams)/
185− --
186− -- @since 0.2.0
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.
190−
191−-- | Same as 'IsStream'.
192−--
193−-- @since 0.1.0
194−{-# DEPRECATED Streaming "Please use IsStream instead." #-}
195−type Streaming = IsStream
196−
197−-------------------------------------------------------------------------------
198−-- Type adapting combinators
199−-------------------------------------------------------------------------------
200−
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.
205−--
206−-- | Adapt any specific stream type to any other specific stream type.
207−--
208−-- /Since: 0.1.0 ("Streamly")/
209−--
210−-- @since 0.8.0
211−adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
212−adapt = fromStream . toStream
213−
214−------------------------------------------------------------------------------
215−-- Building a stream
216−------------------------------------------------------------------------------
142+{-# INLINE fromStreamK #-}
143+fromStreamK :: Stream m a -> Stream m a
144+fromStreamK = id
217145
218−-- XXX The State is always parameterized by "Stream" which means State is not
219−-- different for different stream types. So we have to manually make sure that
220−-- when converting from one stream to another we migrate the state correctly.
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.
224−--
225−-- XXX Since t is coercible we should be able to coerce k
226−-- mkStream k = fromStream $ MkStream $ coerce k
227−--
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)
234− -> (a -> m r)
235− -> m r
236− -> m r)
237− -> t m a
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
241−
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)
246− -> (a -> m r)
247− -> m r
248− -> m r)
249− -> t m a
250−mkStreamFromStream k = fromStream $ MkStream k
146+{-# INLINE toStreamK #-}
147+toStreamK :: Stream m a -> Stream m a
148+toStreamK = id
251149
252−{-# RULES "mkStream stream" mkStream = mkStreamStream #-}
253−mkStreamStream
150+mkStream
254151 :: (forall r. State Stream m a
255152 -> (a -> Stream m a -> m r)
256153 -> (a -> m r)
257154 -> m r
258155 -> m r)
259156 -> Stream m a
260−mkStreamStream = MkStream
157+mkStream = MkStream
261158
262159-- | A terminal function that has no continuation to follow.
263160type StopK m = forall r. m r -> m r
@@ -272,17 +169,17 @@ _wrapM :: Monad m => m a -> YieldK m a
272169_wrapM m = (m >>=)
273170
274171-- | Make an empty stream from a stop function.
275−fromStopK :: IsStream t => StopK m -> t m a
172+fromStopK :: StopK m -> Stream m a
276173fromStopK k = mkStream $ \_ _ _ stp -> k stp
277174
278175-- | Make a singleton stream from a callback function. The callback function
279176-- 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
281178fromYieldK k = mkStream $ \_ _ sng _ -> k sng
282179
283180-- | Add a yield function at the head of the stream.
284−consK :: IsStream t => YieldK m a -> t m a -> t m a
285−consK k r = mkStream $ \_ yld _ _ -> k (`yld` r)
181+consK :: YieldK m a -> Stream m a -> Stream m a
182+consK k r = mkStream $ \_ yld _ _ -> k (\x -> yld x r)
286183
287184-- XXX Build a stream from a repeating callback function.
288185
@@ -305,8 +202,8 @@ infixr 5 `cons`
305202--
306203-- @since 0.1.0
307204{-# INLINE_NORMAL cons #-}
308−cons :: IsStream t => a -> t m a -> t m a
309−cons a r = mkStream $ \_ yld _ _ -> yld a r
205+cons :: a -> Stream m a -> Stream m a
206+cons a r = mkStream $ \_ yield _ _ -> yield a r
310207
311208infixr 5 .:
312209
@@ -319,7 +216,7 @@ infixr 5 .:
319216--
320217-- @since 0.1.1
321218{-# INLINE (.:) #-}
322−(.:) :: IsStream t => a -> t m a -> t m a
219+(.:) :: a -> Stream m a -> Stream m a
323220(.:) = cons
324221
325222-- | An empty stream.
@@ -331,7 +228,7 @@ infixr 5 .:
331228--
332229-- @since 0.1.0
333230{-# INLINE_NORMAL nil #-}
334−nil :: IsStream t => t m a
231+nil :: Stream m a
335232nil = mkStream $ \_ _ _ stp -> stp
336233
337234-- | An empty stream producing a side effect.
@@ -344,22 +241,32 @@ nil = mkStream $ \_ _ _ stp -> stp
344241--
345242-- /Pre-release/
346243{-# INLINE_NORMAL nilM #-}
347−nilM :: (IsStream t, Monad m) => m b -> t m a
244+nilM :: Monad m => m b -> Stream m a
348245nilM m = mkStream $ \_ _ _ stp -> m >> stp
349246
350247{-# INLINE_NORMAL fromPure #-}
351−fromPure :: IsStream t => a -> t m a
248+fromPure :: a -> Stream m a
352249fromPure a = mkStream $ \_ _ single _ -> single a
353250
354251{-# INLINE_NORMAL fromEffect #-}
355−fromEffect :: (Monad m, IsStream t) => m a -> t m a
356−fromEffect m = fromStream $ mkStream $ \_ _ single _ -> m >>= single
252+fromEffect :: Monad m => m a -> Stream m a
253+fromEffect m = mkStream $ \_ _ single _ -> m >>= single
254+
255+infixr 5 `consM`
256+
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.
260+{-# INLINE consM #-}
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
357264
358265-- XXX specialize to IO?
359266{-# INLINE consMBy #-}
360−consMBy :: (IsStream t, MonadAsync m) => (t m a -> t m a -> t m a)
361− -> m a -> t m a -> t m a
362−consMBy f m r = fromStream (fromEffect m) `f` r
267+consMBy :: Monad m =>
268+ (Stream m a -> Stream m a -> Stream m a) -> m a -> Stream m a -> Stream m a
269+consMBy f m r = fromEffect m `f` r
363270
364271------------------------------------------------------------------------------
365272-- Folding a stream
@@ -370,78 +277,27 @@ consMBy f m r = fromStream (fromEffect m) `f` r
370277-- SVar passed via the State.
371278{-# INLINE_EARLY foldStreamShared #-}
372279foldStreamShared
373− :: IsStream t
374− => State Stream m a
375− -> (a -> t m a -> m r)
376− -> (a -> m r)
377− -> m r
378− -> t m a
379− -> 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
384−
385−-- XXX write a similar rule for foldStream as well?
386−{-# RULES "foldStreamShared from stream"
387− foldStreamShared = foldStreamSharedStream #-}
388−foldStreamSharedStream
389280 :: State Stream m a
390281 -> (a -> Stream m a -> m r)
391282 -> (a -> m r)
392283 -> m r
393284 -> Stream m a
394285 -> m r
395−foldStreamSharedStream st yld sng stp m =
396− let MkStream k = toStream m
397− in k st yld sng stp
286+foldStreamShared s yield single stop (MkStream k) = k s yield single stop
398287
399288-- | Fold a stream by providing a State, stop continuation, a singleton
400289-- continuation and a yield continuation. The stream will not use the SVar
401290-- passed via State.
402291{-# INLINE foldStream #-}
403292foldStream
404− :: IsStream t
405− => State Stream m a
406− -> (a -> t m a -> m r)
293+ :: State Stream m a
294+ -> (a -> Stream m a -> m r)
407295 -> (a -> m r)
408296 -> m r
409− -> t m a
297+ -> Stream m a
410298 -> m r
411−foldStream st yld sng stp m =
412− let yieldk a x = yld a (fromStream x)
413− MkStream k = toStream m
414− in k (adaptState st) yieldk sng stp
415−
416−-------------------------------------------------------------------------------
417−-- Instances
418−-------------------------------------------------------------------------------
419−
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
427−
428−-------------------------------------------------------------------------------
429−-- IsStream Stream
430−-------------------------------------------------------------------------------
431−
432−instance IsStream Stream where
433− toStream = id
434− fromStream = id
435−
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
439− consM = consMStream
440−
441− {-# INLINE (|:) #-}
442− {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
443− (|:) :: Monad m => m a -> Stream m a -> Stream m a
444− (|:) = consMStream
299+foldStream s yield single stop (MkStream k) =
300+ k (adaptState s) yield single stop
445301
446302-------------------------------------------------------------------------------
447303-- foldr/build fusion
@@ -454,15 +310,18 @@ instance IsStream Stream where
454310-- reconstruct using a shared state (SVar) or without sharing the state.
455311--
456312{-# INLINE foldrSWith #-}
457−foldrSWith :: IsStream t
458− => (forall r. State Stream m b
459− -> (b -> t m b -> m r)
313+foldrSWith ::
314+ (forall r. State Stream m b
315+ -> (b -> Stream m b -> m r)
460316 -> (b -> m r)
461317 -> m r
462− -> t m b
318+ -> Stream m b
463319 -> m r)
464− -> (a -> t m b -> t m b) -> t m b -> t m a -> t m b
465−foldrSWith f step final = go
320+ -> (a -> Stream m b -> Stream m b)
321+ -> Stream m b
322+ -> Stream m a
323+ -> Stream m b
324+foldrSWith f step final m = go m
466325 where
467326 go m1 = mkStream $ \st yld sng stp ->
468327 let run x = f st yld sng stp x
@@ -480,7 +339,11 @@ foldrSWith f step final = go
480339
481340-- | Fold sharing the SVar state within the reconstructed stream
482341{-# 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)
344+ -> Stream m b
345+ -> Stream m a
346+ -> Stream m b
484347foldrSShared = foldrSWith foldStreamShared
485348
486349-- XXX consM is a typeclass method, therefore rewritten already. Instead maybe
@@ -495,7 +358,11 @@ foldrSShared = foldrSWith foldStreamShared
495358
496359-- | Lazy right associative fold to a stream.
497360{-# 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)
363+ -> Stream m b
364+ -> Stream m a
365+ -> Stream m b
499366foldrS = foldrSWith foldStream
500367
501368{-# RULES "foldrS/id" foldrS cons nil = \x -> x #-}
@@ -512,15 +379,18 @@ foldrS = foldrSWith foldStream
512379-------------------------------------------------------------------------------
513380
514381{-# INLINE foldrSMWith #-}
515−foldrSMWith :: (IsStream t, Monad m)
382+foldrSMWith :: Monad m
516383 => (forall r. State Stream m b
517− -> (b -> t m b -> m r)
384+ -> (b -> Stream m b -> m r)
518385 -> (b -> m r)
519386 -> m r
520− -> t m b
387+ -> Stream m b
521388 -> m r)
522− -> (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
523−foldrSMWith f step final = go
389+ -> (m a -> Stream m b -> Stream m b)
390+ -> Stream m b
391+ -> Stream m a
392+ -> Stream m b
393+foldrSMWith f step final m = go m
524394 where
525395 go m1 = mkStream $ \st yld sng stp ->
526396 let run x = f st yld sng stp x
@@ -530,8 +400,11 @@ foldrSMWith f step final = go
530400 in foldStream (adaptState st) yieldk single stop m1
531401
532402{-# INLINE_NORMAL foldrSM #-}
533−foldrSM :: (IsStream t, Monad m)
534− => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
403+foldrSM :: Monad m
404+ => (m a -> Stream m b -> Stream m b)
405+ -> Stream m b
406+ -> Stream m a
407+ -> Stream m b
535408foldrSM = foldrSMWith foldStream
536409
537410-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
@@ -542,8 +415,11 @@ foldrSM = foldrSMWith foldStream
542415
543416-- Like foldrSM but sharing the SVar state within the recostructed stream.
544417{-# INLINE_NORMAL foldrSMShared #-}
545−foldrSMShared :: (IsStream t, Monad m)
546− => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
418+foldrSMShared :: Monad m
419+ => (m a -> Stream m b -> Stream m b)
420+ -> Stream m b
421+ -> Stream m a
422+ -> Stream m b
547423foldrSMShared = foldrSMWith foldStreamShared
548424
549425-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
@@ -559,7 +435,7 @@ foldrSMShared = foldrSMWith foldStreamShared
559435-------------------------------------------------------------------------------
560436
561437{-# 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
563439build g = g cons nil
564440
565441{-# RULES "foldrM/build"
@@ -584,37 +460,46 @@ build g = g cons nil
584460
585461-- build a stream by applying cons and nil to a build function
586462{-# 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)
465+ -> Stream m a
588466buildS g = g cons nil
589467
590468{-# 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).
592471 foldrS k z (buildS g) = g k z #-}
593472
594473{-# 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).
596476 foldrS k z (x `cons` buildS g) = k x (g k z) #-}
597477
598478{-# 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).
600481 foldrSShared k z (buildS g) = g k z #-}
601482
602483{-# 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).
604486 foldrSShared k z (x `cons` buildS g) = k x (g k z) #-}
605487
606488-- build a stream by applying consM and nil to a build function
607489{-# INLINE_NORMAL buildSM #-}
608−buildSM :: (IsStream t, MonadAsync m)
609− => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a
490+buildSM :: Monad m
491+ => ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
492+ -> Stream m a
610493buildSM g = g consM nil
611494
612495{-# 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).
614498 foldrSM k z (buildSM g) = g k z #-}
615499
616500{-# 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).
618503 foldrSMShared k z (buildSM g) = g k z #-}
619504
620505-- Disabled because this may not fire as consM is a class Op
@@ -629,71 +514,79 @@ buildSM g = g consM nil
629514-- Build using monadic build functions (continuations) instead of
630515-- reconstructing a stream.
631516{-# INLINE_NORMAL buildM #-}
632−buildM :: (IsStream t, MonadAsync m)
633− => (forall r. (a -> t m a -> m r)
517+buildM :: Monad m
518+ => (forall r. (a -> Stream m a -> m r)
634519 -> (a -> m r)
635520 -> m r
636521 -> m r
637522 )
638− -> t m a
523+ -> Stream m a
639524buildM g = mkStream $ \st yld sng stp ->
640525 g (\a r -> foldStream st yld sng stp (return a `consM` r)) sng stp
641526
642527-- | Like 'buildM' but shares the SVar state across computations.
643−{-# INLINE_NORMAL sharedM #-}
644−sharedM :: (IsStream t, MonadAsync m)
645− => (forall r. (a -> t m a -> m r)
528+{-# INLINE_NORMAL sharedMWith #-}
529+sharedMWith :: Monad m
530+ => (m a -> Stream m a -> Stream m a)
531+ -> (forall r. (a -> Stream m a -> m r)
646532 -> (a -> m r)
647533 -> m r
648534 -> m r
649535 )
650− -> t m a
651−sharedM g = mkStream $ \st yld sng stp ->
652− g (\a r -> foldStreamShared st yld sng stp (return a `consM` r)) sng stp
536+ -> Stream m a
537+sharedMWith cns g = mkStream $ \st yld sng stp ->
538+ g (\a r -> foldStreamShared st yld sng stp (return a `cns` r)) sng stp
653539
654540-------------------------------------------------------------------------------
655541-- augment
656542-------------------------------------------------------------------------------
657543
658544{-# INLINE_NORMAL augmentS #-}
659−augmentS :: IsStream t
660− => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
661−augmentS g = g cons
545+augmentS ::
546+ ((a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
547+ -> Stream m a
548+ -> Stream m a
549+augmentS g xs = g cons xs
662550
663551{-# 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).
665553 augmentS g nil = buildS g
666554 #-}
667555
668556{-# 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).
670559 foldrS k z (augmentS g xs) = g k (foldrS k z xs)
671560 #-}
672561
673562{-# RULES "augmentS/buildS"
674− forall (g :: (a -> t m a -> t m a) -> t m a -> t m a)
675− (h :: (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)
564+ (h :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
676565 augmentS g (buildS h) = buildS (\c n -> g c (h c n))
677566 #-}
678567
679568{-# INLINE_NORMAL augmentSM #-}
680−augmentSM :: (IsStream t, MonadAsync m)
681− => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
682−augmentSM g = g consM
569+augmentSM :: Monad m =>
570+ ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)
571+ -> Stream m a -> Stream m a
572+augmentSM g xs = g consM xs
683573
684574{-# 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).
686577 augmentSM g nil = buildSM g
687578 #-}
688579
689580{-# 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).
691583 foldrSM k z (augmentSM g xs) = g k (foldrSM k z xs)
692584 #-}
693585
694586{-# RULES "augmentSM/buildSM"
695− forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
696− (h :: (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)
589+ (h :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).
697590 augmentSM g (buildSM h) = buildSM (\c n -> g c (h c n))
698591 #-}
699592
@@ -703,8 +596,8 @@ augmentSM g = g consM
703596
704597-- | Lazy right fold with a monadic step function.
705598{-# INLINE_NORMAL foldrM #-}
706−foldrM :: IsStream t => (a -> m b -> m b) -> m b -> t m a -> m b
707−foldrM step acc = go
599+foldrM :: (a -> m b -> m b) -> m b -> Stream m a -> m b
600+foldrM step acc m = go m
708601 where
709602 go m1 =
710603 let stop = acc
@@ -715,14 +608,14 @@ foldrM step acc = go
715608{-# INLINE_NORMAL foldrMKWith #-}
716609foldrMKWith
717610 :: (State Stream m a
718− -> (a -> t m a -> m b)
611+ -> (a -> Stream m a -> m b)
719612 -> (a -> m b)
720613 -> m b
721− -> t m a
614+ -> Stream m a
722615 -> m b)
723616 -> (a -> m b -> m b)
724617 -> 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)
726619 -> m b
727620foldrMKWith f step acc = go
728621 where
@@ -742,7 +635,7 @@ foldrMKWith f step acc = go
742635-- XXX in which case will foldrM/buildM fusion be useful?
743636{-# RULES "foldrM/buildM"
744637 forall step acc (g :: (forall r.
745− (a -> t m a -> m r)
638+ (a -> Stream m a -> m r)
746639 -> (a -> m r)
747640 -> m r
748641 -> m r
@@ -750,15 +643,17 @@ foldrMKWith f step acc = go
750643 foldrM step acc (buildM g) = foldrMKWith foldStream step acc g
751644 #-}
752645
646+{-
753647{-# RULES "foldrM/sharedM"
754648 forall step acc (g :: (forall r.
755− (a -> t m a -> m r)
649+ (a -> Stream m a -> m r)
756650 -> (a -> m r)
757651 -> m r
758652 -> m r
759653 )).
760654 foldrM step acc (sharedM g) = foldrMKWith foldStreamShared step acc g
761655 #-}
656+-}
762657
763658------------------------------------------------------------------------------
764659-- Left fold
@@ -771,12 +666,12 @@ foldrMKWith f step acc = go
771666--
772667-- Note that the accumulator is always evaluated including the initial value.
773668{-# INLINE foldlx' #-}
774−foldlx' :: forall t m a b x. (IsStream t, Monad m)
775− => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
669+foldlx' :: forall m a b x. Monad m
670+ => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> m b
776671foldlx' step begin done m = get $ go m begin
777672 where
778673 {-# NOINLINE get #-}
779− get :: t m x -> m b
674+ get :: Stream m x -> m b
780675 get m1 =
781676 -- XXX we are not strictly evaluating the accumulator here. Is this
782677 -- okay?
@@ -788,7 +683,7 @@ foldlx' step begin done m = get $ go m begin
788683 -- however that is more expensive because of unnecessary recursion
789684 -- that cannot be tail call optimized. Unfolding recursion explicitly via
790685 -- continuations is much more efficient.
791− go :: t m a -> x -> t m x
686+ go :: Stream m a -> x -> Stream m x
792687 go m1 !acc = mkStream $ \_ yld sng _ ->
793688 let stop = sng acc
794689 single a = sng $ step acc a
@@ -799,9 +694,44 @@ foldlx' step begin done m = get $ go m begin
799694
800695-- | Strict left associative fold.
801696{-# 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
803698foldl' step begin = foldlx' step begin id
804699
700+------------------------------------------------------------------------------
701+-- Specialized folds
702+------------------------------------------------------------------------------
703+
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 ())
709+
710+-- |
711+-- > drain = foldl' (\_ _ -> ()) ()
712+-- > drain = mapM_ (\_ -> return ())
713+{-# INLINE drain #-}
714+drain :: Monad m => Stream m a -> m ()
715+drain = foldrM (\_ xs -> xs) (return ())
716+{-
717+drain = go
718+ where
719+ go m1 =
720+ let stop = return ()
721+ single _ = return ()
722+ yieldk _ r = go r
723+ in foldStream defState yieldk single stop m1
724+-}
725+
726+{-# INLINE null #-}
727+null :: Monad m => Stream m a -> m Bool
728+-- null = foldrM (\_ _ -> return True) (return False)
729+null m =
730+ let stop = return True
731+ single _ = return False
732+ yieldk _ _ = return False
733+ in foldStream defState yieldk single stop m
734+
805735------------------------------------------------------------------------------
806736-- Semigroup
807737------------------------------------------------------------------------------
@@ -811,19 +741,8 @@ infixr 6 `serial`
811741-- | Appends two streams sequentially, yielding all elements from the first
812742-- stream, and then all elements from the second stream.
813743--
814−-- >>> import Streamly.Prelude (serial)
815−-- >>> stream1 = Stream.fromList [1,2]
816−-- >>> stream2 = Stream.fromList [3,4]
817−-- >>> Stream.toList $ stream1 `serial` stream2
818−-- [1,2,3,4]
819−--
820−-- This operation can be used to fold an infinite lazy container of streams.
821−--
822−-- /Since: 0.2.0 ("Streamly")/
823−--
824−-- @since 0.8.0
825744{-# 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
827746-- XXX This doubles the time of toNullAp benchmark, may not be fusing properly
828747-- serial xs ys = augmentS (\c n -> foldrS c n xs) ys
829748serial m1 m2 = go m1
@@ -836,7 +755,7 @@ serial m1 m2 = go m1
836755
837756-- join/merge/append streams depending on consM
838757{-# 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
840759conjoin xs = augmentSM (\c n -> foldrSM c n xs)
841760
842761instance Semigroup (Stream m a) where
@@ -856,10 +775,13 @@ instance Monoid (Stream m a) where
856775
857776-- Note eta expanded
858777{-# INLINE_LATE mapFB #-}
859−mapFB :: forall (t :: (Type -> Type) -> Type -> Type) b m a.
860− (b -> t m b -> t m b) -> (a -> b) -> a -> t m b -> t m b
861−mapFB c f x ys = c (f x) ys
862−#undef Type
778+mapFB :: forall b m a.
779+ (b -> Stream m b -> Stream m b)
780+ -> (a -> b)
781+ -> a
782+ -> Stream m b
783+ -> Stream m b
784+mapFB c f = \x ys -> c (f x) ys
863785
864786{-# RULES
865787"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
867789 #-}
868790
869791{-# INLINE map #-}
870−map :: IsStream t => (a -> b) -> t m a -> t m b
792+map :: (a -> b) -> Stream m a -> Stream m b
871793map f xs = buildS (\c n -> foldrS (mapFB c f) n xs)
872794
873795-- XXX This definition might potentially be more efficient, but the cost in the
@@ -898,21 +820,6 @@ mapMFB c f x = c (x >>= f)
898820"mapMFB/return" forall c. mapMFB c (\x -> return x) = c
899821-}
900822
901−-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
902−-- other stream types to overload it.
903−{-# INLINE mapM #-}
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.
907−{-
908−mapM f m = go m
909− where
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
914− -}
915−
916823-- This is experimental serial version supporting fusion.
917824--
918825-- XXX what if we do not want to fuse two concurrent mapMs?
@@ -921,9 +828,27 @@ mapM f m = go m
921828-- XXX fusion would be easier for monomoprhic stream types.
922829-- {-# RULES "mapM serial" mapM = mapMSerial #-}
923830{-# 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
925832mapMSerial f xs = buildSM (\c n -> foldrSMShared (mapMFB c f) n xs)
926833
834+{-# INLINE mapMWith #-}
835+mapMWith ::
836+ (m b -> Stream m b -> Stream m b)
837+ -> (a -> m b)
838+ -> Stream m a
839+ -> Stream m b
840+mapMWith cns f = foldrSShared (\x xs -> f x `cns` xs) nil
841+
842+{-
843+-- See note under map definition above.
844+mapMWith cns f = go
845+ where
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
850+-}
851+
927852-- XXX in fact use the Stream type everywhere and only use polymorphism in the
928853-- high level modules/prelude.
929854instance Monad m => Functor (Stream m) where
@@ -943,18 +868,17 @@ instance MonadTrans Stream where
943868
944869-- | Detach a stream from an SVar
945870{-# INLINE unShare #-}
946−unShare :: IsStream t => t m a -> t m a
871+unShare :: Stream m a -> Stream m a
947872unShare x = mkStream $ \st yld sng stp ->
948873 foldStream st yld sng stp x
949874
950875-- XXX the function stream and value stream can run in parallel
951876{-# INLINE apWith #-}
952−apWith
953− :: IsStream t
954− => (t m b -> t m b -> t m b)
955− -> t m (a -> b)
956− -> t m a
957− -> t m b
877+apWith ::
878+ (Stream m b -> Stream m b -> Stream m b)
879+ -> Stream m (a -> b)
880+ -> Stream m a
881+ -> Stream m b
958882apWith par fstream stream = go1 fstream
959883
960884 where
@@ -973,11 +897,10 @@ apWith par fstream stream = go1 fstream
973897 in foldStream (adaptState st) yieldk single stp m
974898
975899{-# INLINE apSerial #-}
976−apSerial
977− :: IsStream t
978− => t m (a -> b)
979− -> t m a
980− -> t m b
900+apSerial ::
901+ Stream m (a -> b)
902+ -> Stream m a
903+ -> Stream m b
981904apSerial fstream stream = go1 fstream
982905
983906 where
@@ -1004,11 +927,10 @@ apSerial fstream stream = go1 fstream
1004927 in foldStream (adaptState st) yieldk single stp m
1005928
1006929{-# INLINE apSerialDiscardFst #-}
1007−apSerialDiscardFst
1008− :: IsStream t
1009− => t m a
1010− -> t m b
1011− -> t m b
930+apSerialDiscardFst ::
931+ Stream m a
932+ -> Stream m b
933+ -> Stream m b
1012934apSerialDiscardFst fstream stream = go1 fstream
1013935
1014936 where
@@ -1029,11 +951,10 @@ apSerialDiscardFst fstream stream = go1 fstream
1029951 in foldStream st yieldk single stop m
1030952
1031953{-# INLINE apSerialDiscardSnd #-}
1032−apSerialDiscardSnd
1033− :: IsStream t
1034− => t m a
1035− -> t m b
1036− -> t m a
954+apSerialDiscardSnd ::
955+ Stream m a
956+ -> Stream m b
957+ -> Stream m a
1037958apSerialDiscardSnd fstream stream = go1 fstream
1038959
1039960 where
@@ -1059,17 +980,16 @@ apSerialDiscardSnd fstream stream = go1 fstream
1059980 yieldk _ r = yld f (go3 f r)
1060981 in foldStream (adaptState st) yieldk single stp m
1061982
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
1063984-- instead of using a concatMap style definition because the bind
1064985-- implementation in Async and WAsync streams show significant perf degradation
1065986-- if the argument order is changed.
1066987{-# INLINE bindWith #-}
1067−bindWith
1068− :: IsStream t
1069− => (t m b -> t m b -> t m b)
1070− -> t m a
1071− -> (a -> t m b)
1072− -> t m b
988+bindWith ::
989+ (Stream m b -> Stream m b -> Stream m b)
990+ -> Stream m a
991+ -> (a -> Stream m b)
992+ -> Stream m b
1073993bindWith par m1 f = go m1
1074994 where
1075995 go m =
@@ -1094,21 +1014,18 @@ bindWith par m1 f = go m1
10941014-- function.
10951015--
10961016-- @since 0.7.0
1097−{-# INLINE concatMapBy #-}
1098−concatMapBy
1099− :: IsStream t
1100− => (t m b -> t m b -> t m b)
1101− -> (a -> t m b)
1102− -> t m a
1103− -> t m b
1104−concatMapBy par f xs = bindWith par xs f
1017+{-# INLINE concatMapWith #-}
1018+concatMapWith
1019+ ::
1020+ (Stream m b -> Stream m b -> Stream m b)
1021+ -> (a -> Stream m b)
1022+ -> Stream m a
1023+ -> Stream m b
1024+concatMapWith par f xs = bindWith par xs f
11051025
11061026{-# INLINE concatMap #-}
1107−concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
1108−concatMap f m = fromStream $
1109− concatMapBy serial
1110− (adapt . toStream . f)
1111− (adapt $ toStream m)
1027+concatMap :: (a -> Stream m b) -> Stream m a -> Stream m b
1028+concatMap = concatMapWith serial
11121029
11131030{-
11141031-- Fused version.
@@ -1133,11 +1050,11 @@ concatMap_ f xs = buildS
11331050--
11341051{-# INLINE concatPairsWith #-}
11351052concatPairsWith
1136− :: IsStream t
1137− => (t m b -> t m b -> t m b)
1138− -> (a -> t m b)
1139− -> t m a
1140− -> t m b
1053+ ::
1054+ (Stream m b -> Stream m b -> Stream m b)
1055+ -> (a -> Stream m b)
1056+ -> Stream m a
1057+ -> Stream m b
11411058concatPairsWith combine f = go Nothing
11421059
11431060 where
@@ -1195,3 +1112,194 @@ concatUnfoldr :: IsStream t
11951112 => (b -> t m (Maybe (a, b))) -> t m b -> t m a
11961113concatUnfoldr = undefined
11971114-}
1115+
1116+------------------------------------------------------------------------------
1117+-- MonadReader
1118+------------------------------------------------------------------------------
1119+
1120+{-# INLINABLE withLocal #-}
1121+withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1122+withLocal f m =
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
1127+
1128+-------------------------------------------------------------------------------
1129+-- Generation
1130+-------------------------------------------------------------------------------
1131+
1132+{-# INLINE unfoldr #-}
1133+unfoldr :: (b -> Maybe (a, b)) -> b -> Stream m a
1134+unfoldr next s0 = build $ \yld stp ->
1135+ let go s =
1136+ case next s of
1137+ Just (a, b) -> yld a (go b)
1138+ Nothing -> stp
1139+ in go s0
1140+
1141+{-# INLINE unfoldrMWith #-}
1142+unfoldrMWith :: Monad m =>
1143+ (m a -> Stream m a -> Stream m a)
1144+ -> (b -> m (Maybe (a, b)))
1145+ -> b
1146+ -> Stream m a
1147+unfoldrMWith cns step = go
1148+
1149+ where
1150+
1151+ go s = sharedMWith cns $ \yld _ stp -> do
1152+ r <- step s
1153+ case r of
1154+ Just (a, b) -> yld a (go b)
1155+ Nothing -> stp
1156+
1157+-- | Generate an infinite stream by repeating a pure value.
1158+--
1159+-- /Pre-release/
1160+{-# INLINE repeat #-}
1161+repeat :: a -> Stream m a
1162+repeat a = let x = cons a x in x
1163+
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.
1167+--
1168+-- /Pre-release/
1169+repeatMWith :: (m a -> t m a -> t m a) -> m a -> t m a
1170+repeatMWith cns = go
1171+
1172+ where
1173+
1174+ go m = m `cns` go m
1175+
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
1179+
1180+ where
1181+
1182+ go cnt = if cnt <= 0 then nil else m `cns` go (cnt - 1)
1183+
1184+{-# INLINE fromIndicesMWith #-}
1185+fromIndicesMWith ::
1186+ (m a -> Stream m a -> Stream m a) -> (Int -> m a) -> Stream m a
1187+fromIndicesMWith cns gen = go 0
1188+
1189+ where
1190+
1191+ go i = mkStream $ \st stp sng yld -> do
1192+ foldStreamShared st stp sng yld (gen i `cns` go (i + 1))
1193+
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
1198+
1199+ where
1200+
1201+ go s = mkStream $ \st stp sng yld -> do
1202+ !next <- s
1203+ foldStreamShared st stp sng yld (return next `cns` go (step next))
1204+
1205+{-# INLINE headPartial #-}
1206+headPartial :: Monad m => Stream m a -> m a
1207+headPartial = foldrM (\x _ -> return x) (error "head of nil")
1208+
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"
1213+ single _ = stp
1214+ yieldk _ r = foldStream st yld sng stp r
1215+ in foldStream st yieldk single stop m
1216+
1217+{-# INLINE mfix #-}
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
1223+
1224+ where
1225+
1226+ -- fix the head element of the stream
1227+ xs = fix (f . headPartial)
1228+
1229+ -- now fix the tail recursively
1230+ ys = mfix (tailPartial . f)
1231+
1232+-------------------------------------------------------------------------------
1233+-- Conversions
1234+-------------------------------------------------------------------------------
1235+
1236+-- |
1237+-- @
1238+-- fromFoldable = 'Prelude.foldr' 'cons' 'nil'
1239+-- @
1240+--
1241+-- Construct a stream from a 'Foldable' containing pure values:
1242+--
1243+-- @since 0.2.0
1244+{-# INLINE fromFoldable #-}
1245+fromFoldable :: Foldable f => f a -> Stream m a
1246+fromFoldable = Prelude.foldr cons nil
1247+
1248+{-# INLINE fromFoldableM #-}
1249+fromFoldableM :: (Foldable f, Monad m) => f (m a) -> Stream m a
1250+fromFoldableM = Prelude.foldr consM nil
1251+
1252+-------------------------------------------------------------------------------
1253+-- Deconstruction
1254+-------------------------------------------------------------------------------
1255+
1256+{-# INLINE uncons #-}
1257+uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
1258+uncons m =
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
1263+
1264+{-# INLINE tail #-}
1265+tail :: Monad m => Stream m a -> m (Maybe (Stream m a))
1266+tail =
1267+ let stop = return Nothing
1268+ single _ = return $ Just nil
1269+ yieldk _ r = return $ Just r
1270+ in foldStream defState yieldk single stop
1271+
1272+{-# INLINE init #-}
1273+init :: Monad m => Stream m a -> m (Maybe (Stream m a))
1274+init = go1
1275+ where
1276+ go1 m1 = do
1277+ r <- uncons m1
1278+ case r of
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
1285+
1286+------------------------------------------------------------------------------
1287+-- Reordering
1288+------------------------------------------------------------------------------
1289+
1290+-- | Lazy left fold to a stream.
1291+{-# INLINE foldlS #-}
1292+foldlS ::
1293+ (Stream m b -> a -> Stream m b) -> Stream m b -> Stream m a -> Stream m b
1294+foldlS step = go
1295+ where
1296+ go acc rest = mkStream $ \st yld sng stp ->
1297+ let run x = foldStream st yld sng stp x
1298+ stop = run acc
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
1302+
1303+{-# INLINE reverse #-}
1304+reverse :: Stream m a -> Stream m a
1305+reverse = foldlS (flip cons) nil