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
83 -- ** Specialized Folds
112 -- ** By folding (scans)
146 -- ** Transformation comprehensions
149 -- * Semigroup Style Composition
158 , Streaming -- deprecated
163import Control.Monad (void)
164import Control.Monad.IO.Class (MonadIO(liftIO))
165import Control.Monad.Reader.Class (MonadReader(..))
166import Control.Monad.Trans.Class (MonadTrans(lift))
167import Data.Semigroup (Semigroup(..))
169 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
170 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
171 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,
172 foldr1, (!!), replicate)
173import qualified Prelude
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 State Stream m a -- state
202 -> (a -> m r) -- singleton
203 -> (a -> Stream m a -> m r) -- yield
207-- Run a stream in detached mode i.e. not joining an SVar
208{-# INLINE unStream #-}
211 -> State Stream m b -- state
213 -> (a -> m r) -- singleton
214 -> (a -> Stream m a -> m r) -- yield
216unStream (Stream runner) st = runner (adaptState st)
218-- | Detach a stream from an SVar
219{-# INLINE unShare #-}
220unShare :: Stream m a -> Stream m a
221unShare x = Stream $ \st stp sng yld ->
222 unStream 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 (Stream runner) st stp sng yld = runner st stp sng yld
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 $ unStreamShared 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 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 unStreamShared (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
478-- repeatM = fix . cons
479-- repeatM = cycle1 . yield
482-- Generate an infinite stream by repeating a pure value.
485repeat :: IsStream t => a -> t m a
486repeat a = let x = cons a x in x
488replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
491 go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)
493replicate :: IsStream t => Int -> a -> t m a
496 go cnt = if cnt <= 0 then nil else a `cons` go (cnt - 1)
498-------------------------------------------------------------------------------
500-------------------------------------------------------------------------------
504-- fromFoldable = 'Prelude.foldr' 'cons' 'nil'
507-- Construct a stream from a 'Foldable' containing pure values:
510{-# INLINE fromFoldable #-}
511fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
512fromFoldable = Prelude.foldr cons nil
514{-# INLINE fromList #-}
515fromList :: IsStream t => [a] -> t m a
516fromList = fromFoldable
518{-# INLINE fromStreamK #-}
519fromStreamK :: Stream m a -> Stream m a
522-------------------------------------------------------------------------------
523-- Elimination by Folding
524-------------------------------------------------------------------------------
526-- | Fold a stream by providing an SVar, a stop continuation, a singleton
527-- continuation and a yield continuation.
533 -> (a -> t m a -> m r)
536foldStream st blank single step m =
537 let yieldk a x = step a (fromStream x)
538 in unStreamShared (toStream m) st blank single yieldk
540-- | Lazy right associative fold.
541foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
542foldr step acc m = go (toStream m)
545 let stop = return acc
546 single a = return (step a acc)
547 yieldk a r = go r >>= \b -> return (step a b)
548 in unStream m1 defState stop single yieldk
550-- | Lazy right fold with a monadic step function.
552foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
553foldrM step acc m = go (toStream m)
556 let stop = return acc
557 single a = step a acc
558 yieldk a r = go r >>= step a
559 in unStream m1 defState stop single yieldk
562foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
566 Nothing -> return Nothing
567 Just (h, t) -> fmap Just (go h (toStream t))
571 single a = return $ step a p
572 yieldk a r = fmap (step p) (go a r)
573 in unStream m1 defState stp single yieldk
575-- | Strict left fold with an extraction function. Like the standard strict
576-- left fold, but applies a user supplied extraction function (the third
577-- argument) to the folded value at the end. This is designed to work with the
578-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
580-- Note that the accumulator is always evaluated including the initial value.
582foldx :: (IsStream t, Monad m)
583 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
584foldx step begin done m = get $ go (toStream m) begin
588 -- XXX we are not strictly evaluating the accumulator here. Is this
590 let single = return . done
591 in unStreamShared m1 undefined undefined single undefined
593 -- Note, this can be implemented by making a recursive call to "go",
594 -- however that is more expensive because of unnecessary recursion
595 -- that cannot be tail call optimized. Unfolding recursion explicitly via
596 -- continuations is much more efficient.
597 go m1 !acc = Stream $ \_ _ sng yld ->
599 single a = sng $ step acc a
601 let stream = go r (step acc a)
602 in unStream stream defState undefined sng yld
603 in unStream m1 defState stop single yieldk
605-- | Strict left associative fold.
607foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
608foldl' step begin = foldx step begin id
610-- XXX replace the recursive "go" with explicit continuations.
611-- | Like 'foldx', but with a monadic step function.
612foldxM :: (IsStream t, Monad m)
613 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
614foldxM step begin done m = go begin (toStream m)
617 let stop = acc >>= done
618 single a = acc >>= \b -> step b a >>= done
619 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
620 in unStream m1 defState stop single yieldk
622-- | Like 'foldl'' but with a monadic step function.
623foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
624foldlM' step begin = foldxM step (return begin) return
626------------------------------------------------------------------------------
628------------------------------------------------------------------------------
630{-# INLINE runStream #-}
631runStream :: (Monad m, IsStream t) => t m a -> m ()
632runStream m = go (toStream m)
637 yieldk _ r = go (toStream r)
638 in unStream m1 defState stop single yieldk
641null :: (IsStream t, Monad m) => t m a -> m Bool
643 let stop = return True
644 single _ = return False
645 yieldk _ _ = return False
646 in unStream (toStream m) defState stop single yieldk
649head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
651 let stop = return Nothing
652 single a = return (Just a)
653 yieldk a _ = return (Just a)
654 in unStream (toStream m) defState stop single yieldk
657tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
659 let stop = return Nothing
660 single _ = return $ Just nil
661 yieldk _ r = return $ Just $ fromStream r
662 in unStream (toStream m) defState stop single yieldk
665init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
666init m = go1 (toStream m)
671 Nothing -> return Nothing
672 Just (h, t) -> return . Just . fromStream $ go h t
673 go p m1 = Stream $ \_ stp sng yld ->
675 yieldk a x = yld p $ go a x
676 in unStream m1 defState stp single yieldk
679elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
680elem e m = go (toStream m)
683 let stop = return False
684 single a = return (a == e)
685 yieldk a r = if a == e then return True else go r
686 in unStream m1 defState stop single yieldk
688{-# INLINE notElem #-}
689notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
690notElem e m = go (toStream m)
693 let stop = return True
694 single a = return (a /= e)
695 yieldk a r = if a == e then return False else go r
696 in unStream m1 defState stop single yieldk
698all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
699all p m = go (toStream m)
702 let single a | p a = return True
703 | otherwise = return False
704 yieldk a r | p a = go r
705 | otherwise = return False
706 in unStream m1 defState (return True) single yieldk
708any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
709any p m = go (toStream m)
712 let single a | p a = return True
713 | otherwise = return False
714 yieldk a r | p a = return True
716 in unStream m1 defState (return False) single yieldk
718-- | Extract the last element of the stream, if any.
720last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
721last = foldx (\_ y -> Just y) Nothing id
723{-# INLINE minimum #-}
724minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
725minimum m = go Nothing (toStream m)
728 let stop = return Nothing
729 single a = return (Just a)
730 yieldk a r = go (Just a) r
731 in unStream m1 defState stop single yieldk
734 let stop = return (Just res)
737 then return (Just res)
743 in unStream m1 defState stop single yieldk
745{-# INLINE minimumBy #-}
747 :: (IsStream t, Monad m)
748 => (a -> a -> Ordering) -> t m a -> m (Maybe a)
749minimumBy cmp m = go Nothing (toStream m)
752 let stop = return Nothing
753 single a = return (Just a)
754 yieldk a r = go (Just a) r
755 in unStream m1 defState stop single yieldk
758 let stop = return (Just res)
759 single a = case cmp res a of
760 GT -> return (Just a)
761 _ -> return (Just res)
762 yieldk a r = case cmp res a of
765 in unStream m1 defState stop single yieldk
767{-# INLINE maximum #-}
768maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
769maximum m = go Nothing (toStream m)
772 let stop = return Nothing
773 single a = return (Just a)
774 yieldk a r = go (Just a) r
775 in unStream m1 defState stop single yieldk
778 let stop = return (Just res)
782 else return (Just res)
787 in unStream m1 defState stop single yieldk
789{-# INLINE maximumBy #-}
790maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
791maximumBy cmp m = go Nothing (toStream m)
794 let stop = return Nothing
795 single a = return (Just a)
796 yieldk a r = go (Just a) r
797 in unStream m1 defState stop single yieldk
800 let stop = return (Just res)
801 single a = case cmp res a of
802 GT -> return (Just res)
804 yieldk a r = case cmp res a of
805 GT -> go (Just res) r
807 in unStream m1 defState stop single yieldk
810(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
811m !! i = go i (toStream m)
814 let single a | n == 0 = return $ Just a
815 | otherwise = return Nothing
816 yieldk a x | n < 0 = return Nothing
817 | n == 0 = return $ Just a
818 | otherwise = go (n - 1) x
819 in unStream m1 defState (return Nothing) single yieldk
822lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
823lookup e m = go (toStream m)
826 let single (a, b) | a == e = return $ Just b
827 | otherwise = return Nothing
828 yieldk (a, b) x | a == e = return $ Just b
830 in unStream m1 defState (return Nothing) single yieldk
833findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
834findM p m = go (toStream m)
839 if b then return $ Just a else return Nothing
842 if b then return $ Just a else go x
843 in unStream m1 defState (return Nothing) single yieldk
846find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
847find p = findM (return . p)
849{-# INLINE findIndices #-}
850findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int
851findIndices p = fromStream . go 0 . toStream
853 go offset m1 = Stream $ \st stp sng yld ->
854 let single a | p a = sng offset
856 yieldk a x | p a = yld offset $ go (offset + 1) x
857 | otherwise = unStream (go (offset + 1) x) st stp sng yld
858 in unStream m1 st stp single yieldk
860------------------------------------------------------------------------------
862------------------------------------------------------------------------------
864-- | Apply a monadic action to each element of the stream and discard the
865-- output of the action.
866mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
867mapM_ f m = go (toStream m)
871 single a = void (f a)
872 yieldk a r = f a >> go r
873 in unStream m1 defState stop single yieldk
875------------------------------------------------------------------------------
877------------------------------------------------------------------------------
879{-# INLINABLE toList #-}
880toList :: (IsStream t, Monad m) => t m a -> m [a]
883{-# INLINE toStreamK #-}
884toStreamK :: Stream m a -> Stream m a
887-------------------------------------------------------------------------------
888-- Transformation by folding (Scans)
889-------------------------------------------------------------------------------
892scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
893scanx step begin done m =
894 cons (done begin) $ fromStream $ go (toStream m) begin
896 go m1 !acc = Stream $ \st stp sng yld ->
897 let single a = sng (done $ step acc a)
900 in yld (done s) (go r s)
901 in unStream m1 st stp single yieldk
904scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
905scanl' step begin = scanx step begin id
907-------------------------------------------------------------------------------
909-------------------------------------------------------------------------------
912filter :: IsStream t => (a -> Bool) -> t m a -> t m a
913filter p m = fromStream $ go (toStream m)
915 go m1 = Stream $ \st stp sng yld ->
916 let single a | p a = sng a
918 yieldk a r | p a = yld a (go r)
919 | otherwise = unStream r st stp single yieldk
920 in unStream m1 st stp single yieldk
923take :: IsStream t => Int -> t m a -> t m a
924take n m = fromStream $ go n (toStream m)
926 go n1 m1 = Stream $ \st stp sng yld ->
927 let yieldk a r = yld a (go (n1 - 1) r)
930 else unStream m1 st stp sng yieldk
932{-# INLINE takeWhile #-}
933takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
934takeWhile p m = fromStream $ go (toStream m)
936 go m1 = Stream $ \st stp sng yld ->
937 let single a | p a = sng a
939 yieldk a r | p a = yld a (go r)
941 in unStream m1 st stp single yieldk
943drop :: IsStream t => Int -> t m a -> t m a
944drop n m = fromStream $ Stream $ \st stp sng yld ->
945 unStream (go n (toStream m)) st stp sng yld
947 go n1 m1 = Stream $ \st stp sng yld ->
949 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
950 -- Somehow "<=" check performs better than a ">"
952 then unStream m1 st stp sng yld
953 else unStream m1 st stp single yieldk
955{-# INLINE dropWhile #-}
956dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
957dropWhile p m = fromStream $ go (toStream m)
959 go m1 = Stream $ \st stp sng yld ->
960 let single a | p a = stp
962 yieldk a r | p a = unStream r st stp single yieldk
963 | otherwise = yld a r
964 in unStream m1 st stp single yieldk
966-------------------------------------------------------------------------------
968-------------------------------------------------------------------------------
971map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
972map f m = fromStream $ Stream $ \st stp sng yld ->
974 yieldk a r = yld (f a) (fmap f r)
975 in unStream (toStream m) st stp single yieldk
977-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
978-- other stream types to overload it.
980mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
981mapM f m = go (toStream m)
983 go m1 = fromStream $ Stream $ \st stp sng yld ->
984 let single a = f a >>= sng
985 yieldk a r = unStreamShared (toStream (f a |: go r)) st stp sng yld
986 in unStream m1 st stp single yieldk
988-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
989-- other stream types to overload it.
990{-# INLINE sequence #-}
991sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
992sequence m = go (toStream m)
994 go m1 = fromStream $ Stream $ \st stp sng yld ->
995 let single ma = ma >>= sng
996 yieldk ma r = unStreamShared (toStream $ ma |: go r) st stp sng yld
997 in unStream m1 st stp single yieldk
999-------------------------------------------------------------------------------
1001-------------------------------------------------------------------------------
1003{-# INLINE intersperseM #-}
1004intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1005intersperseM a m = fromStream $ prependingStart (toStream m)
1007 prependingStart m1 = Stream $ \st stp sng yld ->
1008 let yieldk i x = unStreamShared (return i |: go x) st stp sng yld
1009 in unStream m1 st stp sng yieldk
1010 go m2 = fromStream $ Stream $ \st stp sng yld ->
1011 let single i = unStreamShared (a |: yield i) st stp sng yld
1012 yieldk i x = unStreamShared (a |: return i |: go x) st stp sng yld
1013 in unStream m2 st stp single yieldk
1015{-# INLINE insertBy #-}
1016insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
1017insertBy cmp x m = fromStream $ go (toStream m)
1019 go m1 = Stream $ \st _ _ yld ->
1020 let single a = case cmp x a of
1021 GT -> yld a (yield x)
1022 _ -> yld x (yield a)
1024 yieldk a r = case cmp x a of
1026 _ -> yld x (a `cons` r)
1027 in unStream m1 st stop single yieldk
1029------------------------------------------------------------------------------
1031------------------------------------------------------------------------------
1033{-# INLINE deleteBy #-}
1034deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a
1035deleteBy eq x m = fromStream $ go (toStream m)
1037 go m1 = Stream $ \st stp sng yld ->
1038 let single a = if eq x a then stp else sng a
1039 yieldk a r = if eq x a
1040 then unStream r st stp sng yld
1042 in unStream m1 st stp single yieldk
1044-------------------------------------------------------------------------------
1046-------------------------------------------------------------------------------
1048{-# INLINE mapMaybe #-}
1049mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
1050mapMaybe f m = go (toStream m)
1052 go m1 = fromStream $ Stream $ \st stp sng yld ->
1053 let single a = case f a of
1056 yieldk a r = case f a of
1057 Just b -> yld b (toStream $ go r)
1058 Nothing -> unStream r st stp single yieldk
1059 in unStream m1 st stp single yieldk
1061------------------------------------------------------------------------------
1063------------------------------------------------------------------------------
1065{-# INLINE zipWithS #-}
1066zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1069 go mx my = Stream $ \st stp sng yld -> do
1071 let single2 b = sng (f a b)
1072 yield2 b rb = yld (f a b) (go ra rb)
1073 in unStream my st stp single2 yield2
1074 let single1 a = merge a nil
1076 unStream mx st stp single1 yield1
1078-- | Zip two streams serially using a pure zipping function.
1081{-# INLINABLE zipWith #-}
1082zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
1083zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)
1085-- | Zip two streams serially using a monadic zipping function.
1088zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
1089zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
1091 go mx my = Stream $ \st stp sng yld -> do
1093 let runIt x = unStream x st stp sng yld
1094 single2 b = f a b >>= sng
1095 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
1096 in unStream my st stp single2 yield2
1097 let single1 a = merge a nil
1099 unStream mx st stp single1 yield1
1101------------------------------------------------------------------------------
1103------------------------------------------------------------------------------
1105{-# INLINE mergeByS #-}
1108 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
1111 go mx my = Stream $ \st stp sng yld -> do
1112 let mergeWithY a ra =
1113 let stop2 = unStream mx st stp sng yld
1117 GT -> yld b (go (a `cons` ra) nil)
1118 _ -> yld a (go ra (b `cons` nil))
1122 GT -> yld b (go (a `cons` ra) rb)
1123 _ -> yld a (go ra (b `cons` rb))
1124 in unStream my st stop2 single2 yield2
1125 let stopX = unStream my st stp sng yld
1126 singleX a = mergeWithY a nil
1128 unStream mx st stopX singleX yieldX
1130{-# INLINABLE mergeByM #-}
1132 :: (IsStream t, Monad m)
1133 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1134mergeByM f m1 m2 = fromStream $ mergeByS f (toStream m1) (toStream m2)
1136{-# INLINABLE mergeBy #-}
1138 :: (IsStream t, Monad m)
1139 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
1140mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
1142------------------------------------------------------------------------------
1143-- Transformation comprehensions
1144------------------------------------------------------------------------------
1147the :: (Eq a, IsStream t, Monad m) => t m a -> m (Maybe a)
1151 Nothing -> return Nothing
1152 Just (h, t) -> go h (toStream t)
1155 let single a | h == a = return $ Just h
1156 | otherwise = return Nothing
1157 yieldk a r | h == a = go h r
1158 | otherwise = return Nothing
1159 in unStream m1 defState (return $ Just h) single yieldk
1161------------------------------------------------------------------------------
1163------------------------------------------------------------------------------
1165-- | Concatenates two streams sequentially i.e. the first stream is
1166-- exhausted completely before yielding any element from the second stream.
1167{-# INLINE serial #-}
1168serial :: Stream m a -> Stream m a -> Stream m a
1171 go m = Stream $ \st stp sng yld ->
1172 let stop = unStream m2 st stp sng yld
1174 yieldk a r = yld a (go r)
1175 in unStream m st stop single yieldk
1177instance Semigroup (Stream m a) where
1180------------------------------------------------------------------------------
1182------------------------------------------------------------------------------
1184instance Monoid (Stream m a) where
1188-------------------------------------------------------------------------------
1190-------------------------------------------------------------------------------
1192instance Monad m => Functor (Stream m) where
1195-------------------------------------------------------------------------------
1197-------------------------------------------------------------------------------
1199{-# INLINE bindWith #-}
1201 :: (forall c. Stream m c -> Stream m c -> Stream m c)
1203 -> (a -> Stream m b)
1205bindWith par m1 f = go m1
1208 Stream $ \st stp sng yld ->
1209 let runShared x = unStreamShared x st stp sng yld
1210 runIsolated x = unStream x st stp sng yld
1212 single a = runIsolated $ f a
1213 yieldk a r = runShared $ unShare (f a) `par` go r
1214 in unStream m st stp single yieldk
1216------------------------------------------------------------------------------
1217-- Alternative & MonadPlus
1218------------------------------------------------------------------------------
1220_alt :: Stream m a -> Stream m a -> Stream m a
1221_alt m1 m2 = Stream $ \st stp sng yld ->
1222 let stop = unStream m2 st stp sng yld
1223 in unStream m1 st stop sng yld
1225------------------------------------------------------------------------------
1227------------------------------------------------------------------------------
1229withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1231 Stream $ \st stp sng yld ->
1232 let single = local f . sng
1233 yieldk a r = local f $ yld a (withLocal f r)
1234 in unStream m st (local f stp) single yieldk
1236------------------------------------------------------------------------------
1238------------------------------------------------------------------------------
1241-- XXX handle and test cross thread state transfer
1244 => Stream m a -> (e -> Stream m a) -> Stream m a
1246 Stream $ \_ stp sng yld ->
1247 let run x = unStream x Nothing stp sng yieldk
1248 handle r = r `catchError` \e -> run $ h e
1249 yieldk a r = yld a (withCatchError r h)
1253-------------------------------------------------------------------------------
1255-------------------------------------------------------------------------------
1257instance 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