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 Data.Foldable as F
157import qualified Prelude
161------------------------------------------------------------------------------
162-- The basic stream type
163------------------------------------------------------------------------------
165-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
166-- constructed using actions in monad 'm'. It uses stop, singleton and yield
167-- continuations equivalent to the following direct style type:
170-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
173-- To facilitate parallel composition we maintain a local state in an 'SVar'
174-- that is shared across and is used for synchronization of the streams being
177-- The singleton case can be expressed in terms of stop and yield but we have
178-- it as a separate case to optimize composition operations for streams with
179-- single element. We build singleton streams in the implementation of 'pure'
180-- for Applicative and Monad, and in 'lift' for MonadTrans.
184 unStream :: forall r.
185 State Stream m a -- state
187 -> (a -> m r) -- singleton
188 -> (a -> Stream m a -> m r) -- yield
192instance (Foldable m, Monad m) => Foldable (Stream m) where
193 {-# INLINE foldMap #-}
194 foldMap f = F.fold . toFoldable . map f
196 toFoldable :: (Monad m, Monoid a) => Stream m a -> m a
197 toFoldable = foldr mappend mempty
199instance (Traversable m, Monad m) => Traversable (Stream m) where
200 {-# INLINE traverse #-}
201 traverse f = sequenceA . fmap f
203 {-# INLINE sequenceA #-}
204 sequenceA = traverse id
206-- XXX make this the default "unStream"
207-- | unwraps the Stream type producing the stream function that can be run with
209{-# INLINE unStreamIsolated #-}
212 -> State Stream m a -- state
214 -> (a -> m r) -- singleton
215 -> (a -> Stream m a -> m r) -- yield
217unStreamIsolated x st = unStream x (rstState st)
219{-# INLINE isolateStream #-}
220isolateStream :: Stream m a -> Stream m a
221isolateStream x = Stream $ \st stp sng yld ->
222 unStreamIsolated x st stp sng yld
224-- | Like unstream, but passes a shared SVar across continuations.
225{-# INLINE unstreamShared #-}
228 -> State Stream m a -- state
230 -> (a -> m r) -- singleton
231 -> (a -> Stream m a -> m r) -- yield
233unstreamShared = unStream
235-- Run the stream using a run function associated with the SVar that runs the
236-- streams with a captured snapshot of the monadic state.
237{-# INLINE runStreamSVar #-}
242 -> State Stream m a -- state
244 -> (a -> m r) -- singleton
245 -> (a -> Stream m a -> m r) -- yield
247runStreamSVar sv m st stp sng yld =
248 let mrun = runInIO $ svarMrun sv
249 in void $ liftIO $ mrun $ unStream m st stp sng yld
251------------------------------------------------------------------------------
252-- Types that can behave as a Stream
253------------------------------------------------------------------------------
258-- | Class of types that can represent a stream of elements of some type 'a' in
262class IsStream t where
263 toStream :: t m a -> Stream m a
264 fromStream :: Stream m a -> t m a
265 -- | Constructs a stream by adding a monadic action at the head of an
266 -- existing stream. For example:
269 -- > toList $ getLine \`consM` getLine \`consM` nil
275 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
278 consM :: MonadAsync m => m a -> t m a -> t m a
279 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
280 -- to remember that @|@ comes before ':'.
283 -- > toList $ getLine |: getLine |: nil
290 -- let delay = threadDelay 1000000 >> print 1
291 -- runStream $ serially $ delay |: delay |: delay |: nil
292 -- runStream $ parallely $ delay |: delay |: delay |: nil
295 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
298 (|:) :: MonadAsync m => m a -> t m a -> t m a
299 -- We can define (|:) just as 'consM' but it is defined explicitly for each
300 -- type because we want to use SPECIALIZE pragma on the definition.
302-- | Same as 'IsStream'.
305{-# DEPRECATED Streaming "Please use IsStream instead." #-}
306type Streaming = IsStream
308-------------------------------------------------------------------------------
309-- Type adapting combinators
310-------------------------------------------------------------------------------
312-- | Adapt any specific stream type to any other specific stream type.
315adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
316adapt = fromStream . toStream
318------------------------------------------------------------------------------
320------------------------------------------------------------------------------
322-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
323-- continuation and a yield continuation.
325 => (forall r. State Stream m a
328 -> (a -> t m a -> m r)
331mkStream k = fromStream $ Stream $ \st stp sng yld ->
332 let yieldk a r = yld a (toStream r)
333 in k (rstState st) stp sng yieldk
335------------------------------------------------------------------------------
337------------------------------------------------------------------------------
347nil :: IsStream t => t m a
348nil = fromStream $ Stream $ \_ stp _ _ -> stp
352-- faster than consM because there is no bind.
353-- | Construct a stream by adding a pure value at the head of an existing
354-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
355-- more efficient. For concurrent streams this is not concurrent whereas
356-- 'consM' is concurrent. For example:
359-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
364cons :: IsStream t => a -> t m a -> t m a
365cons a r = fromStream $ Stream $ \_ _ _ yld -> yld a (toStream r)
369-- | Operator equivalent of 'cons'.
372-- > toList $ 1 .: 2 .: 3 .: nil
377(.:) :: IsStream t => a -> t m a -> t m a
380{-# INLINE consMSerial #-}
381consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a
382consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
384------------------------------------------------------------------------------
385-- Asynchronous construction
386------------------------------------------------------------------------------
388-- | Make an empty stream from a callback function.
389nilK :: IsStream t => (forall r. m r -> m r) -> t m a
390nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp
392-- | Make a singleton stream from a one shot callback function.
393yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
394yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng
396-- | Construct a stream from a callback function.
397consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a
398consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))
400-- XXX consK with concurrent callbacks
401-- XXX Build a stream from a repeating callback function.
403-------------------------------------------------------------------------------
405-------------------------------------------------------------------------------
407instance IsStream Stream where
412 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
413 consM :: Monad m => m a -> Stream m a -> Stream m a
417 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
418 (|:) :: Monad m => m a -> Stream m a -> Stream m a
421-------------------------------------------------------------------------------
423-------------------------------------------------------------------------------
426uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
428 let stop = return Nothing
429 single a = return (Just (a, nil))
430 yieldk a r = return (Just (a, fromStream r))
431 in unStream (toStream m) defState stop single yieldk
433-------------------------------------------------------------------------------
435-------------------------------------------------------------------------------
437{-# INLINE unfoldr #-}
438unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
439unfoldr step = fromStream . go
441 go s = Stream $ \_ stp _ yld ->
444 Just (a, b) -> yld a (go b)
446{-# INLINE unfoldrM #-}
447unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
450 go s = fromStream $ Stream $ \svr stp sng yld -> do
455 unStream (toStream (return a |: go b)) svr stp sng yld
457-------------------------------------------------------------------------------
459-------------------------------------------------------------------------------
461yield :: IsStream t => a -> t m a
462yield a = fromStream $ Stream $ \_ _ single _ -> single a
465yieldM :: (Monad m, IsStream t) => m a -> t m a
466yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single
471{-# DEPRECATED once "Please use yieldM instead." #-}
473once :: (Monad m, IsStream t) => m a -> t m a
476-- | Generate an infinite stream by repeating a pure value.
477-- Can be expressed as @cycle1 . yield@.
480repeat :: IsStream t => a -> t m a
481repeat a = let x = cons a x in x
483-------------------------------------------------------------------------------
485-------------------------------------------------------------------------------
487-- | Construct a stream from a 'Foldable' containing pure values. Same as
488-- @'Prelude.foldr' 'cons' 'nil'@.
491{-# INLINE fromFoldable #-}
492fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
493fromFoldable = Prelude.foldr cons nil
495{-# INLINE fromList #-}
496fromList :: IsStream t => [a] -> t m a
497fromList = fromFoldable
499{-# INLINE fromStreamK #-}
500fromStreamK :: Stream m a -> Stream m a
503-------------------------------------------------------------------------------
504-- Elimination by Folding
505-------------------------------------------------------------------------------
507-- | Fold a stream by providing an SVar, a stop continuation, a singleton
508-- continuation and a yield continuation.
514 -> (a -> t m a -> m r)
517foldStream st blank single step m =
518 let yieldk a x = step a (fromStream x)
519 in unStream (toStream m) st blank single yieldk
521-- | Lazy right associative fold.
522foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
523foldr step acc m = go (toStream m)
526 let stop = return acc
527 single a = return (step a acc)
528 yieldk a r = go r >>= \b -> return (step a b)
529 in unStream m1 defState stop single yieldk
531-- | Lazy right fold with a monadic step function.
533foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
534foldrM step acc m = go (toStream m)
537 let stop = return acc
538 single a = step a acc
539 yieldk a r = go r >>= step a
540 in unStream m1 defState stop single yieldk
543foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
547 Nothing -> return Nothing
548 Just (h, t) -> fmap Just (go h (toStream t))
552 single a = return $ step a p
553 yieldk a r = fmap (step p) (go a r)
554 in unStream m1 defState stp single yieldk
556-- | Strict left fold with an extraction function. Like the standard strict
557-- left fold, but applies a user supplied extraction function (the third
558-- argument) to the folded value at the end. This is designed to work with the
559-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
561foldx :: (IsStream t, Monad m)
562 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
563foldx step begin done m = get $ go (toStream m) begin
567 let single = return . done
568 in unStream m1 undefined undefined single undefined
570 -- Note, this can be implemented by making a recursive call to "go",
571 -- however that is more expensive because of unnecessary recursion
572 -- that cannot be tail call optimized. Unfolding recursion explicitly via
573 -- continuations is much more efficient.
574 go m1 !acc = Stream $ \_ _ sng yld ->
576 single a = sng $ step acc a
578 let stream = go r (step acc a)
579 in unStream stream defState undefined sng yld
580 in unStream m1 defState stop single yieldk
582-- | Strict left associative fold.
584foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
585foldl' step begin = foldx step begin id
587-- XXX replace the recursive "go" with explicit continuations.
588-- | Like 'foldx', but with a monadic step function.
589foldxM :: (IsStream t, Monad m)
590 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
591foldxM step begin done m = go begin (toStream m)
594 let stop = acc >>= done
595 single a = acc >>= \b -> step b a >>= done
596 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
597 in unStream m1 defState stop single yieldk
599-- | Like 'foldl'' but with a monadic step function.
600foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
601foldlM' step begin = foldxM step (return begin) return
603------------------------------------------------------------------------------
605------------------------------------------------------------------------------
607{-# INLINE runStream #-}
608runStream :: (Monad m, IsStream t) => t m a -> m ()
609runStream m = go (toStream m)
614 yieldk _ r = go (toStream r)
615 in unStream m1 defState stop single yieldk
618null :: (IsStream t, Monad m) => t m a -> m Bool
620 let stop = return True
621 single _ = return False
622 yieldk _ _ = return False
623 in unStream (toStream m) defState stop single yieldk
626head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
628 let stop = return Nothing
629 single a = return (Just a)
630 yieldk a _ = return (Just a)
631 in unStream (toStream m) defState stop single yieldk
634tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
636 let stop = return Nothing
637 single _ = return $ Just nil
638 yieldk _ r = return $ Just $ fromStream r
639 in unStream (toStream m) defState stop single yieldk
642init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
643init m = go1 (toStream m)
648 Nothing -> return Nothing
649 Just (h, t) -> return . Just . fromStream $ go h t
650 go p m1 = Stream $ \_ stp sng yld ->
652 yieldk a x = yld p $ go a x
653 in unStream m1 defState stp single yieldk
656elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
657elem e m = go (toStream m)
660 let stop = return False
661 single a = return (a == e)
662 yieldk a r = if a == e then return True else go r
663 in unStream m1 defState stop single yieldk
665{-# INLINE notElem #-}
666notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
667notElem e m = go (toStream m)
670 let stop = return True
671 single a = return (a /= e)
672 yieldk a r = if a == e then return False else go r
673 in unStream m1 defState stop single yieldk
675all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
676all p m = go (toStream m)
679 let single a | p a = return True
680 | otherwise = return False
681 yieldk a r | p a = go r
682 | otherwise = return False
683 in unStream m1 defState (return True) single yieldk
685any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
686any p m = go (toStream m)
689 let single a | p a = return True
690 | otherwise = return False
691 yieldk a r | p a = return True
693 in unStream m1 defState (return False) single yieldk
695-- | Extract the last element of the stream, if any.
697last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
698last = foldx (\_ y -> Just y) Nothing id
700{-# INLINE minimum #-}
701minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
702minimum m = go Nothing (toStream m)
705 let stop = return Nothing
706 single a = return (Just a)
707 yieldk a r = go (Just a) r
708 in unStream m1 defState stop single yieldk
711 let stop = return (Just res)
714 then return (Just res)
720 in unStream m1 defState stop single yieldk
722{-# INLINE maximum #-}
723maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
724maximum m = go Nothing (toStream m)
727 let stop = return Nothing
728 single a = return (Just a)
729 yieldk a r = go (Just a) r
730 in unStream m1 defState stop single yieldk
733 let stop = return (Just res)
737 else return (Just res)
742 in unStream m1 defState stop single yieldk
745lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
746lookup e m = go (toStream m)
749 let single (a, b) | a == e = return $ Just b
750 | otherwise = return Nothing
751 yieldk (a, b) x | a == e = return $ Just b
753 in unStream m1 defState (return Nothing) single yieldk
756find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
757find p m = go (toStream m)
760 let single a | p a = return $ Just a
761 | otherwise = return Nothing
762 yieldk a x | p a = return $ Just a
764 in unStream m1 defState (return Nothing) single yieldk
766{-# INLINE findIndices #-}
767findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
768findIndices p = fromStream . go 0 . toStream
770 go offset m1 = Stream $ \st stp sng yld ->
771 let single a | p a = sng offset
773 yieldk a x | p a = yld offset $ go (offset + 1) x
774 | otherwise = unStream (go (offset + 1) x) st stp sng yld
775 in unStream m1 (rstState st) stp single yieldk
777------------------------------------------------------------------------------
779------------------------------------------------------------------------------
781-- | Apply a monadic action to each element of the stream and discard the
782-- output of the action.
783mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
784mapM_ f m = go (toStream m)
788 single a = void (f a)
789 yieldk a r = f a >> go r
790 in unStream m1 defState stop single yieldk
792------------------------------------------------------------------------------
794------------------------------------------------------------------------------
796{-# INLINABLE toList #-}
797toList :: (IsStream t, Monad m) => t m a -> m [a]
800{-# INLINE toStreamK #-}
801toStreamK :: Stream m a -> Stream m a
804-------------------------------------------------------------------------------
805-- Transformation by folding (Scans)
806-------------------------------------------------------------------------------
809scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
810scanx step begin done m =
811 cons (done begin) $ fromStream $ go (toStream m) begin
813 go m1 !acc = Stream $ \st stp sng yld ->
814 let single a = sng (done $ step acc a)
817 in yld (done s) (go r s)
818 in unStream m1 (rstState st) stp single yieldk
821scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
822scanl' step begin = scanx step begin id
824-------------------------------------------------------------------------------
826-------------------------------------------------------------------------------
829filter :: IsStream t => (a -> Bool) -> t m a -> t m a
830filter p m = fromStream $ go (toStream m)
832 go m1 = Stream $ \st stp sng yld ->
833 let single a | p a = sng a
835 yieldk a r | p a = yld a (go r)
836 | otherwise = unStream r (rstState st) stp single yieldk
837 in unStream m1 (rstState st) stp single yieldk
840take :: IsStream t => Int -> t m a -> t m a
841take n m = fromStream $ go n (toStream m)
843 go n1 m1 = Stream $ \st stp sng yld ->
844 let yieldk a r = yld a (go (n1 - 1) r)
847 else unStream m1 (rstState st) stp sng yieldk
849{-# INLINE takeWhile #-}
850takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
851takeWhile p m = fromStream $ go (toStream m)
853 go m1 = Stream $ \st stp sng yld ->
854 let single a | p a = sng a
856 yieldk a r | p a = yld a (go r)
858 in unStream m1 (rstState st) stp single yieldk
860drop :: IsStream t => Int -> t m a -> t m a
861drop n m = fromStream $ Stream $ \st stp sng yld ->
862 unStream (go n (toStream m)) (rstState st) stp sng yld
864 go n1 m1 = Stream $ \st stp sng yld ->
866 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
867 -- Somehow "<=" check performs better than a ">"
869 then unStream m1 st stp sng yld
870 else unStream m1 st stp single yieldk
872{-# INLINE dropWhile #-}
873dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
874dropWhile p m = fromStream $ go (toStream m)
876 go m1 = Stream $ \st stp sng yld ->
877 let single a | p a = stp
879 yieldk a r | p a = unStream r (rstState st) stp single yieldk
880 | otherwise = yld a r
881 in unStream m1 (rstState st) stp single yieldk
883-------------------------------------------------------------------------------
885-------------------------------------------------------------------------------
888map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
889map f m = fromStream $ Stream $ \st stp sng yld ->
891 yieldk a r = yld (f a) (fmap f r)
892 in unStream (toStream m) (rstState st) stp single yieldk
894-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
895-- other stream types to overload it.
897mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
898mapM f m = go (toStream m)
900 go m1 = fromStream $ Stream $ \st stp sng yld ->
901 let single a = f a >>= sng
902 yieldk a r = unStream (toStream (f a |: go r)) st stp sng yld
903 in unStream m1 (rstState st) stp single yieldk
905-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
906-- other stream types to overload it.
907{-# INLINE sequence #-}
908sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
909sequence m = go (toStream m)
911 go m1 = fromStream $ Stream $ \st stp sng yld ->
912 let single ma = ma >>= sng
913 yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld
914 in unStream m1 (rstState st) stp single yieldk
916-------------------------------------------------------------------------------
918-------------------------------------------------------------------------------
920{-# INLINE intersperseM #-}
921intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
922intersperseM a m = fromStream $ prependingStart (toStream m)
924 prependingStart m1 = Stream $ \st stp sng yld ->
925 let yieldk i x = unStream (return i |: go x) st stp sng yld
926 in unStream m1 (rstState st) stp sng yieldk
927 go m2 = fromStream $ Stream $ \st stp sng yld ->
928 let single i = unStream (a |: yield i) st stp sng yld
929 yieldk i x = unStream (a |: return i |: go x) st stp sng yld
930 in unStream m2 (rstState st) stp single yieldk
932-------------------------------------------------------------------------------
934-------------------------------------------------------------------------------
936{-# INLINE mapMaybe #-}
937mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
938mapMaybe f m = go (toStream m)
940 go m1 = fromStream $ Stream $ \st stp sng yld ->
941 let single a = case f a of
944 yieldk a r = case f a of
945 Just b -> yld b (toStream $ go r)
946 Nothing -> unStream r (rstState st) stp single yieldk
947 in unStream m1 (rstState st) stp single yieldk
949------------------------------------------------------------------------------
951------------------------------------------------------------------------------
953{-# INLINE zipWithS #-}
954zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
957 go mx my = Stream $ \st stp sng yld -> do
959 let single2 b = sng (f a b)
960 yield2 b rb = yld (f a b) (go ra rb)
961 in unStream my (rstState st) stp single2 yield2
962 let single1 a = merge a nil
964 unStream mx (rstState st) stp single1 yield1
966-- | Zip two streams serially using a pure zipping function.
969{-# INLINABLE zipWith #-}
970zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
971zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)
973-- | Zip two streams serially using a monadic zipping function.
976zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
977zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
979 go mx my = Stream $ \st stp sng yld -> do
981 let runIt x = unStream x (rstState st) stp sng yld
982 single2 b = f a b >>= sng
983 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
984 in unStream my (rstState st) stp single2 yield2
985 let single1 a = merge a nil
987 unStream mx (rstState st) stp single1 yield1
989------------------------------------------------------------------------------
991------------------------------------------------------------------------------
993-- | Concatenates two streams sequentially i.e. the first stream is
994-- exhausted completely before yielding any element from the second stream.
996serial :: Stream m a -> Stream m a -> Stream m a
999 go (Stream m) = Stream $ \st stp sng yld ->
1000 let stop = unStream m2 (rstState st) stp sng yld
1002 yieldk a r = yld a (go r)
1003 in m (rstState st) stop single yieldk
1005instance Semigroup (Stream m a) where
1008------------------------------------------------------------------------------
1010------------------------------------------------------------------------------
1012instance Monoid (Stream m a) where
1016-------------------------------------------------------------------------------
1018-------------------------------------------------------------------------------
1020instance Monad m => Functor (Stream m) where
1023-------------------------------------------------------------------------------
1025-------------------------------------------------------------------------------
1027{-# INLINE bindWith #-}
1029 :: (forall c. Stream m c -> Stream m c -> Stream m c)
1031 -> (a -> Stream m b)
1033bindWith par m f = go m
1036 Stream $ \st stp sng yld ->
1037 let runShared x = unstreamShared x st stp sng yld
1038 runIsolated x = unStreamIsolated x st stp sng yld
1040 single a = runIsolated $ f a
1041 yieldk a r = runShared $ isolateStream (f a) `par` go r
1042 in g (rstState st) stp single yieldk
1044------------------------------------------------------------------------------
1045-- Alternative & MonadPlus
1046------------------------------------------------------------------------------
1048_alt :: Stream m a -> Stream m a -> Stream m a
1049_alt m1 m2 = Stream $ \st stp sng yld ->
1050 let stop = unStream m2 (rstState st) stp sng yld
1051 in unStream m1 (rstState st) stop sng yld
1053------------------------------------------------------------------------------
1055------------------------------------------------------------------------------
1057withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1059 Stream $ \st stp sng yld ->
1060 let single = local f . sng
1061 yieldk a r = local f $ yld a (withLocal f r)
1062 in unStream m (rstState st) (local f stp) single yieldk
1064------------------------------------------------------------------------------
1066------------------------------------------------------------------------------
1069-- XXX handle and test cross thread state transfer
1072 => Stream m a -> (e -> Stream m a) -> Stream m a
1074 Stream $ \_ stp sng yld ->
1075 let run x = unStream x Nothing stp sng yieldk
1076 handle r = r `catchError` \e -> run $ h e
1077 yieldk a r = yld a (withCatchError r h)
1081-------------------------------------------------------------------------------
1083-------------------------------------------------------------------------------
1085instance 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