1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE FlexibleInstances #-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE LambdaCase #-}
8{-# LANGUAGE MagicHash #-}
9{-# LANGUAGE MultiParamTypeClasses #-}
10{-# LANGUAGE RankNTypes #-}
11{-# LANGUAGE UnboxedTuples #-}
12{-# LANGUAGE UndecidableInstances #-} -- XXX
15-- Module : Streamly.Streams.StreamK
16-- Copyright : (c) 2017 Harendra Kumar
19-- Maintainer : harendra.kumar@gmail.com
20-- Stability : experimental
24-- Continuation passing style (CPS) stream implementation. The symbol 'K' below
25-- denotes a function as well as a Kontinuation.
28-- import qualified Streamly.Streams.StreamK as K
31module Streamly.Streams.StreamK
33 -- * A class for streams
46 -- * Asynchronous construction
59 -- ** Specialized Generation
80 -- ** Specialized Folds
105 -- ** By folding (scans)
131 -- * Semigroup Style Composition
140 , Streaming -- deprecated
145import Control.Monad (void)
146import Control.Monad.Reader.Class (MonadReader(..))
147import Control.Monad.Trans.Class (MonadTrans(lift))
148import Data.Semigroup (Semigroup(..))
150 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
151 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
152 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,
154import qualified Prelude
158------------------------------------------------------------------------------
159-- The basic stream type
160------------------------------------------------------------------------------
162-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
163-- constructed using actions in monad 'm'. It uses stop, singleton and yield
164-- continuations equivalent to the following direct style type:
167-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
170-- To facilitate parallel composition we maintain a local state in an 'SVar'
171-- that is shared across and is used for synchronization of the streams being
174-- The singleton case can be expressed in terms of stop and yield but we have
175-- it as a separate case to optimize composition operations for streams with
176-- single element. We build singleton streams in the implementation of 'pure'
177-- for Applicative and Monad, and in 'lift' for MonadTrans.
181 unStream :: forall r.
182 State Stream m a -- state
184 -> (a -> m r) -- singleton
185 -> (a -> Stream m a -> m r) -- yield
189------------------------------------------------------------------------------
190-- Types that can behave as a Stream
191------------------------------------------------------------------------------
196-- | Class of types that can represent a stream of elements of some type 'a' in
200class IsStream t where
201 toStream :: t m a -> Stream m a
202 fromStream :: Stream m a -> t m a
203 -- | Constructs a stream by adding a monadic action at the head of an
204 -- existing stream. For example:
207 -- > toList $ getLine \`consM` getLine \`consM` nil
213 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
216 consM :: MonadAsync m => m a -> t m a -> t m a
217 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
218 -- to remember that @|@ comes before ':'.
221 -- > toList $ getLine |: getLine |: nil
228 -- let delay = threadDelay 1000000 >> print 1
229 -- runStream $ serially $ delay |: delay |: delay |: nil
230 -- runStream $ parallely $ delay |: delay |: delay |: nil
233 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
236 (|:) :: MonadAsync m => m a -> t m a -> t m a
237 -- We can define (|:) just as 'consM' but it is defined explicitly for each
238 -- type because we want to use SPECIALIZE pragma on the definition.
240-- | Same as 'IsStream'.
243{-# DEPRECATED Streaming "Please use IsStream instead." #-}
244type Streaming = IsStream
246-------------------------------------------------------------------------------
247-- Type adapting combinators
248-------------------------------------------------------------------------------
250-- | Adapt any specific stream type to any other specific stream type.
253adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
254adapt = fromStream . toStream
256------------------------------------------------------------------------------
258------------------------------------------------------------------------------
260-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
261-- continuation and a yield continuation.
263 => (forall r. State Stream m a
266 -> (a -> t m a -> m r)
269mkStream k = fromStream $ Stream $ \st stp sng yld ->
270 let yieldk a r = yld a (toStream r)
271 in k (rstState st) stp sng yieldk
273------------------------------------------------------------------------------
275------------------------------------------------------------------------------
285nil :: IsStream t => t m a
286nil = fromStream $ Stream $ \_ stp _ _ -> stp
290-- faster than consM because there is no bind.
291-- | Construct a stream by adding a pure value at the head of an existing
292-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
293-- more efficient. For concurrent streams this is not concurrent whereas
294-- 'consM' is concurrent. For example:
297-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
302cons :: IsStream t => a -> t m a -> t m a
303cons a r = fromStream $ Stream $ \_ _ _ yld -> yld a (toStream r)
307-- | Operator equivalent of 'cons'.
310-- > toList $ 1 .: 2 .: 3 .: nil
315(.:) :: IsStream t => a -> t m a -> t m a
318{-# INLINE consMSerial #-}
319consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a
320consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
322------------------------------------------------------------------------------
323-- Asynchronous construction
324------------------------------------------------------------------------------
326-- | Make an empty stream from a callback function.
327nilK :: IsStream t => (forall r. m r -> m r) -> t m a
328nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp
330-- | Make a singleton stream from a one shot callback function.
331yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
332yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng
334-- | Construct a stream from a callback function.
335consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a
336consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))
338-- XXX consK with concurrent callbacks
339-- XXX Build a stream from a repeating callback function.
341-------------------------------------------------------------------------------
343-------------------------------------------------------------------------------
345instance IsStream Stream where
350 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
351 consM :: Monad m => m a -> Stream m a -> Stream m a
355 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
356 (|:) :: Monad m => m a -> Stream m a -> Stream m a
359-------------------------------------------------------------------------------
361-------------------------------------------------------------------------------
364uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
366 let stop = return Nothing
367 single a = return (Just (a, nil))
368 yieldk a r = return (Just (a, fromStream r))
369 in (unStream (toStream m)) defState stop single yieldk
371-------------------------------------------------------------------------------
373-------------------------------------------------------------------------------
375{-# INLINE unfoldr #-}
376unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
377unfoldr step = fromStream . go
379 go s = Stream $ \_ stp _ yld ->
382 Just (a, b) -> yld a (go b)
384{-# INLINE unfoldrM #-}
385unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
388 go s = fromStream $ Stream $ \svr stp sng yld -> do
393 unStream (toStream (return a |: go b)) svr stp sng yld
395-------------------------------------------------------------------------------
397-------------------------------------------------------------------------------
399-- Faster than yieldM because there is no bind. Usually we can construct a
400-- stream from a pure value using "pure" in an applicative, however in case of
401-- Zip streams pure creates an infinite stream.
403-- | Create a singleton stream from a pure value. In monadic streams, 'pure' or
404-- 'return' can be used in place of 'yield', however, in Zip applicative
405-- streams 'pure' is equivalent to 'repeat'.
408yield :: IsStream t => a -> t m a
409yield a = fromStream $ Stream $ \_ _ single _ -> single a
411-- | Create a singleton stream from a monadic action. Same as @m \`consM` nil@
412-- but more efficient.
415-- > toList $ yieldM getLine
422yieldM :: (Monad m, IsStream t) => m a -> t m a
423yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single
428{-# DEPRECATED once "Please use yieldM instead." #-}
430once :: (Monad m, IsStream t) => m a -> t m a
433-- | Generate an infinite stream by repeating a pure value.
436repeat :: IsStream t => a -> t m a
437repeat a = let x = cons a x in x
439-------------------------------------------------------------------------------
441-------------------------------------------------------------------------------
443-- | Construct a stream from a 'Foldable' containing pure values.
446{-# INLINE fromFoldable #-}
447fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
448fromFoldable = Prelude.foldr cons nil
450{-# INLINE fromList #-}
451fromList :: IsStream t => [a] -> t m a
452fromList = fromFoldable
454{-# INLINE fromStreamK #-}
455fromStreamK :: Stream m a -> Stream m a
458-------------------------------------------------------------------------------
459-- Elimination by Folding
460-------------------------------------------------------------------------------
462-- | Fold a stream by providing an SVar, a stop continuation, a singleton
463-- continuation and a yield continuation.
469 -> (a -> t m a -> m r)
472foldStream st blank single step m =
473 let yieldk a x = step a (fromStream x)
474 in (unStream (toStream m)) st blank single yieldk
476-- | Lazy right associative fold.
477foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
478foldr step acc m = go (toStream m)
481 let stop = return acc
482 single a = return (step a acc)
483 yieldk a r = go r >>= \b -> return (step a b)
484 in (unStream m1) defState stop single yieldk
486-- | Lazy right fold with a monadic step function.
488foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
489foldrM step acc m = go (toStream m)
492 let stop = return acc
493 single a = step a acc
494 yieldk a r = go r >>= step a
495 in (unStream m1) defState stop single yieldk
498foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
502 Nothing -> return Nothing
503 Just (h, t) -> go h (toStream t) >>= return . Just
507 single a = return $ step a p
508 yieldk a r = go a r >>= return . (step p)
509 in unStream m1 defState stp single yieldk
511-- | Strict left fold with an extraction function. Like the standard strict
512-- left fold, but applies a user supplied extraction function (the third
513-- argument) to the folded value at the end. This is designed to work with the
514-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
516foldx :: (IsStream t, Monad m)
517 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
518foldx step begin done m = get $ go (toStream m) begin
522 let single = return . done
523 in (unStream m1) undefined undefined single undefined
525 -- Note, this can be implemented by making a recursive call to "go",
526 -- however that is more expensive because of unnecessary recursion
527 -- that cannot be tail call optimized. Unfolding recursion explicitly via
528 -- continuations is much more efficient.
529 go m1 !acc = Stream $ \_ _ sng yld ->
531 single a = sng $ step acc a
533 let stream = go r (step acc a)
534 in (unStream stream) defState undefined sng yld
535 in (unStream m1) defState stop single yieldk
537-- | Strict left associative fold.
539foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
540foldl' step begin m = foldx step begin id m
542-- XXX replace the recursive "go" with explicit continuations.
543-- | Like 'foldx', but with a monadic step function.
544foldxM :: (IsStream t, Monad m)
545 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
546foldxM step begin done m = go begin (toStream m)
549 let stop = acc >>= done
550 single a = acc >>= \b -> step b a >>= done
551 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
552 in (unStream m1) defState stop single yieldk
554-- | Like 'foldl'' but with a monadic step function.
555foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
556foldlM' step begin m = foldxM step (return begin) return m
558------------------------------------------------------------------------------
560------------------------------------------------------------------------------
562{-# INLINE runStream #-}
563runStream :: (Monad m, IsStream t) => t m a -> m ()
564runStream m = go (toStream m)
569 yieldk _ r = go (toStream r)
570 in unStream m1 defState stop single yieldk
573null :: (IsStream t, Monad m) => t m a -> m Bool
575 let stop = return True
576 single _ = return False
577 yieldk _ _ = return False
578 in unStream (toStream m) defState stop single yieldk
581head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
583 let stop = return Nothing
584 single a = return (Just a)
585 yieldk a _ = return (Just a)
586 in unStream (toStream m) defState stop single yieldk
589tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
591 let stop = return Nothing
592 single _ = return $ Just nil
593 yieldk _ r = return $ Just $ fromStream r
594 in unStream (toStream m) defState stop single yieldk
597init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
598init m = go1 (toStream m)
603 Nothing -> return Nothing
604 Just (h, t) -> return . Just . fromStream $ go h t
605 go p m1 = Stream $ \_ stp sng yld ->
607 yieldk a x = yld p $ go a x
608 in unStream m1 defState stp single yieldk
611elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
612elem e m = go (toStream m)
615 let stop = return False
616 single a = return (a == e)
617 yieldk a r = if a == e then return True else go r
618 in (unStream m1) defState stop single yieldk
620{-# INLINE notElem #-}
621notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
622notElem e m = go (toStream m)
625 let stop = return True
626 single a = return (a /= e)
627 yieldk a r = if a == e then return False else go r
628 in (unStream m1) defState stop single yieldk
630all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
631all p m = go (toStream m)
634 let single a | p a = return True
635 | otherwise = return False
636 yieldk a r | p a = go r
637 | otherwise = return False
638 in unStream m1 defState (return True) single yieldk
640any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
641any p m = go (toStream m)
644 let single a | p a = return True
645 | otherwise = return False
646 yieldk a r | p a = return True
648 in unStream m1 defState (return False) single yieldk
650-- | Extract the last element of the stream, if any.
652last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
653last = foldx (\_ y -> Just y) Nothing id
655{-# INLINE minimum #-}
656minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
657minimum m = go Nothing (toStream m)
660 let stop = return Nothing
661 single a = return (Just a)
662 yieldk a r = go (Just a) r
663 in unStream m1 defState stop single yieldk
666 let stop = return (Just res)
669 then return (Just res)
675 in unStream m1 defState stop single yieldk
677{-# INLINE maximum #-}
678maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
679maximum m = go Nothing (toStream m)
682 let stop = return Nothing
683 single a = return (Just a)
684 yieldk a r = go (Just a) r
685 in unStream m1 defState stop single yieldk
688 let stop = return (Just res)
692 else return (Just res)
697 in unStream m1 defState stop single yieldk
700lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
701lookup e m = go (toStream m)
704 let single (a, b) | a == e = return $ Just b
705 | otherwise = return Nothing
706 yieldk (a, b) x | a == e = return $ Just b
708 in unStream m1 defState (return Nothing) single yieldk
711find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
712find p m = go (toStream m)
715 let single a | p a = return $ Just a
716 | otherwise = return Nothing
717 yieldk a x | p a = return $ Just a
719 in unStream m1 defState (return Nothing) single yieldk
721{-# INLINE findIndices #-}
722findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
723findIndices p = fromStream . go 0 . toStream
725 go offset m1 = Stream $ \st stp sng yld ->
726 let single a | p a = sng offset
728 yieldk a x | p a = yld offset $ go (offset + 1) x
729 | otherwise = unStream (go (offset + 1) x) st stp sng yld
730 in unStream m1 (rstState st) stp single yieldk
732------------------------------------------------------------------------------
734------------------------------------------------------------------------------
736-- | Apply a monadic action to each element of the stream and discard the
737-- output of the action.
738mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
739mapM_ f m = go (toStream m)
743 single a = void (f a)
744 yieldk a r = f a >> go r
745 in (unStream m1) defState stop single yieldk
747------------------------------------------------------------------------------
749------------------------------------------------------------------------------
751{-# INLINABLE toList #-}
752toList :: (IsStream t, Monad m) => t m a -> m [a]
755{-# INLINE toStreamK #-}
756toStreamK :: Stream m a -> Stream m a
759-------------------------------------------------------------------------------
760-- Transformation by folding (Scans)
761-------------------------------------------------------------------------------
764scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
765scanx step begin done m =
766 cons (done begin) $ fromStream $ go (toStream m) begin
768 go m1 !acc = Stream $ \st stp sng yld ->
769 let single a = sng (done $ step acc a)
772 in yld (done s) (go r s)
773 in unStream m1 (rstState st) stp single yieldk
776scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
777scanl' step begin m = scanx step begin id m
779-------------------------------------------------------------------------------
781-------------------------------------------------------------------------------
784filter :: IsStream t => (a -> Bool) -> t m a -> t m a
785filter p m = fromStream $ go (toStream m)
787 go m1 = Stream $ \st stp sng yld ->
788 let single a | p a = sng a
790 yieldk a r | p a = yld a (go r)
791 | otherwise = (unStream r) (rstState st) stp single yieldk
792 in unStream m1 (rstState st) stp single yieldk
795take :: IsStream t => Int -> t m a -> t m a
796take n m = fromStream $ go n (toStream m)
798 go n1 m1 = Stream $ \st stp sng yld ->
799 let yieldk a r = yld a (go (n1 - 1) r)
802 else unStream m1 (rstState st) stp sng yieldk
804{-# INLINE takeWhile #-}
805takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
806takeWhile p m = fromStream $ go (toStream m)
808 go m1 = Stream $ \st stp sng yld ->
809 let single a | p a = sng a
811 yieldk a r | p a = yld a (go r)
813 in unStream m1 (rstState st) stp single yieldk
815drop :: IsStream t => Int -> t m a -> t m a
816drop n m = fromStream $ Stream $ \st stp sng yld ->
817 unStream (go n (toStream m)) (rstState st) stp sng yld
819 go n1 m1 = Stream $ \st stp sng yld ->
821 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
822 -- Somehow "<=" check performs better than a ">"
824 then unStream m1 st stp sng yld
825 else unStream m1 st stp single yieldk
827{-# INLINE dropWhile #-}
828dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
829dropWhile p m = fromStream $ go (toStream m)
831 go m1 = Stream $ \st stp sng yld ->
832 let single a | p a = stp
834 yieldk a r | p a = (unStream r) (rstState st) stp single yieldk
835 | otherwise = yld a r
836 in unStream m1 (rstState st) stp single yieldk
838-------------------------------------------------------------------------------
840-------------------------------------------------------------------------------
843map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
844map f m = fromStream $ Stream $ \st stp sng yld ->
846 yieldk a r = yld (f a) (fmap f r)
847 in unStream (toStream m) (rstState st) stp single yieldk
849-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
850-- other stream types to overload it.
852mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
853mapM f m = go (toStream m)
855 go m1 = fromStream $ Stream $ \st stp sng yld ->
856 let single a = f a >>= sng
857 yieldk a r = unStream (toStream (f a |: (go r))) st stp sng yld
858 in (unStream m1) (rstState st) stp single yieldk
860-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
861-- other stream types to overload it.
862{-# INLINE sequence #-}
863sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
864sequence m = go (toStream m)
866 go m1 = fromStream $ Stream $ \st stp sng yld ->
867 let single ma = ma >>= sng
868 yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld
869 in (unStream m1) (rstState st) stp single yieldk
871-------------------------------------------------------------------------------
873-------------------------------------------------------------------------------
875{-# INLINE intersperseM #-}
876intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
877intersperseM a m = fromStream $ prependingStart (toStream m)
879 prependingStart m1 = Stream $ \st stp sng yld ->
880 let yieldk i x = unStream (return i |: go x) st stp sng yld
881 in unStream m1 (rstState st) stp sng yieldk
882 go m2 = fromStream $ Stream $ \st stp sng yld ->
883 let single i = unStream (a |: yield i) st stp sng yld
884 yieldk i x = unStream (a |: return i |: go x) st stp sng yld
885 in unStream m2 (rstState st) stp single yieldk
887-------------------------------------------------------------------------------
889-------------------------------------------------------------------------------
891{-# INLINE mapMaybe #-}
892mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
893mapMaybe f m = go (toStream m)
895 go m1 = fromStream $ Stream $ \st stp sng yld ->
896 let single a = case f a of
899 yieldk a r = case f a of
900 Just b -> yld b (toStream $ go r)
901 Nothing -> (unStream r) (rstState st) stp single yieldk
902 in unStream m1 (rstState st) stp single yieldk
904------------------------------------------------------------------------------
906------------------------------------------------------------------------------
908{-# INLINE zipWithS #-}
909zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
910zipWithS f m1 m2 = go m1 m2
912 go mx my = Stream $ \st stp sng yld -> do
914 let single2 b = sng (f a b)
915 yield2 b rb = yld (f a b) (go ra rb)
916 in unStream my (rstState st) stp single2 yield2
917 let single1 a = merge a nil
918 yield1 a ra = merge a ra
919 unStream mx (rstState st) stp single1 yield1
921-- | Zip two streams serially using a pure zipping function.
924{-# INLINABLE zipWith #-}
925zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
926zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)
928-- | Zip two streams serially using a monadic zipping function.
931zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
932zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
934 go mx my = Stream $ \st stp sng yld -> do
936 let runIt x = unStream x (rstState st) stp sng yld
937 single2 b = f a b >>= sng
938 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
939 in unStream my (rstState st) stp single2 yield2
940 let single1 a = merge a nil
941 yield1 a ra = merge a ra
942 unStream mx (rstState st) stp single1 yield1
944------------------------------------------------------------------------------
946------------------------------------------------------------------------------
948-- | Concatenates two streams sequentially i.e. the first stream is
949-- exhausted completely before yielding any element from the second stream.
951serial :: Stream m a -> Stream m a -> Stream m a
954 go (Stream m) = Stream $ \st stp sng yld ->
955 let stop = (unStream m2) (rstState st) stp sng yld
957 yieldk a r = yld a (go r)
958 in m (rstState st) stop single yieldk
960instance Semigroup (Stream m a) where
963------------------------------------------------------------------------------
965------------------------------------------------------------------------------
967instance Monoid (Stream m a) where
971-------------------------------------------------------------------------------
973-------------------------------------------------------------------------------
975instance Monad m => Functor (Stream m) where
978-------------------------------------------------------------------------------
980-------------------------------------------------------------------------------
982{-# INLINE bindWith #-}
984 :: (forall c. Stream m c -> Stream m c -> Stream m c)
988bindWith par m f = go m
991 Stream $ \st stp sng yld ->
992 let run x = (unStream x) st stp sng yld
994 yieldk a r = run $ f a `par` go r
995 in g (rstState st) stp single yieldk
997------------------------------------------------------------------------------
998-- Alternative & MonadPlus
999------------------------------------------------------------------------------
1001_alt :: Stream m a -> Stream m a -> Stream m a
1002_alt m1 m2 = Stream $ \st stp sng yld ->
1003 let stop = unStream m2 (rstState st) stp sng yld
1004 in unStream m1 (rstState st) stop sng yld
1006------------------------------------------------------------------------------
1008------------------------------------------------------------------------------
1010withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1012 Stream $ \st stp sng yld ->
1013 let single = local f . sng
1014 yieldk a r = local f $ yld a (withLocal f r)
1015 in (unStream m) (rstState st) (local f stp) single yieldk
1017------------------------------------------------------------------------------
1019------------------------------------------------------------------------------
1022-- XXX handle and test cross thread state transfer
1025 => Stream m a -> (e -> Stream m a) -> Stream m a
1027 Stream $ \_ stp sng yld ->
1028 let run x = unStream x Nothing stp sng yieldk
1029 handle r = r `catchError` \e -> run $ h e
1030 yieldk a r = yld a (withCatchError r h)
1034-------------------------------------------------------------------------------
1036-------------------------------------------------------------------------------
1038instance MonadTrans Stream where
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list