2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
9{-# LANGUAGE StandaloneDeriving #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
13-- Module : Streamly.Streams
14-- Copyright : (c) 2017 Harendra Kumar
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
22module Streamly.Streams
25 , Streaming -- deprecated
39 , runStreaming -- deprecated
69 -- * Stream Transformers
71 , StreamT -- deprecated
73 , InterleavedT -- deprecated
78 , ZipStream -- deprecated
83 , serially -- deprecated
85 , interleaving -- deprecated
91 , zipping -- deprecated
93 , zippingAsync -- deprecated
97 , runStreamT -- deprecated
98 , runInterleavedT -- deprecated
99 , runAsyncT -- deprecated
100 , runParallelT -- deprecated
101 , runZipStream -- deprecated
102 , runZipAsync -- deprecated
111import Control.Monad (ap)
112import Control.Monad.Base (MonadBase (..), liftBaseDefault)
113import Control.Monad.Catch (MonadThrow, throwM)
114-- import Control.Monad.Error.Class (MonadError(..))
115import Control.Monad.IO.Class (MonadIO(..))
116import Control.Monad.Reader.Class (MonadReader(..))
117import Control.Monad.State.Class (MonadState(..))
118import Control.Monad.Trans.Class (MonadTrans (lift))
119import Data.Semigroup (Semigroup(..))
120import Streamly.Core ( MonadAsync , SVar,
122import qualified Streamly.Core as S
124------------------------------------------------------------------------------
125-- Types that can behave as a Stream
126------------------------------------------------------------------------------
131-- | Class of types that can represent a stream of elements of some type 'a' in
135class IsStream t where
136 toStream :: t m a -> S.Stream m a
137 fromStream :: S.Stream m a -> t m a
138 -- | Constructs a stream by adding a monadic action at the head of an
139 -- existing stream. For example:
142 -- > toList $ getLine \`consM` getLine \`consM` nil
148 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
151 consM :: MonadAsync m => m a -> t m a -> t m a
152 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
153 -- to remember that @|@ comes before ':'.
156 -- > toList $ getLine |: getLine |: nil
163 -- let delay = threadDelay 1000000 >> print 1
164 -- runStream $ serially $ delay |: delay |: delay |: nil
165 -- runStream $ parallely $ delay |: delay |: delay |: nil
168 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
171 (|:) :: MonadAsync m => m a -> t m a -> t m a
172 -- We can define (|:) just as 'consM' but it is defined explicitly for each
173 -- type because we want to use SPECIALIZE pragma on the definition.
175-- | Same as 'IsStream'.
178{-# DEPRECATED Streaming "Please use IsStream instead." #-}
179type Streaming = IsStream
181------------------------------------------------------------------------------
182-- Constructing a stream
183------------------------------------------------------------------------------
193nil :: IsStream t => t m a
194nil = fromStream S.nil
196-- | Constructs a stream by adding a monadic action at the head of an existing
197-- stream. For example:
200-- > toList $ getLine \`consM` getLine \`consM` nil
207consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
208consMSerial m r = fromStream $ S.consM m (toStream r)
212-- | Construct a stream by adding a pure value at the head of an existing
213-- stream. For pure values it can be faster than 'consM'. For example:
216-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
221cons :: IsStream t => a -> t m a -> t m a
222cons a r = fromStream $ S.cons a (toStream r)
226-- | Operator equivalent of 'cons'.
229-- > toList $ 1 .: 2 .: 3 .: nil
234(.:) :: IsStream t => a -> t m a -> t m a
237-- | Build a stream from its church encoding. The function passed maps
238-- directly to the underlying representation of the stream type. The second
239-- parameter to the function is the "yield" function yielding a value and the
240-- remaining stream if any otherwise 'Nothing'. The third parameter is to
241-- represent an "empty" stream.
242streamBuild :: IsStream t
243 => (forall r. Maybe (SVar m a)
244 -> (a -> t m a -> m r)
249streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
250 let yield a r = yld a (toStream r)
251 in k svr yield sng stp
253-- | Build a singleton stream from a callback function.
254fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
255fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
257-- | Read an SVar to get a stream.
258fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a
259fromSVar sv = fromStream $ S.fromStreamVar sv
261------------------------------------------------------------------------------
262-- Destroying a stream
263------------------------------------------------------------------------------
265-- | Fold a stream using its church encoding. The second argument is the "step"
266-- function consuming an element and the remaining stream, if any. The third
267-- argument is for consuming an "empty" stream that yields nothing.
271 -> (a -> t m a -> m r)
276streamFold svr step single blank m =
277 let yield a x = step a (fromStream x)
278 in (S.runStream (toStream m)) svr blank single yield
280-- | Run a streaming composition, discard the results. By default it interprets
281-- the stream as 'SerialT', to run other types of streams use the type adapting
282-- combinators for example @runStream . 'asyncly'@.
285runStream :: Monad m => SerialT m a -> m ()
286runStream m = go (toStream m)
292 in (S.runStream m1) Nothing stop single yield
294-- | Same as 'runStream'
297{-# DEPRECATED runStreaming "Please use runStream instead." #-}
298runStreaming :: (Monad m, IsStream t) => t m a -> m ()
299runStreaming = runStream . adapt
301-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
302-- be read back from the SVar using 'fromSVar'.
303toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()
304toSVar sv m = S.toStreamVar sv (toStream m)
306------------------------------------------------------------------------------
308------------------------------------------------------------------------------
310-- XXX Get rid of this?
311-- | Make a stream asynchronous, triggers the computation and returns a stream
312-- in the underlying monad representing the output generated by the original
313-- computation. The returned action is exhaustible and must be drained once. If
314-- not drained fully we may have a thread blocked forever and once exhausted it
315-- will always return 'empty'.
318mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
320 sv <- S.newStreamVar1 AsyncVar (toStream m)
323{-# INLINE applyWith #-}
324applyWith :: (IsStream t, MonadAsync m)
325 => SVarStyle -> (t m a -> t m b) -> t m a -> t m b
326applyWith style f x = fromStream $
327 S.applyWith style (toStream . f . fromStream) (toStream x)
329{-# INLINE runWith #-}
330runWith :: (IsStream t, MonadAsync m)
331 => SVarStyle -> (t m a -> m b) -> t m a -> m b
332runWith style f x = S.runWith style (f . fromStream) (toStream x)
340-- | Parallel function application operator for streams; just like the regular
341-- function application operator '$' except that it is concurrent. The
342-- following code prints a value every second even though each stage adds a 1
348-- S.mapM (\\x -> threadDelay 1000000 >> print x)
349-- |$ S.repeatM (threadDelay 1000000 >> return 1)
356(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
357f |$ x = applyWith ParallelVar f x
359-- | Parallel reverse function application operator for streams; just like the
360-- regular reverse function application operator '&' except that it is
365-- S.repeatM (threadDelay 1000000 >> return 1)
366-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
373(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
376-- | Parallel function application operator; applies a @run@ or @fold@ function
377-- to a stream such that the fold consumer and the stream producer run in
378-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
379-- underlying monad. The @.@ at the end of the operator is a mnemonic for
380-- termination of the stream.
383-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
384-- |$. S.repeatM (threadDelay 1000000 >> return 1)
391(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
392f |$. x = runWith ParallelVar f x
394-- | Parallel reverse function application operator for applying a run or fold
395-- functions to a stream. Just like '|$.' except that the operands are reversed.
398-- S.repeatM (threadDelay 1000000 >> return 1)
399-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
406(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
409------------------------------------------------------------------------------
410-- CPP macros for common instances
411------------------------------------------------------------------------------
413-- XXX use template haskell instead and include Monoid and IsStream instances
416withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a
418 S.Stream $ \_ stp sng yld ->
419 let single = local f . sng
420 yield a r = local f $ yld a (withLocal f r)
421 in (S.runStream m) Nothing (local f stp) single yield
424-- XXX handle and test cross thread state transfer
427 => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a
429 S.Stream $ \_ stp sng yld ->
430 let run x = S.runStream x Nothing stp sng yield
431 handle r = r `catchError` \e -> run $ h e
432 yield a r = yld a (withCatchError r h)
436#define MONADPARALLEL , MonadAsync m
438#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
439instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
440 pure = STREAM . S.singleton; \
443#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
444instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
445 liftBase = liftBaseDefault }; \
447instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
448 liftIO = lift . liftIO }; \
450instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
451 throwM = lift . throwM }; \
454instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
455 throwError = lift . throwError; \
457 fromStream $ withCatchError (toStream m) (\e -> toStream $ h e) }; \
460instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
462 local f m = fromStream $ withLocal f (toStream m) }; \
464instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
466 put x = lift (put x); \
467 state k = lift (state k) }
469------------------------------------------------------------------------------
471------------------------------------------------------------------------------
473-- | Deep serial composition or serial composition with depth first traversal.
474-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
475-- depth first manner, yielding all elements from the first stream, and then
476-- all elements from the second stream.
480-- import qualified "Streamly.Prelude" as S
482-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
488-- The 'Monad' instance runs the /monadic continuation/ for each
489-- element of the stream, serially.
492-- main = 'runStream' . 'serially' $ do
493-- x <- return 1 \<\> return 2
501-- 'SerialT' nests streams serially in a depth first manner.
504-- main = 'runStream' . 'serially' $ do
505-- x <- return 1 \<\> return 2
506-- y <- return 3 \<\> return 4
507-- S.once $ print (x, y)
516-- This behavior of 'SerialT' is exactly like a list transformer. We call the
517-- monadic code being run for each element of the stream a monadic
518-- continuation. In imperative paradigm we can think of this composition as
519-- nested @for@ loops and the monadic continuation is the body of the loop. The
520-- loop iterates for all elements of the stream.
522-- The 'serially' combinator can be omitted as the default stream type is
524-- Note that serial composition with depth first traversal can be used to
525-- combine an infinite number of streams as it explores only one stream at a
529newtype SerialT m a = SerialT {getSerialT :: S.Stream m a}
530 deriving (Semigroup, Monoid, Functor, MonadTrans)
534{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
535type StreamT = SerialT
537instance IsStream SerialT where
538 toStream = getSerialT
542 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
543 consM :: Monad m => m a -> SerialT m a -> SerialT m a
547 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
548 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
551------------------------------------------------------------------------------
553------------------------------------------------------------------------------
555-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
556-- Appends two streams sequentially, yielding all elements from the first
557-- stream, and then all elements from the second stream.
561serial :: IsStream t => t m a -> t m a -> t m a
562serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
564------------------------------------------------------------------------------
566------------------------------------------------------------------------------
568instance Monad m => Monad (SerialT m) where
570 (SerialT (S.Stream m)) >>= f = SerialT $ S.Stream $ \_ stp sng yld ->
571 let run x = (S.runStream x) Nothing stp sng yld
572 single a = run $ toStream (f a)
573 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
574 in m Nothing stp single yield
576------------------------------------------------------------------------------
578------------------------------------------------------------------------------
580MONAD_APPLICATIVE_INSTANCE(SerialT,)
581MONAD_COMMON_INSTANCES(SerialT,)
583------------------------------------------------------------------------------
585------------------------------------------------------------------------------
587-- | Wide serial composition or serial composition with a breadth first
588-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
589-- the two streams in a breadth first manner. In other words, it interleaves
590-- two streams, yielding one element from each stream alternately.
594-- import qualified "Streamly.Prelude" as S
596-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
602-- Similarly, the 'Monad' instance interleaves the iterations of the
603-- inner and the outer loop, nesting loops in a breadth first manner.
607-- main = 'runStream' . 'wSerially' $ do
608-- x <- return 1 \<\> return 2
609-- y <- return 3 \<\> return 4
610-- S.once $ print (x, y)
619-- Note that a serial composition with breadth first traversal can only combine
620-- a finite number of streams as it needs to retain state for each unfinished
624newtype WSerialT m a = WSerialT {getWSerialT :: S.Stream m a}
625 deriving (Functor, MonadTrans)
629{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
630type InterleavedT = WSerialT
632instance IsStream WSerialT where
633 toStream = getWSerialT
634 fromStream = WSerialT
637 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
638 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
642 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
643 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
646------------------------------------------------------------------------------
648------------------------------------------------------------------------------
650-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
651-- Interleaves two streams, yielding one element from each stream alternately.
654{-# INLINE wSerial #-}
655wSerial :: IsStream t => t m a -> t m a -> t m a
656wSerial m1 m2 = fromStream $ S.wSerial (toStream m1) (toStream m2)
658instance Semigroup (WSerialT m a) where
663-- | Same as 'wSerial'.
666{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
668(<=>) :: IsStream t => t m a -> t m a -> t m a
671------------------------------------------------------------------------------
673------------------------------------------------------------------------------
675instance Monoid (WSerialT m a) where
679------------------------------------------------------------------------------
681------------------------------------------------------------------------------
683instance Monad m => Monad (WSerialT m) where
685 (WSerialT (S.Stream m)) >>= f = WSerialT $ S.Stream $ \_ stp sng yld ->
686 let run x = (S.runStream x) Nothing stp sng yld
687 single a = run $ toStream (f a)
688 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
689 in m Nothing stp single yield
691------------------------------------------------------------------------------
693------------------------------------------------------------------------------
695MONAD_APPLICATIVE_INSTANCE(WSerialT,)
696MONAD_COMMON_INSTANCES(WSerialT,)
698------------------------------------------------------------------------------
700------------------------------------------------------------------------------
702-- | Deep ahead composition or ahead composition with depth first traversal.
703-- The semigroup composition of 'AheadT' appends streams in a depth first
704-- manner just like 'SerialT' except that it can produce elements concurrently
705-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
706-- as it arrives whereas 'AheadT' orders the output in the traversal order.
709-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
715-- Any exceptions generated by a constituent stream are propagated to the
718-- Similarly, the monad instance of 'AheadT' may run each iteration
719-- concurrently ahead of time but presents the results in the same order as
724-- import qualified "Streamly.Prelude" as S
725-- import Control.Concurrent
727-- main = 'runStream' . 'aheadly' $ do
728-- n <- return 3 \<\> return 2 \<\> return 1
730-- threadDelay (n * 1000000)
731-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
734-- ThreadId 40: Delay 1
735-- ThreadId 39: Delay 2
736-- ThreadId 38: Delay 3
739-- All iterations may run in the same thread if they do not block.
741-- Note that ahead composition with depth first traversal can be used to
742-- combine infinite number of streams as it explores only a bounded number of
746newtype AheadT m a = AheadT {getAheadT :: S.Stream m a}
747 deriving (Functor, MonadTrans)
749instance IsStream AheadT where
754 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
755 consM m r = fromStream $ S.consMAhead m (toStream r)
758 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
761------------------------------------------------------------------------------
763------------------------------------------------------------------------------
765-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
766-- Merges two streams sequentially but with concurrent lookahead.
770ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
771ahead m1 m2 = fromStream $ S.ahead (toStream m1) (toStream m2)
773instance MonadAsync m => Semigroup (AheadT m a) where
776------------------------------------------------------------------------------
778------------------------------------------------------------------------------
780instance MonadAsync m => Monoid (AheadT m a) where
784------------------------------------------------------------------------------
786------------------------------------------------------------------------------
788{-# INLINE aheadbind #-}
792 -> (a -> S.Stream m b)
797 S.Stream $ \ctx stp sng yld ->
798 let run x = (S.runStream x) ctx stp sng yld
800 yield a r = run $ f a `S.ahead` go r
801 in g Nothing stp single yield
803instance MonadAsync m => Monad (AheadT m) where
805 (AheadT m) >>= f = AheadT $ aheadbind m (getAheadT . f)
807------------------------------------------------------------------------------
809------------------------------------------------------------------------------
811MONAD_APPLICATIVE_INSTANCE(AheadT,MONADPARALLEL)
812MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
814------------------------------------------------------------------------------
816------------------------------------------------------------------------------
818-- | Deep async composition or async composition with depth first traversal. In
819-- a left to right 'Semigroup' composition it tries to yield elements from the
820-- left stream as long as it can, but it can run the right stream in parallel
821-- if it needs to, based on demand. The right stream can be run if the left
822-- stream blocks on IO or cannot produce elements fast enough for the consumer.
825-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
831-- Any exceptions generated by a constituent stream are propagated to the
832-- output stream. The output and exceptions from a single stream are guaranteed
833-- to arrive in the same order in the resulting stream as they were generated
834-- in the input stream. However, the relative ordering of elements from
835-- different streams in the resulting stream can vary depending on scheduling
836-- and generation delays.
838-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
839-- concurrently based on demand. More concurrent iterations are started only
840-- if the previous iterations are not able to produce enough output for the
845-- import qualified "Streamly.Prelude" as S
846-- import Control.Concurrent
848-- main = 'runStream' . 'asyncly' $ do
849-- n <- return 3 \<\> return 2 \<\> return 1
851-- threadDelay (n * 1000000)
852-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
855-- ThreadId 40: Delay 1
856-- ThreadId 39: Delay 2
857-- ThreadId 38: Delay 3
860-- All iterations may run in the same thread if they do not block.
862-- Note that async composition with depth first traversal can be used to
863-- combine infinite number of streams as it explores only a bounded number of
867newtype AsyncT m a = AsyncT {getAsyncT :: S.Stream m a}
868 deriving (Functor, MonadTrans)
870instance IsStream AsyncT where
875 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
876 consM m r = fromStream $ S.consMAsync m (toStream r)
879 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
882------------------------------------------------------------------------------
884------------------------------------------------------------------------------
886-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
887-- Merges two streams possibly concurrently, preferring the
888-- elements from the left one when available.
892async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
893async m1 m2 = fromStream $ S.async (toStream m1) (toStream m2)
898{-# DEPRECATED (<|) "Please use 'async' instead." #-}
900(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
903instance MonadAsync m => Semigroup (AsyncT m a) where
906------------------------------------------------------------------------------
908------------------------------------------------------------------------------
910instance MonadAsync m => Monoid (AsyncT m a) where
914------------------------------------------------------------------------------
916------------------------------------------------------------------------------
918{-# INLINE parbind #-}
920 :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
922 -> (a -> S.Stream m b)
924parbind par m f = go m
927 S.Stream $ \ctx stp sng yld ->
928 let run x = (S.runStream x) ctx stp sng yld
930 yield a r = run $ f a `par` go r
931 in g Nothing stp single yield
933instance MonadAsync m => Monad (AsyncT m) where
935 (AsyncT m) >>= f = AsyncT $ parbind S.async m (getAsyncT . f)
937------------------------------------------------------------------------------
939------------------------------------------------------------------------------
941MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
942MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
944------------------------------------------------------------------------------
946------------------------------------------------------------------------------
948-- | Wide async composition or async composition with breadth first traversal.
949-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
950-- streams using a depth first travesal or in a round robin fashion, yielding
951-- elements from both streams alternately.
954-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
960-- Any exceptions generated by a constituent stream are propagated to the
961-- output stream. The output and exceptions from a single stream are guaranteed
962-- to arrive in the same order in the resulting stream as they were generated
963-- in the input stream. However, the relative ordering of elements from
964-- different streams in the resulting stream can vary depending on scheduling
965-- and generation delays.
967-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
968-- concurrently using a round robin scheduling.
972-- import qualified "Streamly.Prelude" as S
973-- import Control.Concurrent
975-- main = 'runStream' . 'wAsyncly' $ do
976-- n <- return 3 \<\> return 2 \<\> return 1
978-- threadDelay (n * 1000000)
979-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
982-- ThreadId 40: Delay 1
983-- ThreadId 39: Delay 2
984-- ThreadId 38: Delay 3
987-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
988-- concurrently, unconditionally.
990-- Note that async composition with breadth first traversal can only combine a
991-- finite number of streams as it needs to retain state for each unfinished
995newtype WAsyncT m a = WAsyncT {getWAsyncT :: S.Stream m a}
996 deriving (Functor, MonadTrans)
998instance IsStream WAsyncT where
999 toStream = getWAsyncT
1000 fromStream = WAsyncT
1002 {-# INLINE consM #-}
1003 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1004 consM m r = fromStream $ S.consMWAsync m (toStream r)
1007 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1010------------------------------------------------------------------------------
1012------------------------------------------------------------------------------
1014-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
1015-- Merges two streams concurrently choosing elements from both fairly.
1018{-# INLINE wAsync #-}
1019wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1020wAsync m1 m2 = fromStream $ S.wAsync (toStream m1) (toStream m2)
1022instance MonadAsync m => Semigroup (WAsyncT m a) where
1025------------------------------------------------------------------------------
1027------------------------------------------------------------------------------
1029instance MonadAsync m => Monoid (WAsyncT m a) where
1033------------------------------------------------------------------------------
1035------------------------------------------------------------------------------
1037instance MonadAsync m => Monad (WAsyncT m) where
1040 WAsyncT $ parbind S.wAsync m (getWAsyncT . f)
1042------------------------------------------------------------------------------
1044------------------------------------------------------------------------------
1046MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
1047MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
1049------------------------------------------------------------------------------
1051------------------------------------------------------------------------------
1053-- | Async composition with simultaneous traversal of all streams.
1055-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
1056-- running both strictly concurrently and yielding elements from both streams
1057-- as they arrive. When multiple streams are combined using 'ParallelT' each
1058-- one is evaluated in its own thread and the results produced are presented in
1059-- the combined stream on a first come first serve basis.
1061-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
1062-- specific type of consumption pattern (depth first or breadth first). Since
1063-- they are lookahead, they may introduce certain default latency in starting
1064-- more concurrent tasks for efficiency reasons or may put a default limitation
1065-- on the resource consumption (e.g. number of concurrent threads for
1066-- lookahead). If we look at the implementation detail, they both can share a
1067-- pool of worker threads to evaluate the streams in the desired pattern and at
1068-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
1069-- evaluate each stream.
1071-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
1072-- constituent streams fairly in a round robin fashion.
1073-- However, the key difference is that 'WAsyncT' is lazy or pull driven
1074-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
1075-- starts concurrent evaluation of both the streams (in separate threads) and
1076-- later picks the results whereas 'WAsyncT' may wait for a certain latency
1077-- threshold before initiating concurrent evaluation of the next stream. The
1078-- concurrent scheduling of the next stream or the degree of concurrency is
1079-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
1080-- is evaluated in a separate thread and results are /pushed/ to a shared
1081-- output buffer, the evaluation rate is controlled by blocking when the buffer
1084-- Concurrent lookahead streams are generally more efficient than
1085-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
1086-- they do not necessarily use a separate thread for each task. So they should
1087-- be preferred over 'ParallelT' especially when efficiency is a concern and
1088-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
1089-- for cases when the streams are required to be evaluated simultaneously
1090-- irrespective of how the consumer consumes them e.g. when we want to race
1091-- two tasks and want to start both strictly at the same time or if we have
1092-- timers in the parallel tasks and our results depend on the timers being
1093-- started at the same time. We can say that 'ParallelT' is almost the same
1094-- (modulo some implementation differences) as 'WAsyncT' when the latter is
1095-- used with unlimited lookahead and zero latency in initiating lookahead.
1098-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
1104-- When streams with more than one element are merged, it yields whichever
1105-- stream yields first without any bias, unlike the 'Async' style streams.
1107-- Any exceptions generated by a constituent stream are propagated to the
1108-- output stream. The output and exceptions from a single stream are guaranteed
1109-- to arrive in the same order in the resulting stream as they were generated
1110-- in the input stream. However, the relative ordering of elements from
1111-- different streams in the resulting stream can vary depending on scheduling
1112-- and generation delays.
1114-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
1115-- of the loop concurrently.
1119-- import qualified "Streamly.Prelude" as S
1120-- import Control.Concurrent
1122-- main = 'runStream' . 'parallely' $ do
1123-- n <- return 3 \<\> return 2 \<\> return 1
1125-- threadDelay (n * 1000000)
1126-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1129-- ThreadId 40: Delay 1
1130-- ThreadId 39: Delay 2
1131-- ThreadId 38: Delay 3
1134-- Note that parallel composition can only combine a finite number of
1135-- streams as it needs to retain state for each unfinished stream.
1138newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
1139 deriving (Functor, MonadTrans)
1141instance IsStream ParallelT where
1142 toStream = getParallelT
1143 fromStream = ParallelT
1145 {-# INLINE consM #-}
1146 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1147 consM m r = fromStream $ S.consMParallel m (toStream r)
1150 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1153------------------------------------------------------------------------------
1155------------------------------------------------------------------------------
1157-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
1158-- Merges two streams concurrently.
1161{-# INLINE parallel #-}
1162parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1163parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
1165instance MonadAsync m => Semigroup (ParallelT m a) where
1168------------------------------------------------------------------------------
1170------------------------------------------------------------------------------
1172instance MonadAsync m => Monoid (ParallelT m a) where
1176------------------------------------------------------------------------------
1178------------------------------------------------------------------------------
1180instance MonadAsync m => Monad (ParallelT m) where
1182 (ParallelT m) >>= f = ParallelT $ parbind S.parallel m (getParallelT . f)
1184------------------------------------------------------------------------------
1186------------------------------------------------------------------------------
1188MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
1189MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
1191------------------------------------------------------------------------------
1192-- Serially Zipping Streams
1193------------------------------------------------------------------------------
1195-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
1196-- i.e. it produces one element from each stream serially and then zips all
1200-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1201-- where s1 = fromFoldable [1, 2]
1202-- s2 = fromFoldable [3, 4]
1203-- s3 = fromFoldable [5, 6]
1209-- The 'Semigroup' instance of this type works the same way as that of
1213newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: S.Stream m a}
1214 deriving (Functor, Semigroup, Monoid)
1218{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
1219type ZipStream = ZipSerialM
1221instance IsStream ZipSerialM where
1222 toStream = getZipSerialM
1223 fromStream = ZipSerialM
1225 {-# INLINE consM #-}
1226 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1227 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1231 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1232 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1235instance Monad m => Applicative (ZipSerialM m) where
1236 pure = ZipSerialM . S.repeat
1237 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
1239------------------------------------------------------------------------------
1240-- Parallely Zipping Streams
1241------------------------------------------------------------------------------
1243-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
1244-- be zipped concurrently.
1247-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1248-- where s1 = fromFoldable [1, 2]
1249-- s2 = fromFoldable [3, 4]
1250-- s3 = fromFoldable [5, 6]
1256-- The 'Semigroup' instance of this type works the same way as that of
1260newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: S.Stream m a}
1261 deriving (Functor, Semigroup, Monoid)
1263instance IsStream ZipAsyncM where
1264 toStream = getZipAsyncM
1265 fromStream = ZipAsyncM
1267 {-# INLINE consM #-}
1268 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1269 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1273 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1274 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1277instance MonadAsync m => Applicative (ZipAsyncM m) where
1278 pure = ZipAsyncM . S.repeat
1279 m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)
1281-------------------------------------------------------------------------------
1282-- Type adapting combinators
1283-------------------------------------------------------------------------------
1285-- | Adapt any specific stream type to any other specific stream type.
1288adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
1289adapt = fromStream . toStream
1291-- | Fix the type of a polymorphic stream as 'SerialT'.
1294serially :: IsStream t => SerialT m a -> t m a
1297-- | Fix the type of a polymorphic stream as 'WSerialT'.
1300wSerially :: IsStream t => WSerialT m a -> t m a
1303-- | Same as 'wSerially'.
1306{-# DEPRECATED interleaving "Please use wSerially instead." #-}
1307interleaving :: IsStream t => WSerialT m a -> t m a
1308interleaving = wSerially
1310-- | Fix the type of a polymorphic stream as 'AheadT'.
1313aheadly :: IsStream t => AheadT m a -> t m a
1316-- | Fix the type of a polymorphic stream as 'AsyncT'.
1319asyncly :: IsStream t => AsyncT m a -> t m a
1322-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1325wAsyncly :: IsStream t => WAsyncT m a -> t m a
1328-- | Fix the type of a polymorphic stream as 'ParallelT'.
1331parallely :: IsStream t => ParallelT m a -> t m a
1334-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
1337zipSerially :: IsStream t => ZipSerialM m a -> t m a
1340-- | Same as 'zipSerially'.
1343{-# DEPRECATED zipping "Please use zipSerially instead." #-}
1344zipping :: IsStream t => ZipSerialM m a -> t m a
1345zipping = zipSerially
1347-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
1350zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
1353-- | Same as 'zipAsyncly'.
1356{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
1357zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
1358zippingAsync = zipAsyncly
1360-------------------------------------------------------------------------------
1361-- Running Streams, convenience functions specialized to types
1362-------------------------------------------------------------------------------
1364-- | Same as @runStream@.
1367{-# DEPRECATED runStreamT "Please use runStream instead." #-}
1368runStreamT :: Monad m => SerialT m a -> m ()
1369runStreamT = runStream
1371-- | Same as @runStream . wSerially@.
1374{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
1375runInterleavedT :: Monad m => InterleavedT m a -> m ()
1376runInterleavedT = runStream . wSerially
1378-- | Same as @runStream . asyncly@.
1381{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}
1382runAsyncT :: Monad m => AsyncT m a -> m ()
1383runAsyncT = runStream . asyncly
1385-- | Same as @runStream . parallely@.
1388{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
1389runParallelT :: Monad m => ParallelT m a -> m ()
1390runParallelT = runStream . parallely
1392-- | Same as @runStream . zipping@.
1395{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}
1396runZipStream :: Monad m => ZipSerialM m a -> m ()
1397runZipStream = runStream . zipSerially
1399-- | Same as @runStream . zippingAsync@.
1402{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}
1403runZipAsync :: Monad m => ZipAsyncM m a -> m ()
1404runZipAsync = runStream . zipAsyncly
1406------------------------------------------------------------------------------
1408------------------------------------------------------------------------------
1410-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
1414type Serial a = SerialT IO a
1416-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
1417-- documentation for more details.
1420type WSerial a = WSerialT IO a
1422-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
1423-- 'AheadT' documentation for more details.
1426type Ahead a = AheadT IO a
1428-- | A demand driven left biased parallely composing IO stream of elements of
1429-- type @a@. See 'AsyncT' documentation for more details.
1432type Async a = AsyncT IO a
1434-- | A round robin parallely composing IO stream of elements of type @a@.
1435-- See 'WAsyncT' documentation for more details.
1438type WAsync a = WAsyncT IO a
1440-- | A parallely composing IO stream of elements of type @a@.
1441-- See 'ParallelT' documentation for more details.
1444type Parallel a = ParallelT IO a
1446-- | An IO stream whose applicative instance zips streams serially.
1449type ZipSerial a = ZipSerialM IO a
1451-- | An IO stream whose applicative instance zips streams wAsyncly.
1454type ZipAsync a = ZipAsyncM IO a
1456------------------------------------------------------------------------------
1458------------------------------------------------------------------------------
1460-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
1461-- container of streams using the specified stream sum operation.
1463-- @foldWith 'async' $ map return [1..3]@
1466{-# INLINABLE foldWith #-}
1467foldWith :: (IsStream t, Foldable f)
1468 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
1469foldWith f = foldr f nil
1471-- | A variant of 'foldMap' that allows you to map a monadic streaming action
1472-- on a 'Foldable' container and then fold it using the specified stream sum
1475-- @foldMapWith 'async' return [1..3]@
1478{-# INLINABLE foldMapWith #-}
1479foldMapWith :: (IsStream t, Foldable f)
1480 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1481foldMapWith f g = foldr (f . g) nil
1483-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1484-- monadic streaming function is the last argument.
1487{-# INLINABLE forEachWith #-}
1488forEachWith :: (IsStream t, Foldable f)
1489 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1490forEachWith f xs g = foldr (f . g) nil xs
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