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
79 -- ** Specialized Folds
100 -- ** By folding (scans)
119 -- * Semigroup Style Composition
128 , Streaming -- deprecated
133import Control.Monad (void)
134import Control.Monad.Reader.Class (MonadReader(..))
135import Control.Monad.Trans.Class (MonadTrans(lift))
136import Data.Semigroup (Semigroup(..))
138 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,
139 take, filter, all, any, takeWhile, drop, dropWhile, minimum,
140 maximum, elem, notElem, null, head, tail)
141import qualified Prelude
145------------------------------------------------------------------------------
146-- The basic stream type
147------------------------------------------------------------------------------
149-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
150-- constructed using actions in monad 'm'. It uses stop, singleton and yield
151-- continuations equivalent to the following direct style type:
154-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
157-- To facilitate parallel composition we maintain a local state in an 'SVar'
158-- that is shared across and is used for synchronization of the streams being
161-- The singleton case can be expressed in terms of stop and yield but we have
162-- it as a separate case to optimize composition operations for streams with
163-- single element. We build singleton streams in the implementation of 'pure'
164-- for Applicative and Monad, and in 'lift' for MonadTrans.
168 unStream :: forall r.
169 State Stream m a -- state
171 -> (a -> m r) -- singleton
172 -> (a -> Stream m a -> m r) -- yield
176------------------------------------------------------------------------------
177-- Types that can behave as a Stream
178------------------------------------------------------------------------------
183-- | Class of types that can represent a stream of elements of some type 'a' in
187class IsStream t where
188 toStream :: t m a -> Stream m a
189 fromStream :: Stream m a -> t m a
190 -- | Constructs a stream by adding a monadic action at the head of an
191 -- existing stream. For example:
194 -- > toList $ getLine \`consM` getLine \`consM` nil
200 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
203 consM :: MonadAsync m => m a -> t m a -> t m a
204 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
205 -- to remember that @|@ comes before ':'.
208 -- > toList $ getLine |: getLine |: nil
215 -- let delay = threadDelay 1000000 >> print 1
216 -- runStream $ serially $ delay |: delay |: delay |: nil
217 -- runStream $ parallely $ delay |: delay |: delay |: nil
220 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
223 (|:) :: MonadAsync m => m a -> t m a -> t m a
224 -- We can define (|:) just as 'consM' but it is defined explicitly for each
225 -- type because we want to use SPECIALIZE pragma on the definition.
227-- | Same as 'IsStream'.
230{-# DEPRECATED Streaming "Please use IsStream instead." #-}
231type Streaming = IsStream
233-------------------------------------------------------------------------------
234-- Type adapting combinators
235-------------------------------------------------------------------------------
237-- | Adapt any specific stream type to any other specific stream type.
240adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
241adapt = fromStream . toStream
243------------------------------------------------------------------------------
245------------------------------------------------------------------------------
247-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
248-- continuation and a yield continuation.
250 => (forall r. State Stream m a
253 -> (a -> t m a -> m r)
256mkStream k = fromStream $ Stream $ \st stp sng yld ->
257 let yieldk a r = yld a (toStream r)
258 in k (rstState st) stp sng yieldk
260------------------------------------------------------------------------------
262------------------------------------------------------------------------------
272nil :: IsStream t => t m a
273nil = fromStream $ Stream $ \_ stp _ _ -> stp
277-- faster than consM because there is no bind.
278-- | Construct a stream by adding a pure value at the head of an existing
279-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
280-- more efficient. For concurrent streams this is not concurrent whereas
281-- 'consM' is concurrent. For example:
284-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
289cons :: IsStream t => a -> t m a -> t m a
290cons a r = fromStream $ Stream $ \_ _ _ yld -> yld a (toStream r)
294-- | Operator equivalent of 'cons'.
297-- > toList $ 1 .: 2 .: 3 .: nil
302(.:) :: IsStream t => a -> t m a -> t m a
305{-# INLINE consMSerial #-}
306consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a
307consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
309------------------------------------------------------------------------------
310-- Asynchronous construction
311------------------------------------------------------------------------------
313-- | Make an empty stream from a callback function.
314nilK :: IsStream t => (forall r. m r -> m r) -> t m a
315nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp
317-- | Make a singleton stream from a one shot callback function.
318yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
319yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng
321-- | Construct a stream from a callback function.
322consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a
323consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))
325-- XXX consK with concurrent callbacks
326-- XXX Build a stream from a repeating callback function.
328-------------------------------------------------------------------------------
330-------------------------------------------------------------------------------
332instance IsStream Stream where
337 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
338 consM :: Monad m => m a -> Stream m a -> Stream m a
342 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
343 (|:) :: Monad m => m a -> Stream m a -> Stream m a
346-------------------------------------------------------------------------------
348-------------------------------------------------------------------------------
351uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))
353 let stop = return Nothing
354 single a = return (Just (a, nil))
355 yieldk a r = return (Just (a, fromStream r))
356 in (unStream (toStream m)) defState stop single yieldk
358-------------------------------------------------------------------------------
360-------------------------------------------------------------------------------
362{-# INLINE unfoldr #-}
363unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
364unfoldr step = fromStream . go
366 go s = Stream $ \_ stp _ yld ->
369 Just (a, b) -> yld a (go b)
371{-# INLINE unfoldrM #-}
372unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
375 go s = fromStream $ Stream $ \svr stp sng yld -> do
380 unStream (toStream (return a |: go b)) svr stp sng yld
382-------------------------------------------------------------------------------
384-------------------------------------------------------------------------------
386-- Faster than yieldM because there is no bind. Usually we can construct a
387-- stream from a pure value using "pure" in an applicative, however in case of
388-- Zip streams pure creates an infinite stream.
390-- | Create a singleton stream from a pure value. In monadic streams, 'pure' or
391-- 'return' can be used in place of 'yield', however, in Zip applicative
392-- streams 'pure' is equivalent to 'repeat'.
395yield :: IsStream t => a -> t m a
396yield a = fromStream $ Stream $ \_ _ single _ -> single a
398-- | Create a singleton stream from a monadic action. Same as @m \`consM` nil@
399-- but more efficient.
402-- > toList $ yieldM getLine
409yieldM :: (Monad m, IsStream t) => m a -> t m a
410yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single
415{-# DEPRECATED once "Please use yieldM instead." #-}
417once :: (Monad m, IsStream t) => m a -> t m a
420-- | Generate an infinite stream by repeating a pure value.
423repeat :: IsStream t => a -> t m a
424repeat a = let x = cons a x in x
426-------------------------------------------------------------------------------
428-------------------------------------------------------------------------------
430-- | Construct a stream from a 'Foldable' containing pure values.
433{-# INLINE fromFoldable #-}
434fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
435fromFoldable = Prelude.foldr cons nil
437{-# INLINE fromList #-}
438fromList :: IsStream t => [a] -> t m a
439fromList = fromFoldable
441{-# INLINE fromStreamK #-}
442fromStreamK :: Stream m a -> Stream m a
445-------------------------------------------------------------------------------
446-- Elimination by Folding
447-------------------------------------------------------------------------------
449-- | Fold a stream by providing an SVar, a stop continuation, a singleton
450-- continuation and a yield continuation.
456 -> (a -> t m a -> m r)
459foldStream st blank single step m =
460 let yieldk a x = step a (fromStream x)
461 in (unStream (toStream m)) st blank single yieldk
463-- | Lazy right associative fold.
464foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
465foldr step acc m = go (toStream m)
468 let stop = return acc
469 single a = return (step a acc)
470 yieldk a r = go r >>= \b -> return (step a b)
471 in (unStream m1) defState stop single yieldk
473-- | Lazy right fold with a monadic step function.
475foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b
476foldrM step acc m = go (toStream m)
479 let stop = return acc
480 single a = step a acc
481 yieldk a r = go r >>= step a
482 in (unStream m1) defState stop single yieldk
484-- | Strict left fold with an extraction function. Like the standard strict
485-- left fold, but applies a user supplied extraction function (the third
486-- argument) to the folded value at the end. This is designed to work with the
487-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
489foldx :: (IsStream t, Monad m)
490 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b
491foldx step begin done m = get $ go (toStream m) begin
495 let single = return . done
496 in (unStream m1) undefined undefined single undefined
498 -- Note, this can be implemented by making a recursive call to "go",
499 -- however that is more expensive because of unnecessary recursion
500 -- that cannot be tail call optimized. Unfolding recursion explicitly via
501 -- continuations is much more efficient.
502 go m1 !acc = Stream $ \_ _ sng yld ->
504 single a = sng $ step acc a
506 let stream = go r (step acc a)
507 in (unStream stream) defState undefined sng yld
508 in (unStream m1) defState stop single yieldk
510-- | Strict left associative fold.
512foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b
513foldl' step begin m = foldx step begin id m
515-- XXX replace the recursive "go" with explicit continuations.
516-- | Like 'foldx', but with a monadic step function.
517foldxM :: (IsStream t, Monad m)
518 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b
519foldxM step begin done m = go begin (toStream m)
522 let stop = acc >>= done
523 single a = acc >>= \b -> step b a >>= done
524 yieldk a r = acc >>= \b -> go (step b a) r
525 in (unStream m1) defState stop single yieldk
527-- | Like 'foldl'' but with a monadic step function.
528foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
529foldlM' step begin m = foldxM step (return begin) return m
531------------------------------------------------------------------------------
533------------------------------------------------------------------------------
535{-# INLINE runStream #-}
536runStream :: (Monad m, IsStream t) => t m a -> m ()
537runStream m = go (toStream m)
542 yieldk _ r = go (toStream r)
543 in unStream m1 defState stop single yieldk
546null :: (IsStream t, Monad m) => t m a -> m Bool
548 let stop = return True
549 single _ = return False
550 yieldk _ _ = return False
551 in unStream (toStream m) defState stop single yieldk
554head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
556 let stop = return Nothing
557 single a = return (Just a)
558 yieldk a _ = return (Just a)
559 in unStream (toStream m) defState stop single yieldk
562tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
564 let stop = return Nothing
565 single _ = return $ Just nil
566 yieldk _ r = return $ Just $ fromStream r
567 in unStream (toStream m) defState stop single yieldk
570elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
571elem e m = go (toStream m)
574 let stop = return False
575 single a = return (a == e)
576 yieldk a r = if a == e then return True else go r
577 in (unStream m1) defState stop single yieldk
579{-# INLINE notElem #-}
580notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
581notElem e m = go (toStream m)
584 let stop = return True
585 single a = return (a /= e)
586 yieldk a r = if a == e then return False else go r
587 in (unStream m1) defState stop single yieldk
589all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
590all p m = go (toStream m)
593 let single a | p a = return True
594 | otherwise = return False
595 yieldk a r | p a = go r
596 | otherwise = return False
597 in unStream m1 defState (return True) single yieldk
599any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
600any p m = go (toStream m)
603 let single a | p a = return True
604 | otherwise = return False
605 yieldk a r | p a = return True
607 in unStream m1 defState (return False) single yieldk
609-- | Extract the last element of the stream, if any.
611last :: (IsStream t, Monad m) => t m a -> m (Maybe a)
612last = foldx (\_ y -> Just y) Nothing id
614{-# INLINE minimum #-}
615minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
616minimum m = go Nothing (toStream m)
619 let stop = return Nothing
620 single a = return (Just a)
621 yieldk a r = go (Just a) r
622 in unStream m1 defState stop single yieldk
625 let stop = return (Just res)
628 then return (Just res)
634 in unStream m1 defState stop single yieldk
636{-# INLINE maximum #-}
637maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
638maximum m = go Nothing (toStream m)
641 let stop = return Nothing
642 single a = return (Just a)
643 yieldk a r = go (Just a) r
644 in unStream m1 defState stop single yieldk
647 let stop = return (Just res)
651 else return (Just res)
656 in unStream m1 defState stop single yieldk
658------------------------------------------------------------------------------
660------------------------------------------------------------------------------
662-- | Apply a monadic action to each element of the stream and discard the
663-- output of the action.
664mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()
665mapM_ f m = go (toStream m)
669 single a = void (f a)
670 yieldk a r = f a >> go r
671 in (unStream m1) defState stop single yieldk
673------------------------------------------------------------------------------
675------------------------------------------------------------------------------
677{-# INLINABLE toList #-}
678toList :: (IsStream t, Monad m) => t m a -> m [a]
681{-# INLINE toStreamK #-}
682toStreamK :: Stream m a -> Stream m a
685-------------------------------------------------------------------------------
686-- Transformation by folding (Scans)
687-------------------------------------------------------------------------------
690scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
691scanx step begin done m =
692 cons (done begin) $ fromStream $ go (toStream m) begin
694 go m1 !acc = Stream $ \st stp sng yld ->
695 let single a = sng (done $ step acc a)
698 in yld (done s) (go r s)
699 in unStream m1 (rstState st) stp single yieldk
702scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
703scanl' step begin m = scanx step begin id m
705-------------------------------------------------------------------------------
707-------------------------------------------------------------------------------
710filter :: IsStream t => (a -> Bool) -> t m a -> t m a
711filter p m = fromStream $ go (toStream m)
713 go m1 = Stream $ \st stp sng yld ->
714 let single a | p a = sng a
716 yieldk a r | p a = yld a (go r)
717 | otherwise = (unStream r) (rstState st) stp single yieldk
718 in unStream m1 (rstState st) stp single yieldk
721take :: IsStream t => Int -> t m a -> t m a
722take n m = fromStream $ go n (toStream m)
724 go n1 m1 = Stream $ \st stp sng yld ->
725 let yieldk a r = yld a (go (n1 - 1) r)
728 else unStream m1 (rstState st) stp sng yieldk
730{-# INLINE takeWhile #-}
731takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
732takeWhile p m = fromStream $ go (toStream m)
734 go m1 = Stream $ \st stp sng yld ->
735 let single a | p a = sng a
737 yieldk a r | p a = yld a (go r)
739 in unStream m1 (rstState st) stp single yieldk
741drop :: IsStream t => Int -> t m a -> t m a
742drop n m = fromStream $ Stream $ \st stp sng yld ->
743 unStream (go n (toStream m)) (rstState st) stp sng yld
745 go n1 m1 = Stream $ \st stp sng yld ->
747 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
748 -- Somehow "<=" check performs better than a ">"
750 then unStream m1 st stp sng yld
751 else unStream m1 st stp single yieldk
753{-# INLINE dropWhile #-}
754dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
755dropWhile p m = fromStream $ go (toStream m)
757 go m1 = Stream $ \st stp sng yld ->
758 let single a | p a = stp
760 yieldk a r | p a = (unStream r) (rstState st) stp single yieldk
761 | otherwise = yld a r
762 in unStream m1 (rstState st) stp single yieldk
764-------------------------------------------------------------------------------
766-------------------------------------------------------------------------------
769map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
770map f m = fromStream $ Stream $ \st stp sng yld ->
772 yieldk a r = yld (f a) (fmap f r)
773 in unStream (toStream m) (rstState st) stp single yieldk
775-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
776-- other stream types to overload it.
778mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
779mapM f m = go (toStream m)
781 go m1 = fromStream $ Stream $ \st stp sng yld ->
782 let single a = f a >>= sng
783 yieldk a r = unStream (toStream (f a |: (go r))) st stp sng yld
784 in (unStream m1) (rstState st) stp single yieldk
786-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
787-- other stream types to overload it.
788{-# INLINE sequence #-}
789sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
790sequence m = go (toStream m)
792 go m1 = fromStream $ Stream $ \st stp sng yld ->
793 let single ma = ma >>= sng
794 yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld
795 in (unStream m1) (rstState st) stp single yieldk
797-------------------------------------------------------------------------------
799-------------------------------------------------------------------------------
801{-# INLINE mapMaybe #-}
802mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b
803mapMaybe f m = go (toStream m)
805 go m1 = fromStream $ Stream $ \st stp sng yld ->
806 let single a = case f a of
809 yieldk a r = case f a of
810 Just b -> yld b (toStream $ go r)
811 Nothing -> (unStream r) (rstState st) stp single yieldk
812 in unStream m1 (rstState st) stp single yieldk
814------------------------------------------------------------------------------
816------------------------------------------------------------------------------
818-- | Concatenates two streams sequentially i.e. the first stream is
819-- exhausted completely before yielding any element from the second stream.
821serial :: Stream m a -> Stream m a -> Stream m a
824 go (Stream m) = Stream $ \st stp sng yld ->
825 let stop = (unStream m2) (rstState st) stp sng yld
827 yieldk a r = yld a (go r)
828 in m (rstState st) stop single yieldk
830instance Semigroup (Stream m a) where
833------------------------------------------------------------------------------
835------------------------------------------------------------------------------
837instance Monoid (Stream m a) where
841-------------------------------------------------------------------------------
843-------------------------------------------------------------------------------
845instance Monad m => Functor (Stream m) where
848-------------------------------------------------------------------------------
850-------------------------------------------------------------------------------
852{-# INLINE bindWith #-}
854 :: (forall c. Stream m c -> Stream m c -> Stream m c)
858bindWith par m f = go m
861 Stream $ \st stp sng yld ->
862 let run x = (unStream x) st stp sng yld
864 yieldk a r = run $ f a `par` go r
865 in g (rstState st) stp single yieldk
867------------------------------------------------------------------------------
868-- Alternative & MonadPlus
869------------------------------------------------------------------------------
871_alt :: Stream m a -> Stream m a -> Stream m a
872_alt m1 m2 = Stream $ \st stp sng yld ->
873 let stop = unStream m2 (rstState st) stp sng yld
874 in unStream m1 (rstState st) stop sng yld
876------------------------------------------------------------------------------
878------------------------------------------------------------------------------
880withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
882 Stream $ \st stp sng yld ->
883 let single = local f . sng
884 yieldk a r = local f $ yld a (withLocal f r)
885 in (unStream m) (rstState st) (local f stp) single yieldk
887------------------------------------------------------------------------------
889------------------------------------------------------------------------------
892-- XXX handle and test cross thread state transfer
895 => Stream m a -> (e -> Stream m a) -> Stream m a
897 Stream $ \_ stp sng yld ->
898 let run x = unStream x Nothing stp sng yieldk
899 handle r = r `catchError` \e -> run $ h e
900 yieldk a r = yld a (withCatchError r h)
904-------------------------------------------------------------------------------
906-------------------------------------------------------------------------------
908instance 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