1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE FlexibleInstances #-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
9{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- Module : Streamly.Streams.StreamK
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
21-- Continuation passing style (CPS) stream implementation. The symbol 'K' below
22-- denotes a function as well as a Kontinuation.
25-- import qualified Streamly.Streams.StreamK as K
28module Streamly.Streams.StreamK
30 -- * A class for streams
47 -- * Asynchronous construction
60 -- ** Specialized Generation
81 -- ** Specialized Folds
106 -- ** By folding (scans)
132 -- * Semigroup Style Composition
141 , Streaming -- deprecated
146import Control.Monad (void)
147import Control.Monad.IO.Class (MonadIO(liftIO))
148import Control.Monad.Reader.Class (MonadReader(..))
149import Control.Monad.Trans.Class (MonadTrans(lift))
150import Data.Semigroup (Semigroup(..))
152 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
153 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
154 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,
156import qualified Prelude
160------------------------------------------------------------------------------
161-- The basic stream type
162------------------------------------------------------------------------------
164-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
165-- constructed using actions in monad 'm'. It uses stop, singleton and yield
166-- continuations equivalent to the following direct style type:
169-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
172-- To facilitate parallel composition we maintain a local state in an 'SVar'
173-- that is shared across and is used for synchronization of the streams being
176-- The singleton case can be expressed in terms of stop and yield but we have
177-- it as a separate case to optimize composition operations for streams with
178-- single element. We build singleton streams in the implementation of 'pure'
179-- for Applicative and Monad, and in 'lift' for MonadTrans.
183 unStream :: forall r.
184 State Stream m a -- state
186 -> (a -> m r) -- singleton
187 -> (a -> Stream m a -> m r) -- yield
191-- XXX make this the default "unStream"
192-- | unwraps the Stream type producing the stream function that can be run with
194{-# INLINE unStreamIsolated #-}
197 -> State Stream m a -- state
199 -> (a -> m r) -- singleton
200 -> (a -> Stream m a -> m r) -- yield
202unStreamIsolated x st = unStream x (rstState st)
204{-# INLINE isolateStream #-}
205isolateStream :: Stream m a -> Stream m a
206isolateStream x = Stream $ \st stp sng yld ->
207 unStreamIsolated x st stp sng yld
209-- | Like unstream, but passes a shared SVar across continuations.
210{-# INLINE unstreamShared #-}
213 -> State Stream m a -- state
215 -> (a -> m r) -- singleton
216 -> (a -> Stream m a -> m r) -- yield
218unstreamShared = unStream
220-- Run the stream using a run function associated with the SVar that runs the
221-- streams with a captured snapshot of the monadic state.
222{-# INLINE runStreamSVar #-}
227 -> State Stream m a -- state
229 -> (a -> m r) -- singleton
230 -> (a -> Stream m a -> m r) -- yield
232runStreamSVar sv m st stp sng yld =
233 let mrun = runInIO $ svarMrun sv
234 in void $ liftIO $ mrun $ unStream m st stp sng yld
236------------------------------------------------------------------------------
237-- Types that can behave as a Stream
238------------------------------------------------------------------------------
243-- | Class of types that can represent a stream of elements of some type 'a' in
247class IsStream t where
248 toStream :: t m a -> Stream m a
249 fromStream :: Stream m a -> t m a
250 -- | Constructs a stream by adding a monadic action at the head of an
251 -- existing stream. For example:
254 -- > toList $ getLine \`consM` getLine \`consM` nil
260 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
263 consM :: MonadAsync m => m a -> t m a -> t m a
264 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
265 -- to remember that @|@ comes before ':'.
268 -- > toList $ getLine |: getLine |: nil
275 -- let delay = threadDelay 1000000 >> print 1
276 -- runStream $ serially $ delay |: delay |: delay |: nil
277 -- runStream $ parallely $ delay |: delay |: delay |: nil
280 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
283 (|:) :: MonadAsync m => m a -> t m a -> t m a
284 -- We can define (|:) just as 'consM' but it is defined explicitly for each
285 -- type because we want to use SPECIALIZE pragma on the definition.
287-- | Same as 'IsStream'.
290{-# DEPRECATED Streaming "Please use IsStream instead." #-}
291type Streaming = IsStream
293-------------------------------------------------------------------------------
294-- Type adapting combinators
295-------------------------------------------------------------------------------
297-- | Adapt any specific stream type to any other specific stream type.
300adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
301adapt = fromStream . toStream
303------------------------------------------------------------------------------
305------------------------------------------------------------------------------
307-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
308-- continuation and a yield continuation.
310 => (forall r. State Stream m a
313 -> (a -> t m a -> m r)
316mkStream k = fromStream $ Stream $ \st stp sng yld ->
317 let yieldk a r = yld a (toStream r)
318 in k (rstState st) stp sng yieldk
320------------------------------------------------------------------------------
322------------------------------------------------------------------------------
332nil :: IsStream t => t m a
333nil = fromStream $ Stream $ \_ stp _ _ -> stp
337-- faster than consM because there is no bind.
338-- | Construct a stream by adding a pure value at the head of an existing
339-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
340-- more efficient. For concurrent streams this is not concurrent whereas
341-- 'consM' is concurrent. For example:
344-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
349cons :: IsStream t => a -> t m a -> t m a
350cons a r = fromStream $ Stream $ \_ _ _ yld -> yld a (toStream r)
354-- | Operator equivalent of 'cons'.
357-- > toList $ 1 .: 2 .: 3 .: nil
362(.:) :: IsStream t => a -> t m a -> t m a
365{-# INLINE consMSerial #-}
366consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a
367consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
369------------------------------------------------------------------------------
370-- Asynchronous construction
371------------------------------------------------------------------------------
373-- | Make an empty stream from a callback function.
374nilK :: IsStream t => (forall r. m r -> m r) -> t m a
375nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp
377-- | Make a singleton stream from a one shot callback function.
378yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
379yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng
381-- | Construct a stream from a callback function.
382consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a
383consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))
385-- XXX consK with concurrent callbacks
386-- XXX Build a stream from a repeating callback function.
388-------------------------------------------------------------------------------
390-------------------------------------------------------------------------------
392instance IsStream Stream where
397 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
398 consM :: Monad m => m a -> Stream m a -> Stream m a
402 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
403 (|:) :: Monad m => m a -> Stream m a -> Stream m a
406-------------------------------------------------------------------------------
408-------------------------------------------------------------------------------
411uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
413 let stop = return Nothing
414 single a = return (Just (a, nil))
415 yieldk a r = return (Just (a, fromStream r))
416 in unStream (toStream m) defState stop single yieldk
418-------------------------------------------------------------------------------
420-------------------------------------------------------------------------------
422{-# INLINE unfoldr #-}
423unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
424unfoldr step = fromStream . go
426 go s = Stream $ \_ stp _ yld ->
429 Just (a, b) -> yld a (go b)
431{-# INLINE unfoldrM #-}
432unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
435 go s = fromStream $ Stream $ \svr stp sng yld -> do
440 unStream (toStream (return a |: go b)) svr stp sng yld
442-------------------------------------------------------------------------------
444-------------------------------------------------------------------------------
446yield :: IsStream t => a -> t m a
447yield a = fromStream $ Stream $ \_ _ single _ -> single a
450yieldM :: (Monad m, IsStream t) => m a -> t m a
451yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single
456{-# DEPRECATED once "Please use yieldM instead." #-}
458once :: (Monad m, IsStream t) => m a -> t m a
461-- | Generate an infinite stream by repeating a pure value.
462-- Can be expressed as @cycle1 . yield@.
465repeat :: IsStream t => a -> t m a
466repeat a = let x = cons a x in x
468-------------------------------------------------------------------------------
470-------------------------------------------------------------------------------
472-- | Construct a stream from a 'Foldable' containing pure values. Same as
473-- @'Prelude.foldr' 'cons' 'nil'@.
476{-# INLINE fromFoldable #-}
477fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
478fromFoldable = Prelude.foldr cons nil
480{-# INLINE fromList #-}
481fromList :: IsStream t => [a] -> t m a
482fromList = fromFoldable
484{-# INLINE fromStreamK #-}
485fromStreamK :: Stream m a -> Stream m a
488-------------------------------------------------------------------------------
489-- Elimination by Folding
490-------------------------------------------------------------------------------
492-- | Fold a stream by providing an SVar, a stop continuation, a singleton
493-- continuation and a yield continuation.
499 -> (a -> t m a -> m r)
502foldStream st blank single step m =
503 let yieldk a x = step a (fromStream x)
504 in unStream (toStream m) st blank single yieldk
506-- | Lazy right associative fold.
507foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
508foldr step acc m = go (toStream m)
511 let stop = return acc
512 single a = return (step a acc)
513 yieldk a r = go r >>= \b -> return (step a b)
514 in unStream m1 defState stop single yieldk
516-- | Lazy right fold with a monadic step function.
518foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
519foldrM step acc m = go (toStream m)
522 let stop = return acc
523 single a = step a acc
524 yieldk a r = go r >>= step a
525 in unStream m1 defState stop single yieldk
528foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
532 Nothing -> return Nothing
533 Just (h, t) -> fmap Just (go h (toStream t))
537 single a = return $ step a p
538 yieldk a r = fmap (step p) (go a r)
539 in unStream m1 defState stp single yieldk
541-- | Strict left fold with an extraction function. Like the standard strict
542-- left fold, but applies a user supplied extraction function (the third
543-- argument) to the folded value at the end. This is designed to work with the
544-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
546foldx :: (IsStream t, Monad m)
547 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
548foldx step begin done m = get $ go (toStream m) begin
552 let single = return . done
553 in unStream m1 undefined undefined single undefined
555 -- Note, this can be implemented by making a recursive call to "go",
556 -- however that is more expensive because of unnecessary recursion
557 -- that cannot be tail call optimized. Unfolding recursion explicitly via
558 -- continuations is much more efficient.
559 go m1 !acc = Stream $ \_ _ sng yld ->
561 single a = sng $ step acc a
563 let stream = go r (step acc a)
564 in unStream stream defState undefined sng yld
565 in unStream m1 defState stop single yieldk
567-- | Strict left associative fold.
569foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
570foldl' step begin = foldx step begin id
572-- XXX replace the recursive "go" with explicit continuations.
573-- | Like 'foldx', but with a monadic step function.
574foldxM :: (IsStream t, Monad m)
575 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
576foldxM step begin done m = go begin (toStream m)
579 let stop = acc >>= done
580 single a = acc >>= \b -> step b a >>= done
581 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
582 in unStream m1 defState stop single yieldk
584-- | Like 'foldl'' but with a monadic step function.
585foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
586foldlM' step begin = foldxM step (return begin) return
588------------------------------------------------------------------------------
590------------------------------------------------------------------------------
592{-# INLINE runStream #-}
593runStream :: (Monad m, IsStream t) => t m a -> m ()
594runStream m = go (toStream m)
599 yieldk _ r = go (toStream r)
600 in unStream m1 defState stop single yieldk
603null :: (IsStream t, Monad m) => t m a -> m Bool
605 let stop = return True
606 single _ = return False
607 yieldk _ _ = return False
608 in unStream (toStream m) defState stop single yieldk
611head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
613 let stop = return Nothing
614 single a = return (Just a)
615 yieldk a _ = return (Just a)
616 in unStream (toStream m) defState stop single yieldk
619tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
621 let stop = return Nothing
622 single _ = return $ Just nil
623 yieldk _ r = return $ Just $ fromStream r
624 in unStream (toStream m) defState stop single yieldk
627init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
628init m = go1 (toStream m)
633 Nothing -> return Nothing
634 Just (h, t) -> return . Just . fromStream $ go h t
635 go p m1 = Stream $ \_ stp sng yld ->
637 yieldk a x = yld p $ go a x
638 in unStream m1 defState stp single yieldk
641elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
642elem e m = go (toStream m)
645 let stop = return False
646 single a = return (a == e)
647 yieldk a r = if a == e then return True else go r
648 in unStream m1 defState stop single yieldk
650{-# INLINE notElem #-}
651notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
652notElem e m = go (toStream m)
655 let stop = return True
656 single a = return (a /= e)
657 yieldk a r = if a == e then return False else go r
658 in unStream m1 defState stop single yieldk
660all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
661all p m = go (toStream m)
664 let single a | p a = return True
665 | otherwise = return False
666 yieldk a r | p a = go r
667 | otherwise = return False
668 in unStream m1 defState (return True) single yieldk
670any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
671any p m = go (toStream m)
674 let single a | p a = return True
675 | otherwise = return False
676 yieldk a r | p a = return True
678 in unStream m1 defState (return False) single yieldk
680-- | Extract the last element of the stream, if any.
682last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
683last = foldx (\_ y -> Just y) Nothing id
685{-# INLINE minimum #-}
686minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
687minimum m = go Nothing (toStream m)
690 let stop = return Nothing
691 single a = return (Just a)
692 yieldk a r = go (Just a) r
693 in unStream m1 defState stop single yieldk
696 let stop = return (Just res)
699 then return (Just res)
705 in unStream m1 defState stop single yieldk
707{-# INLINE maximum #-}
708maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
709maximum m = go Nothing (toStream m)
712 let stop = return Nothing
713 single a = return (Just a)
714 yieldk a r = go (Just a) r
715 in unStream m1 defState stop single yieldk
718 let stop = return (Just res)
722 else return (Just res)
727 in unStream m1 defState stop single yieldk
730lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
731lookup e m = go (toStream m)
734 let single (a, b) | a == e = return $ Just b
735 | otherwise = return Nothing
736 yieldk (a, b) x | a == e = return $ Just b
738 in unStream m1 defState (return Nothing) single yieldk
741find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
742find p m = go (toStream m)
745 let single a | p a = return $ Just a
746 | otherwise = return Nothing
747 yieldk a x | p a = return $ Just a
749 in unStream m1 defState (return Nothing) single yieldk
751{-# INLINE findIndices #-}
752findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
753findIndices p = fromStream . go 0 . toStream
755 go offset m1 = Stream $ \st stp sng yld ->
756 let single a | p a = sng offset
758 yieldk a x | p a = yld offset $ go (offset + 1) x
759 | otherwise = unStream (go (offset + 1) x) st stp sng yld
760 in unStream m1 (rstState st) stp single yieldk
762------------------------------------------------------------------------------
764------------------------------------------------------------------------------
766-- | Apply a monadic action to each element of the stream and discard the
767-- output of the action.
768mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
769mapM_ f m = go (toStream m)
773 single a = void (f a)
774 yieldk a r = f a >> go r
775 in unStream m1 defState stop single yieldk
777------------------------------------------------------------------------------
779------------------------------------------------------------------------------
781{-# INLINABLE toList #-}
782toList :: (IsStream t, Monad m) => t m a -> m [a]
785{-# INLINE toStreamK #-}
786toStreamK :: Stream m a -> Stream m a
789-------------------------------------------------------------------------------
790-- Transformation by folding (Scans)
791-------------------------------------------------------------------------------
794scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
795scanx step begin done m =
796 cons (done begin) $ fromStream $ go (toStream m) begin
798 go m1 !acc = Stream $ \st stp sng yld ->
799 let single a = sng (done $ step acc a)
802 in yld (done s) (go r s)
803 in unStream m1 (rstState st) stp single yieldk
806scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
807scanl' step begin = scanx step begin id
809-------------------------------------------------------------------------------
811-------------------------------------------------------------------------------
814filter :: IsStream t => (a -> Bool) -> t m a -> t m a
815filter p m = fromStream $ go (toStream m)
817 go m1 = Stream $ \st stp sng yld ->
818 let single a | p a = sng a
820 yieldk a r | p a = yld a (go r)
821 | otherwise = unStream r (rstState st) stp single yieldk
822 in unStream m1 (rstState st) stp single yieldk
825take :: IsStream t => Int -> t m a -> t m a
826take n m = fromStream $ go n (toStream m)
828 go n1 m1 = Stream $ \st stp sng yld ->
829 let yieldk a r = yld a (go (n1 - 1) r)
832 else unStream m1 (rstState st) stp sng yieldk
834{-# INLINE takeWhile #-}
835takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
836takeWhile p m = fromStream $ go (toStream m)
838 go m1 = Stream $ \st stp sng yld ->
839 let single a | p a = sng a
841 yieldk a r | p a = yld a (go r)
843 in unStream m1 (rstState st) stp single yieldk
845drop :: IsStream t => Int -> t m a -> t m a
846drop n m = fromStream $ Stream $ \st stp sng yld ->
847 unStream (go n (toStream m)) (rstState st) stp sng yld
849 go n1 m1 = Stream $ \st stp sng yld ->
851 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
852 -- Somehow "<=" check performs better than a ">"
854 then unStream m1 st stp sng yld
855 else unStream m1 st stp single yieldk
857{-# INLINE dropWhile #-}
858dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
859dropWhile p m = fromStream $ go (toStream m)
861 go m1 = Stream $ \st stp sng yld ->
862 let single a | p a = stp
864 yieldk a r | p a = unStream r (rstState st) stp single yieldk
865 | otherwise = yld a r
866 in unStream m1 (rstState st) stp single yieldk
868-------------------------------------------------------------------------------
870-------------------------------------------------------------------------------
873map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
874map f m = fromStream $ Stream $ \st stp sng yld ->
876 yieldk a r = yld (f a) (fmap f r)
877 in unStream (toStream m) (rstState st) stp single yieldk
879-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
880-- other stream types to overload it.
882mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
883mapM f m = go (toStream m)
885 go m1 = fromStream $ Stream $ \st stp sng yld ->
886 let single a = f a >>= sng
887 yieldk a r = unStream (toStream (f a |: go r)) st stp sng yld
888 in unStream m1 (rstState st) stp single yieldk
890-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
891-- other stream types to overload it.
892{-# INLINE sequence #-}
893sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
894sequence m = go (toStream m)
896 go m1 = fromStream $ Stream $ \st stp sng yld ->
897 let single ma = ma >>= sng
898 yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld
899 in unStream m1 (rstState st) stp single yieldk
901-------------------------------------------------------------------------------
903-------------------------------------------------------------------------------
905{-# INLINE intersperseM #-}
906intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
907intersperseM a m = fromStream $ prependingStart (toStream m)
909 prependingStart m1 = Stream $ \st stp sng yld ->
910 let yieldk i x = unStream (return i |: go x) st stp sng yld
911 in unStream m1 (rstState st) stp sng yieldk
912 go m2 = fromStream $ Stream $ \st stp sng yld ->
913 let single i = unStream (a |: yield i) st stp sng yld
914 yieldk i x = unStream (a |: return i |: go x) st stp sng yld
915 in unStream m2 (rstState st) stp single yieldk
917-------------------------------------------------------------------------------
919-------------------------------------------------------------------------------
921{-# INLINE mapMaybe #-}
922mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
923mapMaybe f m = go (toStream m)
925 go m1 = fromStream $ Stream $ \st stp sng yld ->
926 let single a = case f a of
929 yieldk a r = case f a of
930 Just b -> yld b (toStream $ go r)
931 Nothing -> unStream r (rstState st) stp single yieldk
932 in unStream m1 (rstState st) stp single yieldk
934------------------------------------------------------------------------------
936------------------------------------------------------------------------------
938{-# INLINE zipWithS #-}
939zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
942 go mx my = Stream $ \st stp sng yld -> do
944 let single2 b = sng (f a b)
945 yield2 b rb = yld (f a b) (go ra rb)
946 in unStream my (rstState st) stp single2 yield2
947 let single1 a = merge a nil
949 unStream mx (rstState st) stp single1 yield1
951-- | Zip two streams serially using a pure zipping function.
954{-# INLINABLE zipWith #-}
955zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
956zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)
958-- | Zip two streams serially using a monadic zipping function.
961zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
962zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
964 go mx my = Stream $ \st stp sng yld -> do
966 let runIt x = unStream x (rstState st) stp sng yld
967 single2 b = f a b >>= sng
968 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
969 in unStream my (rstState st) stp single2 yield2
970 let single1 a = merge a nil
972 unStream mx (rstState st) stp single1 yield1
974------------------------------------------------------------------------------
976------------------------------------------------------------------------------
978-- | Concatenates two streams sequentially i.e. the first stream is
979-- exhausted completely before yielding any element from the second stream.
981serial :: Stream m a -> Stream m a -> Stream m a
984 go (Stream m) = Stream $ \st stp sng yld ->
985 let stop = unStream m2 (rstState st) stp sng yld
987 yieldk a r = yld a (go r)
988 in m (rstState st) stop single yieldk
990instance Semigroup (Stream m a) where
993------------------------------------------------------------------------------
995------------------------------------------------------------------------------
997instance Monoid (Stream m a) where
1001-------------------------------------------------------------------------------
1003-------------------------------------------------------------------------------
1005instance Monad m => Functor (Stream m) where
1008-------------------------------------------------------------------------------
1010-------------------------------------------------------------------------------
1012{-# INLINE bindWith #-}
1014 :: (forall c. Stream m c -> Stream m c -> Stream m c)
1016 -> (a -> Stream m b)
1018bindWith par m f = go m
1021 Stream $ \st stp sng yld ->
1022 let runShared x = unstreamShared x st stp sng yld
1023 runIsolated x = unStreamIsolated x st stp sng yld
1025 single a = runIsolated $ f a
1026 yieldk a r = runShared $ isolateStream (f a) `par` go r
1027 in g (rstState st) stp single yieldk
1029------------------------------------------------------------------------------
1030-- Alternative & MonadPlus
1031------------------------------------------------------------------------------
1033_alt :: Stream m a -> Stream m a -> Stream m a
1034_alt m1 m2 = Stream $ \st stp sng yld ->
1035 let stop = unStream m2 (rstState st) stp sng yld
1036 in unStream m1 (rstState st) stop sng yld
1038------------------------------------------------------------------------------
1040------------------------------------------------------------------------------
1042withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1044 Stream $ \st stp sng yld ->
1045 let single = local f . sng
1046 yieldk a r = local f $ yld a (withLocal f r)
1047 in unStream m (rstState st) (local f stp) single yieldk
1049------------------------------------------------------------------------------
1051------------------------------------------------------------------------------
1054-- XXX handle and test cross thread state transfer
1057 => Stream m a -> (e -> Stream m a) -> Stream m a
1059 Stream $ \_ stp sng yld ->
1060 let run x = unStream x Nothing stp sng yieldk
1061 handle r = r `catchError` \e -> run $ h e
1062 yieldk a r = yld a (withCatchError r h)
1066-------------------------------------------------------------------------------
1068-------------------------------------------------------------------------------
1070instance 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