@@ -6,8 +6,11 @@
6{-# LANGUAGE InstanceSigs #-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
8{-# LANGUAGE RankNTypes #-}
9+{-# LANGUAGE ScopedTypeVariables #-}
9{-# LANGUAGE UndecidableInstances #-} -- XXX
10{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- Module : Streamly.Streams.StreamK
15-- Module : Streamly.Streams.StreamK
13-- Copyright : (c) 2017 Harendra Kumar
16-- Copyright : (c) 2017 Harendra Kumar
@@ -32,22 +35,21 @@ module Streamly.Streams.StreamK
34 -- * The stream type
37 -- * The stream type
35− , Stream (..)
38+ , Stream
41− -- * Construction
40+ -- * Construction Primitives
42 , mkStream
41 , mkStream
47− -- * Asynchronous construction
46+ -- * Elimination Primitives
48− , nilK
47+ , foldStream
49− , yieldK
48+ , foldStreamShared
50− , consK
49+ , foldStreamSVar
51+ -- * Transformation Primitives
52 -- * Deconstruction
54 -- * Deconstruction
@@ -71,7 +73,6 @@ module Streamly.Streams.StreamK
72 -- * Elimination
74 -- * Elimination
73 -- ** General Folds
75 -- ** General Folds
@@ -150,7 +151,7 @@ module Streamly.Streams.StreamK
152 -- * Utilities
153 -- * Utilities
153− , consMSerial
154+ , consMStream
154 , bindWith
155 , bindWith
155 , withLocal
156 , withLocal
@@ -161,10 +162,7 @@ module Streamly.Streams.StreamK
163import Control.Monad (void)
164import Control.Monad (void)
164−import Control.Monad.IO.Class (MonadIO(liftIO))
165import Control.Monad.Reader.Class (MonadReader(..))
165import Control.Monad.Reader.Class (MonadReader(..))
166−import Control.Monad.Trans.Class (MonadTrans(lift))
167−import Data.Semigroup (Semigroup(..))
168import Prelude
166import Prelude
169 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
167 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
170 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
168 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
@@ -173,182 +171,18 @@ import Prelude
173import qualified Prelude
171import qualified Prelude
175import Streamly.SVar
173import Streamly.SVar
174+import Streamly.Streams.StreamK.Type
177−------------------------------------------------------------------------------
178−-- The basic stream type
179−------------------------------------------------------------------------------
181−-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
182−-- constructed using actions in monad 'm'. It uses stop, singleton and yield
183−-- continuations equivalent to the following direct style type:
186−-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
189−-- To facilitate parallel composition we maintain a local state in an 'SVar'
190−-- that is shared across and is used for synchronization of the streams being
193−-- The singleton case can be expressed in terms of stop and yield but we have
194−-- it as a separate case to optimize composition operations for streams with
195−-- single element. We build singleton streams in the implementation of 'pure'
196−-- for Applicative and Monad, and in 'lift' for MonadTrans.
200− unStream :: forall r.
201− State Stream m a -- state
203− -> (a -> m r) -- singleton
204− -> (a -> Stream m a -> m r) -- yield
208−-- XXX make this the default "unStream"
209−-- | unwraps the Stream type producing the stream function that can be run with
211−{-# INLINE unStreamIsolated #-}
214− -> State Stream m a -- state
216− -> (a -> m r) -- singleton
217− -> (a -> Stream m a -> m r) -- yield
219−unStreamIsolated x st = unStream x (rstState st)
221−{-# INLINE isolateStream #-}
222−isolateStream :: Stream m a -> Stream m a
223−isolateStream x = Stream $ \st stp sng yld ->
224− unStreamIsolated x st stp sng yld
226−-- | Like unstream, but passes a shared SVar across continuations.
227−{-# INLINE unstreamShared #-}
230− -> State Stream m a -- state
232− -> (a -> m r) -- singleton
233− -> (a -> Stream m a -> m r) -- yield
235−unstreamShared = unStream
237−-- Run the stream using a run function associated with the SVar that runs the
238−-- streams with a captured snapshot of the monadic state.
239−{-# INLINE runStreamSVar #-}
244− -> State Stream m a -- state
246− -> (a -> m r) -- singleton
247− -> (a -> Stream m a -> m r) -- yield
249−runStreamSVar sv m st stp sng yld =
250− let mrun = runInIO $ svarMrun sv
251− in void $ liftIO $ mrun $ unStream m st stp sng yld
253−------------------------------------------------------------------------------
254−-- Types that can behave as a Stream
255−------------------------------------------------------------------------------
260−-- | Class of types that can represent a stream of elements of some type 'a' in
264−class IsStream t where
265− toStream :: t m a -> Stream m a
266− fromStream :: Stream m a -> t m a
267− -- | Constructs a stream by adding a monadic action at the head of an
268− -- existing stream. For example:
271− -- > toList $ getLine \`consM` getLine \`consM` nil
274− -- ["hello","world"]
277− -- /Concurrent (do not use 'parallely' to construct infinite streams)/
280− consM :: MonadAsync m => m a -> t m a -> t m a
281− -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
282− -- to remember that @|@ comes before ':'.
285− -- > toList $ getLine |: getLine |: nil
288− -- ["hello","world"]
292− -- let delay = threadDelay 1000000 >> print 1
293− -- runStream $ serially $ delay |: delay |: delay |: nil
294− -- runStream $ parallely $ delay |: delay |: delay |: nil
297− -- /Concurrent (do not use 'parallely' to construct infinite streams)/
300− (|:) :: MonadAsync m => m a -> t m a -> t m a
301− -- We can define (|:) just as 'consM' but it is defined explicitly for each
302− -- type because we want to use SPECIALIZE pragma on the definition.
304−-- | Same as 'IsStream'.
307−{-# DEPRECATED Streaming "Please use IsStream instead." #-}
308−type Streaming = IsStream
310−-------------------------------------------------------------------------------
311−-- Type adapting combinators
312−-------------------------------------------------------------------------------
314−-- | Adapt any specific stream type to any other specific stream type.
317−adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
318−adapt = fromStream . toStream
320−------------------------------------------------------------------------------
322−------------------------------------------------------------------------------
324−-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
176+-- | Detach a stream from an SVar
325−-- continuation and a yield continuation.
177+{-# INLINE unShare #-}
326−mkStream:: IsStream t
178+unShare :: IsStream t => t m a -> t m a
327− => (forall r. State Stream m a
179+unShare x = mkStream $ \st yld sng stp ->
328− -> m r
180+ foldStream st yld sng stp x
330− -> (a -> t m a -> m r)
333−mkStream k = fromStream $ Stream $ \st stp sng yld ->
334− let yieldk a r = yld a (toStream r)
335− in k (rstState st) stp sng yieldk
337------------------------------------------------------------------------------
182------------------------------------------------------------------------------
338-- Construction
183-- Construction
339------------------------------------------------------------------------------
184------------------------------------------------------------------------------
341−-- | An empty stream.
349−nil :: IsStream t => t m a
350−nil = fromStream $ Stream $ \_ stp _ _ -> stp
352infixr 5 `cons`
186infixr 5 `cons`
354-- faster than consM because there is no bind.
188-- faster than consM because there is no bind.
@@ -363,8 +197,9 @@ infixr 5 `cons`
365-- @since 0.1.0
199-- @since 0.1.0
366cons :: IsStream t => a -> t m a -> t m a
201cons :: IsStream t => a -> t m a -> t m a
367−cons a r = fromStream $ Stream $ \_ _ _ yld -> yld a (toStream r)
202+cons a r = mkStream $ \_ yld _ _ -> yld a r
369infixr 5 .:
204infixr 5 .:
@@ -376,50 +211,10 @@ infixr 5 .:
378-- @since 0.1.1
213-- @since 0.1.1
379(.:) :: IsStream t => a -> t m a -> t m a
215(.:) :: IsStream t => a -> t m a -> t m a
380(.:) = cons
216(.:) = cons
382−{-# INLINE consMSerial #-}
383−consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a
384−consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
386−------------------------------------------------------------------------------
387−-- Asynchronous construction
388−------------------------------------------------------------------------------
390−-- | Make an empty stream from a callback function.
391−nilK :: IsStream t => (forall r. m r -> m r) -> t m a
392−nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp
394−-- | Make a singleton stream from a one shot callback function.
395−yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
396−yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng
398−-- | Construct a stream from a callback function.
399−consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a
400−consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))
402−-- XXX consK with concurrent callbacks
403−-- XXX Build a stream from a repeating callback function.
405−-------------------------------------------------------------------------------
407−-------------------------------------------------------------------------------
409−instance IsStream Stream where
413− {-# INLINE consM #-}
414− {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
415− consM :: Monad m => m a -> Stream m a -> Stream m a
419− {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
420− (|:) :: Monad m => m a -> Stream m a -> Stream m a
423-------------------------------------------------------------------------------
218-------------------------------------------------------------------------------
424-- Deconstruction
219-- Deconstruction
425-------------------------------------------------------------------------------
220-------------------------------------------------------------------------------
@@ -429,43 +224,56 @@ uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
429uncons m =
224uncons m =
430 let stop = return Nothing
225 let stop = return Nothing
431 single a = return (Just (a, nil))
226 single a = return (Just (a, nil))
432− yieldk a r = return (Just (a, fromStream r))
227+ yieldk a r = return (Just (a, r))
433− in unStream (toStream m) defState stop single yieldk
228+ in foldStream defState yieldk single stop m
435-------------------------------------------------------------------------------
230-------------------------------------------------------------------------------
436-- Generation
231-- Generation
437-------------------------------------------------------------------------------
232-------------------------------------------------------------------------------
234+{-# INLINE_NORMAL build #-}
235+build :: IsStream t => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a
238+{-# INLINE_NORMAL _augment #-}
241+ => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a -> t m a
242+_augment g xs = g cons xs
244+{-# INLINE_NORMAL _buildM #-}
246+ :: (IsStream t, MonadAsync m)
247+ => forall a. ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a
248+_buildM g = g consM nil
439{-# INLINE unfoldr #-}
250{-# INLINE unfoldr #-}
440unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
251unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
441−unfoldr step = fromStream . go
252+unfoldr step b0 = build $ \cns nl ->
442− where
253+ let go s =
443− go s = Stream $ \_ stp _ yld ->
444− case step s of
254+ case step s of
446− Just (a, b) -> yld a (go b)
255+ Just (a, b) -> a `cns` go b
448{-# INLINE unfoldrM #-}
259{-# INLINE unfoldrM #-}
449unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
260unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
450unfoldrM step = go
261unfoldrM step = go
452− go s = fromStream $ Stream $ \svr stp sng yld -> do
263+ go s = mkStream $ \st yld sng stp -> do
453 mayb <- step s
264 mayb <- step s
454 case mayb of
265 case mayb of
455 Nothing -> stp
266 Nothing -> stp
456 Just (a, b) ->
267 Just (a, b) ->
457− unStream (toStream (return a |: go b)) svr stp sng yld
268+ foldStreamShared st yld sng stp $ return a |: go b
459-------------------------------------------------------------------------------
270-------------------------------------------------------------------------------
460-- Special generation
271-- Special generation
461-------------------------------------------------------------------------------
272-------------------------------------------------------------------------------
463yield :: IsStream t => a -> t m a
275yield :: IsStream t => a -> t m a
464−yield a = fromStream $ Stream $ \_ _ single _ -> single a
276+yield a = mkStream $ \_ _ single _ -> single a
466−{-# INLINE yieldM #-}
467−yieldM :: (Monad m, IsStream t) => m a -> t m a
468−yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single
470-- | Same as yieldM
278-- | Same as yieldM
@@ -484,14 +292,17 @@ once = yieldM
484-- Generate an infinite stream by repeating a pure value.
292-- Generate an infinite stream by repeating a pure value.
486-- @since 0.4.0
294-- @since 0.4.0
295+{-# INLINE repeat #-}
487repeat :: IsStream t => a -> t m a
296repeat :: IsStream t => a -> t m a
488repeat a = let x = cons a x in x
297repeat a = let x = cons a x in x
299+{-# INLINE replicateM #-}
490replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
300replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
491replicateM n m = go n
301replicateM n m = go n
493 go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
303 go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
305+{-# INLINE replicate #-}
495replicate :: IsStream t => Int -> a -> t m a
306replicate :: IsStream t => Int -> a -> t m a
496replicate n a = go n
307replicate n a = go n
@@ -518,47 +329,34 @@ fromList :: IsStream t => [a] -> t m a
518fromList = fromFoldable
329fromList = fromFoldable
520{-# INLINE fromStreamK #-}
331{-# INLINE fromStreamK #-}
521−fromStreamK :: Stream m a -> Stream m a
332+fromStreamK :: IsStream t => Stream m a -> t m a
522−fromStreamK = id
333+fromStreamK = fromStream
524-------------------------------------------------------------------------------
335-------------------------------------------------------------------------------
525-- Elimination by Folding
336-- Elimination by Folding
526-------------------------------------------------------------------------------
337-------------------------------------------------------------------------------
528−-- | Fold a stream by providing an SVar, a stop continuation, a singleton
529−-- continuation and a yield continuation.
535− -> (a -> t m a -> m r)
538−foldStream st blank single step m =
539− let yieldk a x = step a (fromStream x)
540− in unStream (toStream m) st blank single yieldk
542-- | Lazy right associative fold.
339-- | Lazy right associative fold.
543foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
341foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
544−foldr step acc m = go (toStream m)
342+foldr step acc m = go m
547 let stop = return acc
345 let stop = return acc
548 single a = return (step a acc)
346 single a = return (step a acc)
549 yieldk a r = go r >>= \b -> return (step a b)
347 yieldk a r = go r >>= \b -> return (step a b)
550− in unStream m1 defState stop single yieldk
348+ in foldStream defState yieldk single stop m1
552-- | Lazy right fold with a monadic step function.
350-- | Lazy right fold with a monadic step function.
553{-# INLINE foldrM #-}
351{-# INLINE foldrM #-}
554foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
352foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
555−foldrM step acc m = go (toStream m)
353+foldrM step acc m = go m
558 let stop = return acc
356 let stop = return acc
559 single a = step a acc
357 single a = step a acc
560 yieldk a r = go r >>= step a
358 yieldk a r = go r >>= step a
561− in unStream m1 defState stop single yieldk
359+ in foldStream defState yieldk single stop m1
563{-# INLINE foldr1 #-}
361{-# INLINE foldr1 #-}
564foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
362foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
@@ -566,13 +364,13 @@ foldr1 step m = do
566 r <- uncons m
364 r <- uncons m
567 case r of
365 case r of
568 Nothing -> return Nothing
366 Nothing -> return Nothing
569− Just (h, t) -> fmap Just (go h (toStream t))
367+ Just (h, t) -> fmap Just (go h t)
571 go p m1 =
369 go p m1 =
572 let stp = return p
370 let stp = return p
573 single a = return $ step a p
371 single a = return $ step a p
574 yieldk a r = fmap (step p) (go a r)
372 yieldk a r = fmap (step p) (go a r)
575− in unStream m1 defState stp single yieldk
373+ in foldStream defState yieldk single stp m1
577-- | Strict left fold with an extraction function. Like the standard strict
375-- | Strict left fold with an extraction function. Like the standard strict
578-- left fold, but applies a user supplied extraction function (the third
376-- left fold, but applies a user supplied extraction function (the third
@@ -581,28 +379,31 @@ foldr1 step m = do
582-- Note that the accumulator is always evaluated including the initial value.
380-- Note that the accumulator is always evaluated including the initial value.
583{-# INLINE foldx #-}
381{-# INLINE foldx #-}
584−foldx :: (IsStream t, Monad m)
382+foldx :: forall t m a b x. (IsStream t, Monad m)
585 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
383 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
586−foldx step begin done m = get $ go (toStream m) begin
384+foldx step begin done m = get $ go m begin
588 {-# NOINLINE get #-}
386 {-# NOINLINE get #-}
590 -- XXX we are not strictly evaluating the accumulator here. Is this
389 -- XXX we are not strictly evaluating the accumulator here. Is this
592 let single = return . done
391 let single = return . done
392+ -- XXX this is foldSingleton. why foldStreamShared?
593− in unStream m1 undefined undefined single undefined
393+ in foldStreamShared undefined undefined single undefined m1
595 -- Note, this can be implemented by making a recursive call to "go",
395 -- Note, this can be implemented by making a recursive call to "go",
596 -- however that is more expensive because of unnecessary recursion
396 -- however that is more expensive because of unnecessary recursion
597 -- that cannot be tail call optimized. Unfolding recursion explicitly via
397 -- that cannot be tail call optimized. Unfolding recursion explicitly via
598 -- continuations is much more efficient.
398 -- continuations is much more efficient.
399+ go :: t m a -> x -> t m x
599− go m1 !acc = Stream $ \_ _ sng yld ->
400+ go m1 !acc = mkStream $ \_ yld sng _ ->
600 let stop = sng acc
401 let stop = sng acc
601 single a = sng $ step acc a
402 single a = sng $ step acc a
602− yieldk a r =
403+ -- XXX this is foldNonEmptyStream
404+ yieldk a r = foldStream defState yld sng undefined $
603− let stream = go r (step acc a)
405+ go r (step acc a)
604− in unStream stream defState undefined sng yld
605− in unStream m1 defState stop single yieldk
406+ in foldStream defState yieldk single stop m1
607-- | Strict left associative fold.
408-- | Strict left associative fold.
608{-# INLINE foldl' #-}
409{-# INLINE foldl' #-}
@@ -611,17 +412,19 @@ foldl' step begin = foldx step begin id
612-- XXX replace the recursive "go" with explicit continuations.
413-- XXX replace the recursive "go" with explicit continuations.
613-- | Like 'foldx', but with a monadic step function.
414-- | Like 'foldx', but with a monadic step function.
415+{-# INLINABLE foldxM #-}
614foldxM :: (IsStream t, Monad m)
416foldxM :: (IsStream t, Monad m)
615 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
417 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
616−foldxM step begin done m = go begin (toStream m)
418+foldxM step begin done m = go begin m
618 go !acc m1 =
420 go !acc m1 =
619 let stop = acc >>= done
421 let stop = acc >>= done
620 single a = acc >>= \b -> step b a >>= done
422 single a = acc >>= \b -> step b a >>= done
621 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
423 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
622− in unStream m1 defState stop single yieldk
424+ in foldStream defState yieldk single stop m1
624-- | Like 'foldl'' but with a monadic step function.
426-- | Like 'foldl'' but with a monadic step function.
427+{-# INLINE foldlM' #-}
625foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
428foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
626foldlM' step begin = foldxM step (return begin) return
429foldlM' step begin = foldxM step (return begin) return
@@ -629,15 +432,18 @@ foldlM' step begin = foldxM step (return begin) return
629-- Specialized folds
432-- Specialized folds
630------------------------------------------------------------------------------
433------------------------------------------------------------------------------
436+-- > runStream = foldl' (\_ _ -> ()) ()
437+-- > runStream = mapM_ (\_ -> return ())
632{-# INLINE runStream #-}
438{-# INLINE runStream #-}
633runStream :: (Monad m, IsStream t) => t m a -> m ()
439runStream :: (Monad m, IsStream t) => t m a -> m ()
634−runStream m = go (toStream m)
440+runStream = go
637 let stop = return ()
443 let stop = return ()
638 single _ = return ()
444 single _ = return ()
639− yieldk _ r = go (toStream r)
445+ yieldk _ r = go r
640− in unStream m1 defState stop single yieldk
446+ in foldStream defState yieldk single stop m1
642{-# INLINE null #-}
448{-# INLINE null #-}
643null :: (IsStream t, Monad m) => t m a -> m Bool
449null :: (IsStream t, Monad m) => t m a -> m Bool
@@ -645,7 +451,7 @@ null m =
645 let stop = return True
451 let stop = return True
646 single _ = return False
452 single _ = return False
647 yieldk _ _ = return False
453 yieldk _ _ = return False
648− in unStream (toStream m) defState stop single yieldk
454+ in foldStream defState yieldk single stop m
650{-# INLINE head #-}
456{-# INLINE head #-}
651head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
457head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
@@ -653,69 +459,71 @@ head m =
653 let stop = return Nothing
459 let stop = return Nothing
654 single a = return (Just a)
460 single a = return (Just a)
655 yieldk a _ = return (Just a)
461 yieldk a _ = return (Just a)
656− in unStream (toStream m) defState stop single yieldk
462+ in foldStream defState yieldk single stop m
658{-# INLINE tail #-}
464{-# INLINE tail #-}
659tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
465tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
661 let stop = return Nothing
467 let stop = return Nothing
662 single _ = return $ Just nil
468 single _ = return $ Just nil
663− yieldk _ r = return $ Just $ fromStream r
469+ yieldk _ r = return $ Just r
664− in unStream (toStream m) defState stop single yieldk
470+ in foldStream defState yieldk single stop m
666{-# INLINE init #-}
472{-# INLINE init #-}
667init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
473init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
668−init m = go1 (toStream m)
474+init m = go1 m
670 go1 m1 = do
476 go1 m1 = do
671 r <- uncons m1
477 r <- uncons m1
672 case r of
478 case r of
673 Nothing -> return Nothing
479 Nothing -> return Nothing
674− Just (h, t) -> return . Just . fromStream $ go h t
480+ Just (h, t) -> return . Just $ go h t
675− go p m1 = Stream $ \_ stp sng yld ->
481+ go p m1 = mkStream $ \_ yld sng stp ->
676 let single _ = sng p
482 let single _ = sng p
677 yieldk a x = yld p $ go a x
483 yieldk a x = yld p $ go a x
678− in unStream m1 defState stp single yieldk
484+ in foldStream defState yieldk single stp m1
680{-# INLINE elem #-}
486{-# INLINE elem #-}
681elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
487elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
682−elem e m = go (toStream m)
488+elem e m = go m
685 let stop = return False
491 let stop = return False
686 single a = return (a == e)
492 single a = return (a == e)
687 yieldk a r = if a == e then return True else go r
493 yieldk a r = if a == e then return True else go r
688− in unStream m1 defState stop single yieldk
494+ in foldStream defState yieldk single stop m1
690{-# INLINE notElem #-}
496{-# INLINE notElem #-}
691notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
497notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
692−notElem e m = go (toStream m)
498+notElem e m = go m
695 let stop = return True
501 let stop = return True
696 single a = return (a /= e)
502 single a = return (a /= e)
697 yieldk a r = if a == e then return False else go r
503 yieldk a r = if a == e then return False else go r
698− in unStream m1 defState stop single yieldk
504+ in foldStream defState yieldk single stop m1
506+{-# INLINABLE all #-}
700all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
507all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
701−all p m = go (toStream m)
508+all p m = go m
704 let single a | p a = return True
511 let single a | p a = return True
705 | otherwise = return False
512 | otherwise = return False
706 yieldk a r | p a = go r
513 yieldk a r | p a = go r
707 | otherwise = return False
514 | otherwise = return False
708− in unStream m1 defState (return True) single yieldk
515+ in foldStream defState yieldk single (return True) m1
517+{-# INLINABLE any #-}
710any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
518any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
711−any p m = go (toStream m)
519+any p m = go m
714 let single a | p a = return True
522 let single a | p a = return True
715 | otherwise = return False
523 | otherwise = return False
716 yieldk a r | p a = return True
524 yieldk a r | p a = return True
717 | otherwise = go r
525 | otherwise = go r
718− in unStream m1 defState (return False) single yieldk
526+ in foldStream defState yieldk single (return False) m1
720-- | Extract the last element of the stream, if any.
528-- | Extract the last element of the stream, if any.
721{-# INLINE last #-}
529{-# INLINE last #-}
@@ -724,13 +532,13 @@ last = foldx (\_ y -> Just y) Nothing id
725{-# INLINE minimum #-}
533{-# INLINE minimum #-}
726minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
534minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
727−minimum m = go Nothing (toStream m)
535+minimum m = go Nothing m
729 go Nothing m1 =
537 go Nothing m1 =
730 let stop = return Nothing
538 let stop = return Nothing
731 single a = return (Just a)
539 single a = return (Just a)
732 yieldk a r = go (Just a) r
540 yieldk a r = go (Just a) r
733− in unStream m1 defState stop single yieldk
541+ in foldStream defState yieldk single stop m1
735 go (Just res) m1 =
543 go (Just res) m1 =
736 let stop = return (Just res)
544 let stop = return (Just res)
@@ -742,19 +550,19 @@ minimum m = go Nothing (toStream m)
742 if res <= a
550 if res <= a
743 then go (Just res) r
551 then go (Just res) r
744 else go (Just a) r
552 else go (Just a) r
745− in unStream m1 defState stop single yieldk
553+ in foldStream defState yieldk single stop m1
747{-# INLINE minimumBy #-}
555{-# INLINE minimumBy #-}
749 :: (IsStream t, Monad m)
557 :: (IsStream t, Monad m)
750 => (a -> a -> Ordering) -> t m a -> m (Maybe a)
558 => (a -> a -> Ordering) -> t m a -> m (Maybe a)
751−minimumBy cmp m = go Nothing (toStream m)
559+minimumBy cmp m = go Nothing m
753 go Nothing m1 =
561 go Nothing m1 =
754 let stop = return Nothing
562 let stop = return Nothing
755 single a = return (Just a)
563 single a = return (Just a)
756 yieldk a r = go (Just a) r
564 yieldk a r = go (Just a) r
757− in unStream m1 defState stop single yieldk
565+ in foldStream defState yieldk single stop m1
759 go (Just res) m1 =
567 go (Just res) m1 =
760 let stop = return (Just res)
568 let stop = return (Just res)
@@ -764,17 +572,17 @@ minimumBy cmp m = go Nothing (toStream m)
764 yieldk a r = case cmp res a of
572 yieldk a r = case cmp res a of
765 GT -> go (Just a) r
573 GT -> go (Just a) r
766 _ -> go (Just res) r
574 _ -> go (Just res) r
767− in unStream m1 defState stop single yieldk
575+ in foldStream defState yieldk single stop m1
769{-# INLINE maximum #-}
577{-# INLINE maximum #-}
770maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
578maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
771−maximum m = go Nothing (toStream m)
579+maximum m = go Nothing m
773 go Nothing m1 =
581 go Nothing m1 =
774 let stop = return Nothing
582 let stop = return Nothing
775 single a = return (Just a)
583 single a = return (Just a)
776 yieldk a r = go (Just a) r
584 yieldk a r = go (Just a) r
777− in unStream m1 defState stop single yieldk
585+ in foldStream defState yieldk single stop m1
779 go (Just res) m1 =
587 go (Just res) m1 =
780 let stop = return (Just res)
588 let stop = return (Just res)
@@ -786,17 +594,17 @@ maximum m = go Nothing (toStream m)
786 if res <= a
594 if res <= a
787 then go (Just a) r
595 then go (Just a) r
788 else go (Just res) r
596 else go (Just res) r
789− in unStream m1 defState stop single yieldk
597+ in foldStream defState yieldk single stop m1
791{-# INLINE maximumBy #-}
599{-# INLINE maximumBy #-}
792maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
600maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
793−maximumBy cmp m = go Nothing (toStream m)
601+maximumBy cmp m = go Nothing m
795 go Nothing m1 =
603 go Nothing m1 =
796 let stop = return Nothing
604 let stop = return Nothing
797 single a = return (Just a)
605 single a = return (Just a)
798 yieldk a r = go (Just a) r
606 yieldk a r = go (Just a) r
799− in unStream m1 defState stop single yieldk
607+ in foldStream defState yieldk single stop m1
801 go (Just res) m1 =
609 go (Just res) m1 =
802 let stop = return (Just res)
610 let stop = return (Just res)
@@ -806,11 +614,11 @@ maximumBy cmp m = go Nothing (toStream m)
806 yieldk a r = case cmp res a of
614 yieldk a r = case cmp res a of
807 GT -> go (Just res) r
615 GT -> go (Just res) r
808 _ -> go (Just a) r
616 _ -> go (Just a) r
809− in unStream m1 defState stop single yieldk
617+ in foldStream defState yieldk single stop m1
811{-# INLINE (!!) #-}
619{-# INLINE (!!) #-}
812(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
620(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
813−m !! i = go i (toStream m)
621+m !! i = go i m
815 go n m1 =
623 go n m1 =
816 let single a | n == 0 = return $ Just a
624 let single a | n == 0 = return $ Just a
@@ -818,22 +626,22 @@ m !! i = go i (toStream m)
818 yieldk a x | n < 0 = return Nothing
626 yieldk a x | n < 0 = return Nothing
819 | n == 0 = return $ Just a
627 | n == 0 = return $ Just a
820 | otherwise = go (n - 1) x
628 | otherwise = go (n - 1) x
821− in unStream m1 defState (return Nothing) single yieldk
629+ in foldStream defState yieldk single (return Nothing) m1
823{-# INLINE lookup #-}
631{-# INLINE lookup #-}
824lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
632lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
825−lookup e m = go (toStream m)
633+lookup e m = go m
828 let single (a, b) | a == e = return $ Just b
636 let single (a, b) | a == e = return $ Just b
829 | otherwise = return Nothing
637 | otherwise = return Nothing
830 yieldk (a, b) x | a == e = return $ Just b
638 yieldk (a, b) x | a == e = return $ Just b
831 | otherwise = go x
639 | otherwise = go x
832− in unStream m1 defState (return Nothing) single yieldk
640+ in foldStream defState yieldk single (return Nothing) m1
834{-# INLINE findM #-}
642{-# INLINE findM #-}
835findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
643findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
836−findM p m = go (toStream m)
644+findM p m = go m
839 let single a = do
647 let single a = do
@@ -842,7 +650,7 @@ findM p m = go (toStream m)
842 yieldk a x = do
650 yieldk a x = do
844 if b then return $ Just a else go x
652 if b then return $ Just a else go x
845− in unStream m1 defState (return Nothing) single yieldk
653+ in foldStream defState yieldk single (return Nothing) m1
847{-# INLINE find #-}
655{-# INLINE find #-}
848find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
656find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
@@ -850,14 +658,15 @@ find p = findM (return . p)
851{-# INLINE findIndices #-}
659{-# INLINE findIndices #-}
852findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
660findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
853−findIndices p = fromStream . go 0 . toStream
661+findIndices p = go 0
855− go offset m1 = Stream $ \st stp sng yld ->
663+ go offset m1 = mkStream $ \st yld sng stp ->
856 let single a | p a = sng offset
664 let single a | p a = sng offset
857 | otherwise = stp
665 | otherwise = stp
858 yieldk a x | p a = yld offset $ go (offset + 1) x
666 yieldk a x | p a = yld offset $ go (offset + 1) x
667+ | otherwise = foldStream (adaptState st) yld sng stp $
859− | otherwise = unStream (go (offset + 1) x) st stp sng yld
668+ go (offset + 1) x
860− in unStream m1 (rstState st) stp single yieldk
669+ in foldStream (adaptState st) yieldk single stp m1
862------------------------------------------------------------------------------
671------------------------------------------------------------------------------
863-- Map and Fold
672-- Map and Fold
@@ -865,14 +674,15 @@ findIndices p = fromStream . go 0 . toStream
866-- | Apply a monadic action to each element of the stream and discard the
675-- | Apply a monadic action to each element of the stream and discard the
867-- output of the action.
676-- output of the action.
868mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
678mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
869−mapM_ f m = go (toStream m)
679+mapM_ f m = go m
872 let stop = return ()
682 let stop = return ()
873 single a = void (f a)
683 single a = void (f a)
874 yieldk a r = f a >> go r
684 yieldk a r = f a >> go r
875− in unStream m1 defState stop single yieldk
685+ in foldStream defState yieldk single stop m1
877------------------------------------------------------------------------------
687------------------------------------------------------------------------------
878-- Converting folds
688-- Converting folds
@@ -893,14 +703,14 @@ toStreamK = id
893{-# INLINE scanx #-}
703{-# INLINE scanx #-}
894scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
704scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
895scanx step begin done m =
705scanx step begin done m =
896− cons (done begin) $ fromStream $ go (toStream m) begin
706+ cons (done begin) $ go m begin
898− go m1 !acc = Stream $ \st stp sng yld ->
708+ go m1 !acc = mkStream $ \st yld sng stp ->
899 let single a = sng (done $ step acc a)
709 let single a = sng (done $ step acc a)
900 yieldk a r =
710 yieldk a r =
901 let s = step acc a
711 let s = step acc a
902 in yld (done s) (go r s)
712 in yld (done s) (go r s)
903− in unStream m1 (rstState st) stp single yieldk
713+ in foldStream (adaptState st) yieldk single stp m1
905{-# INLINE scanl' #-}
715{-# INLINE scanl' #-}
906scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
716scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
@@ -912,91 +722,84 @@ scanl' step begin = scanx step begin id
913{-# INLINE filter #-}
723{-# INLINE filter #-}
914filter :: IsStream t => (a -> Bool) -> t m a -> t m a
724filter :: IsStream t => (a -> Bool) -> t m a -> t m a
915−filter p m = fromStream $ go (toStream m)
725+filter p m = go m
917− go m1 = Stream $ \st stp sng yld ->
727+ go m1 = mkStream $ \st yld sng stp ->
918 let single a | p a = sng a
728 let single a | p a = sng a
919 | otherwise = stp
729 | otherwise = stp
920 yieldk a r | p a = yld a (go r)
730 yieldk a r | p a = yld a (go r)
921− | otherwise = unStream r (rstState st) stp single yieldk
731+ | otherwise = foldStream st yieldk single stp r
922− in unStream m1 (rstState st) stp single yieldk
732+ in foldStream st yieldk single stp m1
924{-# INLINE take #-}
734{-# INLINE take #-}
925take :: IsStream t => Int -> t m a -> t m a
735take :: IsStream t => Int -> t m a -> t m a
926−take n m = fromStream $ go n (toStream m)
736+take n m = go n m
928− go n1 m1 = Stream $ \st stp sng yld ->
738+ go n1 m1 = mkStream $ \st yld sng stp ->
929 let yieldk a r = yld a (go (n1 - 1) r)
739 let yieldk a r = yld a (go (n1 - 1) r)
930 in if n1 <= 0
740 in if n1 <= 0
932− else unStream m1 (rstState st) stp sng yieldk
742+ else foldStream st yieldk sng stp m1
934{-# INLINE takeWhile #-}
744{-# INLINE takeWhile #-}
935takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
745takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
936−takeWhile p m = fromStream $ go (toStream m)
746+takeWhile p m = go m
938− go m1 = Stream $ \st stp sng yld ->
748+ go m1 = mkStream $ \st yld sng stp ->
939 let single a | p a = sng a
749 let single a | p a = sng a
940 | otherwise = stp
750 | otherwise = stp
941 yieldk a r | p a = yld a (go r)
751 yieldk a r | p a = yld a (go r)
942 | otherwise = stp
752 | otherwise = stp
943− in unStream m1 (rstState st) stp single yieldk
753+ in foldStream st yieldk single stp m1
945drop :: IsStream t => Int -> t m a -> t m a
756drop :: IsStream t => Int -> t m a -> t m a
946−drop n m = fromStream $ Stream $ \st stp sng yld ->
757+drop n m = fromStream $ unShare (go n (toStream m))
947− unStream (go n (toStream m)) (rstState st) stp sng yld
949− go n1 m1 = Stream $ \st stp sng yld ->
759+ go n1 m1 = mkStream $ \st yld sng stp ->
950 let single _ = stp
760 let single _ = stp
951− yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
761+ yieldk _ r = foldStreamShared st yld sng stp $ go (n1 - 1) r
952 -- Somehow "<=" check performs better than a ">"
762 -- Somehow "<=" check performs better than a ">"
953 in if n1 <= 0
763 in if n1 <= 0
954− then unStream m1 st stp sng yld
764+ then foldStreamShared st yld sng stp m1
955− else unStream m1 st stp single yieldk
765+ else foldStreamShared st yieldk single stp m1
957{-# INLINE dropWhile #-}
767{-# INLINE dropWhile #-}
958dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
768dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
959−dropWhile p m = fromStream $ go (toStream m)
769+dropWhile p m = go m
961− go m1 = Stream $ \st stp sng yld ->
771+ go m1 = mkStream $ \st yld sng stp ->
962 let single a | p a = stp
772 let single a | p a = stp
963 | otherwise = sng a
773 | otherwise = sng a
964− yieldk a r | p a = unStream r (rstState st) stp single yieldk
774+ yieldk a r | p a = foldStream st yieldk single stp r
965 | otherwise = yld a r
775 | otherwise = yld a r
966− in unStream m1 (rstState st) stp single yieldk
776+ in foldStream st yieldk single stp m1
968-------------------------------------------------------------------------------
778-------------------------------------------------------------------------------
969-- Mapping
779-- Mapping
970-------------------------------------------------------------------------------
780-------------------------------------------------------------------------------
973−map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
974−map f m = fromStream $ Stream $ \st stp sng yld ->
975− let single = sng . f
976− yieldk a r = yld (f a) (fmap f r)
977− in unStream (toStream m) (rstState st) stp single yieldk
979-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
782-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
980-- other stream types to overload it.
783-- other stream types to overload it.
981{-# INLINE mapM #-}
784{-# INLINE mapM #-}
982mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
785mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
983−mapM f m = go (toStream m)
786+mapM f m = go m
985− go m1 = fromStream $ Stream $ \st stp sng yld ->
788+ go m1 = mkStream $ \st yld sng stp ->
986 let single a = f a >>= sng
789 let single a = f a >>= sng
987− yieldk a r = unStream (toStream (f a |: go r)) st stp sng yld
790+ yieldk a r = foldStreamShared st yld sng stp $ f a |: go r
988− in unStream m1 (rstState st) stp single yieldk
791+ in foldStream (adaptState st) yieldk single stp m1
990-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
793-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
991-- other stream types to overload it.
794-- other stream types to overload it.
992{-# INLINE sequence #-}
795{-# INLINE sequence #-}
993sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
796sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
994−sequence m = go (toStream m)
797+sequence m = go m
996− go m1 = fromStream $ Stream $ \st stp sng yld ->
799+ go m1 = mkStream $ \st yld sng stp ->
997 let single ma = ma >>= sng
800 let single ma = ma >>= sng
998− yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld
801+ yieldk ma r = foldStreamShared st yld sng stp $ ma |: go r
999− in unStream m1 (rstState st) stp single yieldk
802+ in foldStream (adaptState st) yieldk single stp m1
1001-------------------------------------------------------------------------------
804-------------------------------------------------------------------------------
1002-- Inserting
805-- Inserting
@@ -1004,21 +807,21 @@ sequence m = go (toStream m)
1005{-# INLINE intersperseM #-}
808{-# INLINE intersperseM #-}
1006intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
809intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1007−intersperseM a m = fromStream $ prependingStart (toStream m)
810+intersperseM a m = prependingStart m
1009− prependingStart m1 = Stream $ \st stp sng yld ->
812+ prependingStart m1 = mkStream $ \st yld sng stp ->
1010− let yieldk i x = unStream (return i |: go x) st stp sng yld
813+ let yieldk i x = foldStreamShared st yld sng stp $ return i |: go x
1011− in unStream m1 (rstState st) stp sng yieldk
814+ in foldStream st yieldk sng stp m1
1012− go m2 = fromStream $ Stream $ \st stp sng yld ->
815+ go m2 = mkStream $ \st yld sng stp ->
1013− let single i = unStream (a |: yield i) st stp sng yld
816+ let single i = foldStreamShared st yld sng stp $ a |: yield i
1014− yieldk i x = unStream (a |: return i |: go x) st stp sng yld
817+ yieldk i x = foldStreamShared st yld sng stp $ a |: return i |: go x
1015− in unStream m2 (rstState st) stp single yieldk
818+ in foldStream st yieldk single stp m2
1017{-# INLINE insertBy #-}
820{-# INLINE insertBy #-}
1018insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
821insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
1019−insertBy cmp x m = fromStream $ go (toStream m)
822+insertBy cmp x m = go m
1021− go m1 = Stream $ \st _ _ yld ->
824+ go m1 = mkStream $ \st yld _ _ ->
1022 let single a = case cmp x a of
825 let single a = case cmp x a of
1023 GT -> yld a (yield x)
826 GT -> yld a (yield x)
1024 _ -> yld x (yield a)
827 _ -> yld x (yield a)
@@ -1026,7 +829,7 @@ insertBy cmp x m = fromStream $ go (toStream m)
1026 yieldk a r = case cmp x a of
829 yieldk a r = case cmp x a of
1027 GT -> yld a (go r)
830 GT -> yld a (go r)
1028 _ -> yld x (a `cons` r)
831 _ -> yld x (a `cons` r)
1029− in unStream m1 (rstState st) stop single yieldk
832+ in foldStream st yieldk single stop m1
1031------------------------------------------------------------------------------
834------------------------------------------------------------------------------
1032-- Deleting
835-- Deleting
@@ -1034,14 +837,14 @@ insertBy cmp x m = fromStream $ go (toStream m)
1035{-# INLINE deleteBy #-}
838{-# INLINE deleteBy #-}
1036deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
839deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
1037−deleteBy eq x m = fromStream $ go (toStream m)
840+deleteBy eq x m = go m
1039− go m1 = Stream $ \st stp sng yld ->
842+ go m1 = mkStream $ \st yld sng stp ->
1040 let single a = if eq x a then stp else sng a
843 let single a = if eq x a then stp else sng a
1041 yieldk a r = if eq x a
844 yieldk a r = if eq x a
1042− then unStream r (rstState st) stp sng yld
845+ then foldStream st yld sng stp r
1043 else yld a (go r)
846 else yld a (go r)
1044− in unStream m1 (rstState st) stp single yieldk
847+ in foldStream st yieldk single stp m1
1046-------------------------------------------------------------------------------
849-------------------------------------------------------------------------------
1047-- Map and Filter
850-- Map and Filter
@@ -1049,70 +852,67 @@ deleteBy eq x m = fromStream $ go (toStream m)
1050{-# INLINE mapMaybe #-}
853{-# INLINE mapMaybe #-}
1051mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
854mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
1052−mapMaybe f m = go (toStream m)
855+mapMaybe f m = go m
1054− go m1 = fromStream $ Stream $ \st stp sng yld ->
857+ go m1 = mkStream $ \st yld sng stp ->
1055 let single a = case f a of
858 let single a = case f a of
1056 Just b -> sng b
859 Just b -> sng b
1057 Nothing -> stp
860 Nothing -> stp
1058 yieldk a r = case f a of
861 yieldk a r = case f a of
1059− Just b -> yld b (toStream $ go r)
862+ Just b -> yld b $ go r
1060− Nothing -> unStream r (rstState st) stp single yieldk
863+ Nothing -> foldStream (adaptState st) yieldk single stp r
1061− in unStream m1 (rstState st) stp single yieldk
864+ in foldStream (adaptState st) yieldk single stp m1
1063------------------------------------------------------------------------------
866------------------------------------------------------------------------------
1064-- Serial Zipping
867-- Serial Zipping
1065------------------------------------------------------------------------------
868------------------------------------------------------------------------------
1067−{-# INLINE zipWithS #-}
870+-- | Zip two streams serially using a pure zipping function.
1068−zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
871+--
1069−zipWithS f = go
872+-- @since 0.1.0
873+{-# INLINABLE zipWith #-}
874+zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
1071− go mx my = Stream $ \st stp sng yld -> do
877+ go mx my = mkStream $ \st yld sng stp -> do
1072 let merge a ra =
878 let merge a ra =
1073 let single2 b = sng (f a b)
879 let single2 b = sng (f a b)
1074 yield2 b rb = yld (f a b) (go ra rb)
880 yield2 b rb = yld (f a b) (go ra rb)
1075− in unStream my (rstState st) stp single2 yield2
881+ in foldStream (adaptState st) yield2 single2 stp my
1076 let single1 a = merge a nil
882 let single1 a = merge a nil
1077 yield1 = merge
883 yield1 = merge
1078− unStream mx (rstState st) stp single1 yield1
884+ foldStream (adaptState st) yield1 single1 stp mx
1080−-- | Zip two streams serially using a pure zipping function.
1083−{-# INLINABLE zipWith #-}
1084−zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
1085−zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)
1087-- | Zip two streams serially using a monadic zipping function.
886-- | Zip two streams serially using a monadic zipping function.
1089-- @since 0.1.0
888-- @since 0.1.0
889+{-# INLINABLE zipWithM #-}
1090zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
890zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
1091−zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
891+zipWithM f m1 m2 = go m1 m2
1093− go mx my = Stream $ \st stp sng yld -> do
893+ go mx my = mkStream $ \st yld sng stp -> do
1094 let merge a ra =
894 let merge a ra =
1095− let runIt x = unStream x (rstState st) stp sng yld
895+ let runIt x = foldStream st yld sng stp x
1096 single2 b = f a b >>= sng
896 single2 b = f a b >>= sng
1097 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
897 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
1098− in unStream my (rstState st) stp single2 yield2
898+ in foldStream (adaptState st) yield2 single2 stp my
1099 let single1 a = merge a nil
899 let single1 a = merge a nil
1100 yield1 = merge
900 yield1 = merge
1101− unStream mx (rstState st) stp single1 yield1
901+ foldStream (adaptState st) yield1 single1 stp mx
1103------------------------------------------------------------------------------
903------------------------------------------------------------------------------
1104-- Merging
904-- Merging
1105------------------------------------------------------------------------------
905------------------------------------------------------------------------------
1107−{-# INLINE mergeByS #-}
907+{-# INLINE mergeByM #-}
1108−mergeByS
908+mergeByM
1109− :: Monad m
909+ :: (IsStream t, Monad m)
1110− => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
910+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1111−mergeByS cmp = go
911+mergeByM cmp = go
1113− go mx my = Stream $ \st stp sng yld -> do
913+ go mx my = mkStream $ \st yld sng stp -> do
1114 let mergeWithY a ra =
914 let mergeWithY a ra =
1115− let stop2 = unStream mx (rstState st) stp sng yld
915+ let stop2 = foldStream st yld sng stp mx
1116 single2 b = do
916 single2 b = do
1117 r <- cmp a b
917 r <- cmp a b
1118 case r of
918 case r of
@@ -1123,17 +923,11 @@ mergeByS cmp = go
1123 case r of
923 case r of
1124 GT -> yld b (go (a `cons` ra) rb)
924 GT -> yld b (go (a `cons` ra) rb)
1125 _ -> yld a (go ra (b `cons` rb))
925 _ -> yld a (go ra (b `cons` rb))
1126− in unStream my (rstState st) stop2 single2 yield2
926+ in foldStream st yield2 single2 stop2 my
1127− let stopX = unStream my (rstState st) stp sng yld
927+ let stopX = foldStream st yld sng stp my
1128 singleX a = mergeWithY a nil
928 singleX a = mergeWithY a nil
1129 yieldX = mergeWithY
929 yieldX = mergeWithY
1130− unStream mx (rstState st) stopX singleX yieldX
930+ foldStream st yieldX singleX stopX mx
1132−{-# INLINABLE mergeByM #-}
1134− :: (IsStream t, Monad m)
1135− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1136−mergeByM f m1 m2 = fromStream $ mergeByS f (toStream m1) (toStream m2)
1138{-# INLINABLE mergeBy #-}
932{-# INLINABLE mergeBy #-}
@@ -1151,48 +945,14 @@ the m = do
1151 r <- uncons m
945 r <- uncons m
1152 case r of
946 case r of
1153 Nothing -> return Nothing
947 Nothing -> return Nothing
1154− Just (h, t) -> go h (toStream t)
948+ Just (h, t) -> go h t
1156 go h m1 =
950 go h m1 =
1157 let single a | h == a = return $ Just h
951 let single a | h == a = return $ Just h
1158 | otherwise = return Nothing
952 | otherwise = return Nothing
1159 yieldk a r | h == a = go h r
953 yieldk a r | h == a = go h r
1160 | otherwise = return Nothing
954 | otherwise = return Nothing
1161− in unStream m1 defState (return $ Just h) single yieldk
955+ in foldStream defState yieldk single (return $ Just h) m1
1163−------------------------------------------------------------------------------
1165−------------------------------------------------------------------------------
1167−-- | Concatenates two streams sequentially i.e. the first stream is
1168−-- exhausted completely before yielding any element from the second stream.
1169−{-# INLINE serial #-}
1170−serial :: Stream m a -> Stream m a -> Stream m a
1171−serial m1 m2 = go m1
1173− go (Stream m) = Stream $ \st stp sng yld ->
1174− let stop = unStream m2 (rstState st) stp sng yld
1175− single a = yld a m2
1176− yieldk a r = yld a (go r)
1177− in m (rstState st) stop single yieldk
1179−instance Semigroup (Stream m a) where
1182−------------------------------------------------------------------------------
1184−------------------------------------------------------------------------------
1186−instance Monoid (Stream m a) where
1190−-------------------------------------------------------------------------------
1192−-------------------------------------------------------------------------------
1194−instance Monad m => Functor (Stream m) where
1197-------------------------------------------------------------------------------
957-------------------------------------------------------------------------------
1198-- Bind utility
958-- Bind utility
@@ -1200,40 +960,40 @@ instance Monad m => Functor (Stream m) where
1201{-# INLINE bindWith #-}
961{-# INLINE bindWith #-}
1203− :: (forall c. Stream m c -> Stream m c -> Stream m c)
964+ => (forall c. t m c -> t m c -> t m c)
1204− -> Stream m a
965+ -> t m a
1205− -> (a -> Stream m b)
966+ -> (a -> t m b)
1206− -> Stream m b
967+ -> t m b
1207−bindWith par m f = go m
968+bindWith par m1 f = go m1
1209− go (Stream g) =
970+ go m =
1210− Stream $ \st stp sng yld ->
971+ mkStream $ \st yld sng stp ->
1211− let runShared x = unstreamShared x st stp sng yld
972+ let foldShared = foldStreamShared st yld sng stp
1212− runIsolated x = unStreamIsolated x st stp sng yld
973+ single a = foldShared $ unShare (f a)
1214− single a = runIsolated $ f a
1215− yieldk a r = runShared $ isolateStream (f a) `par` go r
974+ yieldk a r = foldShared $ unShare (f a) `par` go r
1216− in g (rstState st) stp single yieldk
975+ in foldStream (adaptState st) yieldk single stp m
1218------------------------------------------------------------------------------
977------------------------------------------------------------------------------
1219-- Alternative & MonadPlus
978-- Alternative & MonadPlus
1220------------------------------------------------------------------------------
979------------------------------------------------------------------------------
1222_alt :: Stream m a -> Stream m a -> Stream m a
981_alt :: Stream m a -> Stream m a -> Stream m a
1223−_alt m1 m2 = Stream $ \st stp sng yld ->
982+_alt m1 m2 = mkStream $ \st yld sng stp ->
1224− let stop = unStream m2 (rstState st) stp sng yld
983+ let stop = foldStream st yld sng stp m2
1225− in unStream m1 (rstState st) stop sng yld
984+ in foldStream st yld sng stop m1
1227------------------------------------------------------------------------------
986------------------------------------------------------------------------------
1228-- MonadReader
987-- MonadReader
1229------------------------------------------------------------------------------
988------------------------------------------------------------------------------
990+{-# INLINABLE withLocal #-}
1231withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
991withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1232withLocal f m =
992withLocal f m =
1233− Stream $ \st stp sng yld ->
993+ mkStream $ \st yld sng stp ->
1234 let single = local f . sng
994 let single = local f . sng
1235 yieldk a r = local f $ yld a (withLocal f r)
995 yieldk a r = local f $ yld a (withLocal f r)
1236− in unStream m (rstState st) (local f stp) single yieldk
996+ in foldStream st yieldk single (local f stp) m
1238------------------------------------------------------------------------------
998------------------------------------------------------------------------------
1239-- MonadError
999-- MonadError
@@ -1245,16 +1005,9 @@ withCatchError
1245 :: MonadError e m
1005 :: MonadError e m
1246 => Stream m a -> (e -> Stream m a) -> Stream m a
1006 => Stream m a -> (e -> Stream m a) -> Stream m a
1247withCatchError m h =
1007withCatchError m h =
1248− Stream $ \_ stp sng yld ->
1008+ mkStream $ \_ stp sng yld ->
1249 let run x = unStream x Nothing stp sng yieldk
1009 let run x = unStream x Nothing stp sng yieldk
1250 handle r = r `catchError` \e -> run $ h e
1010 handle r = r `catchError` \e -> run $ h e
1251 yieldk a r = yld a (withCatchError r h)
1011 yieldk a r = yld a (withCatchError r h)
1252 in handle $ run m
1012 in handle $ run m
1255−-------------------------------------------------------------------------------
1257−-------------------------------------------------------------------------------
1259−instance MonadTrans Stream where