@@ -1,1489 +0,0 @@
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
22−module 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
111−import Control.Monad (ap)
112−import Control.Monad.Base (MonadBase (..), liftBaseDefault)
113−import Control.Monad.Catch (MonadThrow, throwM)
114−-- import Control.Monad.Error.Class (MonadError(..))
115−import Control.Monad.IO.Class (MonadIO(..))
116−import Control.Monad.Reader.Class (MonadReader(..))
117−import Control.Monad.State.Class (MonadState(..))
118−import Control.Monad.Trans.Class (MonadTrans (lift))
119−import Data.Semigroup (Semigroup(..))
120−import Streamly.SVar (MonadAsync, SVar, toStreamVar)
122−import qualified Streamly.CPS.Stream as C
123−import qualified Streamly.CPS.Concurrent as CC
125−------------------------------------------------------------------------------
126−-- Types that can behave as a Stream
127−------------------------------------------------------------------------------
132−-- | Class of types that can represent a stream of elements of some type 'a' in
136−class IsStream t where
137− toStream :: t m a -> C.Stream m a
138− fromStream :: C.Stream m a -> t m a
139− -- | Constructs a stream by adding a monadic action at the head of an
140− -- existing stream. For example:
143− -- > toList $ getLine \`consM` getLine \`consM` nil
146− -- ["hello","world"]
149− -- /Concurrent (do not use 'parallely' to construct infinite streams)/
152− consM :: MonadAsync m => m a -> t m a -> t m a
153− -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
154− -- to remember that @|@ comes before ':'.
157− -- > toList $ getLine |: getLine |: nil
160− -- ["hello","world"]
164− -- let delay = threadDelay 1000000 >> print 1
165− -- runStream $ serially $ delay |: delay |: delay |: nil
166− -- runStream $ parallely $ delay |: delay |: delay |: nil
169− -- /Concurrent (do not use 'parallely' to construct infinite streams)/
172− (|:) :: MonadAsync m => m a -> t m a -> t m a
173− -- We can define (|:) just as 'consM' but it is defined explicitly for each
174− -- type because we want to use SPECIALIZE pragma on the definition.
176−-- | Same as 'IsStream'.
179−{-# DEPRECATED Streaming "Please use IsStream instead." #-}
180−type Streaming = IsStream
182−------------------------------------------------------------------------------
183−-- Constructing a stream
184−------------------------------------------------------------------------------
186−-- | An empty stream.
194−nil :: IsStream t => t m a
195−nil = fromStream C.nil
197−-- | Constructs a stream by adding a monadic action at the head of an existing
198−-- stream. For example:
201−-- > toList $ getLine \`consM` getLine \`consM` nil
208−consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
209−consMSerial m r = fromStream $ C.consM m (toStream r)
213−-- | Construct a stream by adding a pure value at the head of an existing
214−-- stream. For pure values it can be faster than 'consM'. For example:
217−-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
222−cons :: IsStream t => a -> t m a -> t m a
223−cons a r = fromStream $ C.cons a (toStream r)
227−-- | Operator equivalent of 'cons'.
230−-- > toList $ 1 .: 2 .: 3 .: nil
235−(.:) :: IsStream t => a -> t m a -> t m a
238−-- | Build a stream from its church encoding. The function passed maps
239−-- directly to the underlying representation of the stream type. The second
240−-- parameter to the function is the "yield" function yielding a value and the
241−-- remaining stream if any otherwise 'Nothing'. The third parameter is to
242−-- represent an "empty" stream.
243−streamBuild :: IsStream t
244− => (forall r. Maybe (SVar C.Stream m a)
245− -> (a -> t m a -> m r)
250−streamBuild k = fromStream $ C.Stream $ \svr stp sng yld ->
251− let yield a r = yld a (toStream r)
252− in k svr yield sng stp
254−-- | Build a singleton stream from a callback function.
255−fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
256−fromCallback k = fromStream $ C.Stream $ \_ _ sng _ -> k sng
258−-- | Read an SVar to get a stream.
259−fromSVar :: (MonadAsync m, IsStream t) => SVar C.Stream m a -> t m a
260−fromSVar sv = fromStream $ CC.fromStreamVar sv
262−------------------------------------------------------------------------------
263−-- Destroying a stream
264−------------------------------------------------------------------------------
266−-- | Fold a stream using its church encoding. The second argument is the "step"
267−-- function consuming an element and the remaining stream, if any. The third
268−-- argument is for consuming an "empty" stream that yields nothing.
271− => Maybe (SVar C.Stream m a)
272− -> (a -> t m a -> m r)
277−streamFold svr step single blank m =
278− let yield a x = step a (fromStream x)
279− in (C.runStream (toStream m)) svr blank single yield
281−-- | Run a streaming composition, discard the results. By default it interprets
282−-- the stream as 'SerialT', to run other types of streams use the type adapting
283−-- combinators for example @runStream . 'asyncly'@.
286−runStream :: Monad m => SerialT m a -> m ()
287−runStream m = go (toStream m)
290− let stop = return ()
291− single _ = return ()
293− in (C.runStream m1) Nothing stop single yield
295−-- | Same as 'runStream'
298−{-# DEPRECATED runStreaming "Please use runStream instead." #-}
299−runStreaming :: (Monad m, IsStream t) => t m a -> m ()
300−runStreaming = runStream . adapt
302−-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
303−-- be read back from the SVar using 'fromSVar'.
304−toSVar :: (IsStream t, MonadAsync m) => SVar C.Stream m a -> t m a -> m ()
305−toSVar sv m = toStreamVar sv (toStream m)
307−------------------------------------------------------------------------------
309−------------------------------------------------------------------------------
311−-- XXX Get rid of this?
312−-- | Make a stream asynchronous, triggers the computation and returns a stream
313−-- in the underlying monad representing the output generated by the original
314−-- computation. The returned action is exhaustible and must be drained once. If
315−-- not drained fully we may have a thread blocked forever and once exhausted it
316−-- will always return 'empty'.
319−mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
321− s <- CC.mkParallel (toStream m)
322− return $ fromStream s
324−{-# INLINE applyWith #-}
325−applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
326−applyWith f x = fromStream $
327− CC.applyWith (toStream . f . fromStream) (toStream x)
329−{-# INLINE runWith #-}
330−runWith :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
331−runWith f x = CC.runWith (f . fromStream) (toStream x)
339−-- | Parallel function application operator for streams; just like the regular
340−-- function application operator '$' except that it is concurrent. The
341−-- following code prints a value every second even though each stage adds a 1
347−-- S.mapM (\\x -> threadDelay 1000000 >> print x)
348−-- |$ S.repeatM (threadDelay 1000000 >> return 1)
355−(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
356−f |$ x = applyWith f x
358−-- | Parallel reverse function application operator for streams; just like the
359−-- regular reverse function application operator '&' except that it is
364−-- S.repeatM (threadDelay 1000000 >> return 1)
365−-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
372−(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
375−-- | Parallel function application operator; applies a @run@ or @fold@ function
376−-- to a stream such that the fold consumer and the stream producer run in
377−-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
378−-- underlying monad. The @.@ at the end of the operator is a mnemonic for
379−-- termination of the stream.
382−-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
383−-- |$. S.repeatM (threadDelay 1000000 >> return 1)
390−(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
391−f |$. x = runWith f x
393−-- | Parallel reverse function application operator for applying a run or fold
394−-- functions to a stream. Just like '|$.' except that the operands are reversed.
397−-- S.repeatM (threadDelay 1000000 >> return 1)
398−-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
405−(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
408−------------------------------------------------------------------------------
409−-- CPP macros for common instances
410−------------------------------------------------------------------------------
412−-- XXX use template haskell instead and include Monoid and IsStream instances
415−withLocal :: MonadReader r m => (r -> r) -> C.Stream m a -> C.Stream m a
417− C.Stream $ \_ stp sng yld ->
418− let single = local f . sng
419− yield a r = local f $ yld a (withLocal f r)
420− in (C.runStream m) Nothing (local f stp) single yield
423−-- XXX handle and test cross thread state transfer
426− => C.Stream m a -> (e -> C.Stream m a) -> C.Stream m a
428− C.Stream $ \_ stp sng yld ->
429− let run x = C.runStream x Nothing stp sng yield
430− handle r = r `catchError` \e -> run $ h e
431− yield a r = yld a (withCatchError r h)
435−#define MONADPARALLEL , MonadAsync m
437−#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
438−instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
439− pure = STREAM . C.singleton; \
442−#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
443−instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
444− liftBase = liftBaseDefault }; \
446−instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
447− liftIO = lift . liftIO }; \
449−instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
450− throwM = lift . throwM }; \
453−instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
454− throwError = lift . throwError; \
456− fromStream $ withCatchError (toStream m) (\e -> toStream $ h e) }; \
459−instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
461− local f m = fromStream $ withLocal f (toStream m) }; \
463−instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
465− put x = lift (put x); \
466− state k = lift (state k) }
468−------------------------------------------------------------------------------
470−------------------------------------------------------------------------------
472−-- | Deep serial composition or serial composition with depth first traversal.
473−-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
474−-- depth first manner, yielding all elements from the first stream, and then
475−-- all elements from the second stream.
479−-- import qualified "Streamly.Prelude" as S
481−-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
487−-- The 'Monad' instance runs the /monadic continuation/ for each
488−-- element of the stream, serially.
491−-- main = 'runStream' . 'serially' $ do
492−-- x <- return 1 \<\> return 2
500−-- 'SerialT' nests streams serially in a depth first manner.
503−-- main = 'runStream' . 'serially' $ do
504−-- x <- return 1 \<\> return 2
505−-- y <- return 3 \<\> return 4
506−-- S.once $ print (x, y)
515−-- This behavior of 'SerialT' is exactly like a list transformer. We call the
516−-- monadic code being run for each element of the stream a monadic
517−-- continuation. In imperative paradigm we can think of this composition as
518−-- nested @for@ loops and the monadic continuation is the body of the loop. The
519−-- loop iterates for all elements of the stream.
521−-- The 'serially' combinator can be omitted as the default stream type is
523−-- Note that serial composition with depth first traversal can be used to
524−-- combine an infinite number of streams as it explores only one stream at a
528−newtype SerialT m a = SerialT {getSerialT :: C.Stream m a}
529− deriving (Semigroup, Monoid, Functor, MonadTrans)
533−{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
534−type StreamT = SerialT
536−instance IsStream SerialT where
537− toStream = getSerialT
538− fromStream = SerialT
540− {-# INLINE consM #-}
541− {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
542− consM :: Monad m => m a -> SerialT m a -> SerialT m a
546− {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
547− (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
550−------------------------------------------------------------------------------
552−------------------------------------------------------------------------------
554−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
555−-- Appends two streams sequentially, yielding all elements from the first
556−-- stream, and then all elements from the second stream.
559−{-# INLINE serial #-}
560−serial :: IsStream t => t m a -> t m a -> t m a
561−serial m1 m2 = fromStream $ C.serial (toStream m1) (toStream m2)
563−------------------------------------------------------------------------------
565−------------------------------------------------------------------------------
567−instance Monad m => Monad (SerialT m) where
569− (SerialT (C.Stream m)) >>= f = SerialT $ C.Stream $ \_ stp sng yld ->
570− let run x = (C.runStream x) Nothing stp sng yld
571− single a = run $ toStream (f a)
572− yield a r = run $ toStream $ f a <> (fromStream r >>= f)
573− in m Nothing stp single yield
575−------------------------------------------------------------------------------
577−------------------------------------------------------------------------------
579−MONAD_APPLICATIVE_INSTANCE(SerialT,)
580−MONAD_COMMON_INSTANCES(SerialT,)
582−------------------------------------------------------------------------------
584−------------------------------------------------------------------------------
586−-- | Wide serial composition or serial composition with a breadth first
587−-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
588−-- the two streams in a breadth first manner. In other words, it interleaves
589−-- two streams, yielding one element from each stream alternately.
593−-- import qualified "Streamly.Prelude" as S
595−-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
601−-- Similarly, the 'Monad' instance interleaves the iterations of the
602−-- inner and the outer loop, nesting loops in a breadth first manner.
606−-- main = 'runStream' . 'wSerially' $ do
607−-- x <- return 1 \<\> return 2
608−-- y <- return 3 \<\> return 4
609−-- S.once $ print (x, y)
618−-- Note that a serial composition with breadth first traversal can only combine
619−-- a finite number of streams as it needs to retain state for each unfinished
623−newtype WSerialT m a = WSerialT {getWSerialT :: C.Stream m a}
624− deriving (Functor, MonadTrans)
628−{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
629−type InterleavedT = WSerialT
631−instance IsStream WSerialT where
632− toStream = getWSerialT
633− fromStream = WSerialT
635− {-# INLINE consM #-}
636− {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
637− consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
641− {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
642− (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
645−------------------------------------------------------------------------------
647−------------------------------------------------------------------------------
649−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
650−-- Interleaves two streams, yielding one element from each stream alternately.
653−{-# INLINE wSerial #-}
654−wSerial :: IsStream t => t m a -> t m a -> t m a
655−wSerial m1 m2 = fromStream $ C.wSerial (toStream m1) (toStream m2)
657−instance Semigroup (WSerialT m a) where
662−-- | Same as 'wSerial'.
665−{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
667−(<=>) :: IsStream t => t m a -> t m a -> t m a
670−------------------------------------------------------------------------------
672−------------------------------------------------------------------------------
674−instance Monoid (WSerialT m a) where
678−------------------------------------------------------------------------------
680−------------------------------------------------------------------------------
682−instance Monad m => Monad (WSerialT m) where
684− (WSerialT (C.Stream m)) >>= f = WSerialT $ C.Stream $ \_ stp sng yld ->
685− let run x = (C.runStream x) Nothing stp sng yld
686− single a = run $ toStream (f a)
687− yield a r = run $ toStream $ f a <> (fromStream r >>= f)
688− in m Nothing stp single yield
690−------------------------------------------------------------------------------
692−------------------------------------------------------------------------------
694−MONAD_APPLICATIVE_INSTANCE(WSerialT,)
695−MONAD_COMMON_INSTANCES(WSerialT,)
697−------------------------------------------------------------------------------
699−------------------------------------------------------------------------------
701−-- | Deep ahead composition or ahead composition with depth first traversal.
702−-- The semigroup composition of 'AheadT' appends streams in a depth first
703−-- manner just like 'SerialT' except that it can produce elements concurrently
704−-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
705−-- as it arrives whereas 'AheadT' orders the output in the traversal order.
708−-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
714−-- Any exceptions generated by a constituent stream are propagated to the
717−-- Similarly, the monad instance of 'AheadT' may run each iteration
718−-- concurrently ahead of time but presents the results in the same order as
723−-- import qualified "Streamly.Prelude" as S
724−-- import Control.Concurrent
726−-- main = 'runStream' . 'aheadly' $ do
727−-- n <- return 3 \<\> return 2 \<\> return 1
729−-- threadDelay (n * 1000000)
730−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
733−-- ThreadId 40: Delay 1
734−-- ThreadId 39: Delay 2
735−-- ThreadId 38: Delay 3
738−-- All iterations may run in the same thread if they do not block.
740−-- Note that ahead composition with depth first traversal can be used to
741−-- combine infinite number of streams as it explores only a bounded number of
742−-- streams at a time.
745−newtype AheadT m a = AheadT {getAheadT :: C.Stream m a}
746− deriving (Functor, MonadTrans)
748−instance IsStream AheadT where
749− toStream = getAheadT
752− {-# INLINE consM #-}
753− {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
754− consM m r = fromStream $ CC.consMAhead m (toStream r)
757− {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
760−------------------------------------------------------------------------------
762−------------------------------------------------------------------------------
764−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
765−-- Merges two streams sequentially but with concurrent lookahead.
769−ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
770−ahead m1 m2 = fromStream $ CC.ahead (toStream m1) (toStream m2)
772−instance MonadAsync m => Semigroup (AheadT m a) where
775−------------------------------------------------------------------------------
777−------------------------------------------------------------------------------
779−instance MonadAsync m => Monoid (AheadT m a) where
783−------------------------------------------------------------------------------
785−------------------------------------------------------------------------------
787−{-# INLINE aheadbind #-}
791− -> (a -> C.Stream m b)
796− C.Stream $ \ctx stp sng yld ->
797− let run x = (C.runStream x) ctx stp sng yld
798− single a = run $ f a
799− yield a r = run $ f a `CC.ahead` go r
800− in g Nothing stp single yield
802−instance MonadAsync m => Monad (AheadT m) where
804− (AheadT m) >>= f = AheadT $ aheadbind m (getAheadT . f)
806−------------------------------------------------------------------------------
808−------------------------------------------------------------------------------
810−MONAD_APPLICATIVE_INSTANCE(AheadT,MONADPARALLEL)
811−MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
813−------------------------------------------------------------------------------
815−------------------------------------------------------------------------------
817−-- | Deep async composition or async composition with depth first traversal. In
818−-- a left to right 'Semigroup' composition it tries to yield elements from the
819−-- left stream as long as it can, but it can run the right stream in parallel
820−-- if it needs to, based on demand. The right stream can be run if the left
821−-- stream blocks on IO or cannot produce elements fast enough for the consumer.
824−-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
830−-- Any exceptions generated by a constituent stream are propagated to the
831−-- output stream. The output and exceptions from a single stream are guaranteed
832−-- to arrive in the same order in the resulting stream as they were generated
833−-- in the input stream. However, the relative ordering of elements from
834−-- different streams in the resulting stream can vary depending on scheduling
835−-- and generation delays.
837−-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
838−-- concurrently based on demand. More concurrent iterations are started only
839−-- if the previous iterations are not able to produce enough output for the
844−-- import qualified "Streamly.Prelude" as S
845−-- import Control.Concurrent
847−-- main = 'runStream' . 'asyncly' $ do
848−-- n <- return 3 \<\> return 2 \<\> return 1
850−-- threadDelay (n * 1000000)
851−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
854−-- ThreadId 40: Delay 1
855−-- ThreadId 39: Delay 2
856−-- ThreadId 38: Delay 3
859−-- All iterations may run in the same thread if they do not block.
861−-- Note that async composition with depth first traversal can be used to
862−-- combine infinite number of streams as it explores only a bounded number of
863−-- streams at a time.
866−newtype AsyncT m a = AsyncT {getAsyncT :: C.Stream m a}
867− deriving (Functor, MonadTrans)
869−instance IsStream AsyncT where
870− toStream = getAsyncT
873− {-# INLINE consM #-}
874− {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
875− consM m r = fromStream $ CC.consMAsync m (toStream r)
878− {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
881−------------------------------------------------------------------------------
883−------------------------------------------------------------------------------
885−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
886−-- Merges two streams possibly concurrently, preferring the
887−-- elements from the left one when available.
891−async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
892−async m1 m2 = fromStream $ CC.async (toStream m1) (toStream m2)
894−-- | Same as 'async'.
897−{-# DEPRECATED (<|) "Please use 'async' instead." #-}
899−(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
902−instance MonadAsync m => Semigroup (AsyncT m a) where
905−------------------------------------------------------------------------------
907−------------------------------------------------------------------------------
909−instance MonadAsync m => Monoid (AsyncT m a) where
913−------------------------------------------------------------------------------
915−------------------------------------------------------------------------------
917−{-# INLINE parbind #-}
919− :: (forall c. C.Stream m c -> C.Stream m c -> C.Stream m c)
921− -> (a -> C.Stream m b)
923−parbind par m f = go m
926− C.Stream $ \ctx stp sng yld ->
927− let run x = (C.runStream x) ctx stp sng yld
928− single a = run $ f a
929− yield a r = run $ f a `par` go r
930− in g Nothing stp single yield
932−instance MonadAsync m => Monad (AsyncT m) where
934− (AsyncT m) >>= f = AsyncT $ parbind CC.async m (getAsyncT . f)
936−------------------------------------------------------------------------------
938−------------------------------------------------------------------------------
940−MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
941−MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
943−------------------------------------------------------------------------------
945−------------------------------------------------------------------------------
947−-- | Wide async composition or async composition with breadth first traversal.
948−-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
949−-- streams using a depth first travesal or in a round robin fashion, yielding
950−-- elements from both streams alternately.
953−-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
959−-- Any exceptions generated by a constituent stream are propagated to the
960−-- output stream. The output and exceptions from a single stream are guaranteed
961−-- to arrive in the same order in the resulting stream as they were generated
962−-- in the input stream. However, the relative ordering of elements from
963−-- different streams in the resulting stream can vary depending on scheduling
964−-- and generation delays.
966−-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
967−-- concurrently using a round robin scheduling.
971−-- import qualified "Streamly.Prelude" as S
972−-- import Control.Concurrent
974−-- main = 'runStream' . 'wAsyncly' $ do
975−-- n <- return 3 \<\> return 2 \<\> return 1
977−-- threadDelay (n * 1000000)
978−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
981−-- ThreadId 40: Delay 1
982−-- ThreadId 39: Delay 2
983−-- ThreadId 38: Delay 3
986−-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
987−-- concurrently, unconditionally.
989−-- Note that async composition with breadth first traversal can only combine a
990−-- finite number of streams as it needs to retain state for each unfinished
994−newtype WAsyncT m a = WAsyncT {getWAsyncT :: C.Stream m a}
995− deriving (Functor, MonadTrans)
997−instance IsStream WAsyncT where
998− toStream = getWAsyncT
999− fromStream = WAsyncT
1001− {-# INLINE consM #-}
1002− {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1003− consM m r = fromStream $ CC.consMWAsync m (toStream r)
1005− {-# INLINE (|:) #-}
1006− {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1009−------------------------------------------------------------------------------
1011−------------------------------------------------------------------------------
1013−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
1014−-- Merges two streams concurrently choosing elements from both fairly.
1017−{-# INLINE wAsync #-}
1018−wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1019−wAsync m1 m2 = fromStream $ CC.wAsync (toStream m1) (toStream m2)
1021−instance MonadAsync m => Semigroup (WAsyncT m a) where
1024−------------------------------------------------------------------------------
1026−------------------------------------------------------------------------------
1028−instance MonadAsync m => Monoid (WAsyncT m a) where
1032−------------------------------------------------------------------------------
1034−------------------------------------------------------------------------------
1036−instance MonadAsync m => Monad (WAsyncT m) where
1038− (WAsyncT m) >>= f =
1039− WAsyncT $ parbind CC.wAsync m (getWAsyncT . f)
1041−------------------------------------------------------------------------------
1043−------------------------------------------------------------------------------
1045−MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
1046−MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
1048−------------------------------------------------------------------------------
1050−------------------------------------------------------------------------------
1052−-- | Async composition with simultaneous traversal of all streams.
1054−-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
1055−-- running both strictly concurrently and yielding elements from both streams
1056−-- as they arrive. When multiple streams are combined using 'ParallelT' each
1057−-- one is evaluated in its own thread and the results produced are presented in
1058−-- the combined stream on a first come first serve basis.
1060−-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
1061−-- specific type of consumption pattern (depth first or breadth first). Since
1062−-- they are lookahead, they may introduce certain default latency in starting
1063−-- more concurrent tasks for efficiency reasons or may put a default limitation
1064−-- on the resource consumption (e.g. number of concurrent threads for
1065−-- lookahead). If we look at the implementation detail, they both can share a
1066−-- pool of worker threads to evaluate the streams in the desired pattern and at
1067−-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
1068−-- evaluate each stream.
1070−-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
1071−-- constituent streams fairly in a round robin fashion.
1072−-- However, the key difference is that 'WAsyncT' is lazy or pull driven
1073−-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
1074−-- starts concurrent evaluation of both the streams (in separate threads) and
1075−-- later picks the results whereas 'WAsyncT' may wait for a certain latency
1076−-- threshold before initiating concurrent evaluation of the next stream. The
1077−-- concurrent scheduling of the next stream or the degree of concurrency is
1078−-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
1079−-- is evaluated in a separate thread and results are /pushed/ to a shared
1080−-- output buffer, the evaluation rate is controlled by blocking when the buffer
1083−-- Concurrent lookahead streams are generally more efficient than
1084−-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
1085−-- they do not necessarily use a separate thread for each task. So they should
1086−-- be preferred over 'ParallelT' especially when efficiency is a concern and
1087−-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
1088−-- for cases when the streams are required to be evaluated simultaneously
1089−-- irrespective of how the consumer consumes them e.g. when we want to race
1090−-- two tasks and want to start both strictly at the same time or if we have
1091−-- timers in the parallel tasks and our results depend on the timers being
1092−-- started at the same time. We can say that 'ParallelT' is almost the same
1093−-- (modulo some implementation differences) as 'WAsyncT' when the latter is
1094−-- used with unlimited lookahead and zero latency in initiating lookahead.
1097−-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
1103−-- When streams with more than one element are merged, it yields whichever
1104−-- stream yields first without any bias, unlike the 'Async' style streams.
1106−-- Any exceptions generated by a constituent stream are propagated to the
1107−-- output stream. The output and exceptions from a single stream are guaranteed
1108−-- to arrive in the same order in the resulting stream as they were generated
1109−-- in the input stream. However, the relative ordering of elements from
1110−-- different streams in the resulting stream can vary depending on scheduling
1111−-- and generation delays.
1113−-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
1114−-- of the loop concurrently.
1117−-- import "Streamly"
1118−-- import qualified "Streamly.Prelude" as S
1119−-- import Control.Concurrent
1121−-- main = 'runStream' . 'parallely' $ do
1122−-- n <- return 3 \<\> return 2 \<\> return 1
1124−-- threadDelay (n * 1000000)
1125−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1128−-- ThreadId 40: Delay 1
1129−-- ThreadId 39: Delay 2
1130−-- ThreadId 38: Delay 3
1133−-- Note that parallel composition can only combine a finite number of
1134−-- streams as it needs to retain state for each unfinished stream.
1137−newtype ParallelT m a = ParallelT {getParallelT :: C.Stream m a}
1138− deriving (Functor, MonadTrans)
1140−instance IsStream ParallelT where
1141− toStream = getParallelT
1142− fromStream = ParallelT
1144− {-# INLINE consM #-}
1145− {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1146− consM m r = fromStream $ CC.consMParallel m (toStream r)
1148− {-# INLINE (|:) #-}
1149− {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1152−------------------------------------------------------------------------------
1154−------------------------------------------------------------------------------
1156−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
1157−-- Merges two streams concurrently.
1160−{-# INLINE parallel #-}
1161−parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1162−parallel m1 m2 = fromStream $ CC.parallel (toStream m1) (toStream m2)
1164−instance MonadAsync m => Semigroup (ParallelT m a) where
1167−------------------------------------------------------------------------------
1169−------------------------------------------------------------------------------
1171−instance MonadAsync m => Monoid (ParallelT m a) where
1175−------------------------------------------------------------------------------
1177−------------------------------------------------------------------------------
1179−instance MonadAsync m => Monad (ParallelT m) where
1181− (ParallelT m) >>= f = ParallelT $ parbind CC.parallel m (getParallelT . f)
1183−------------------------------------------------------------------------------
1185−------------------------------------------------------------------------------
1187−MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
1188−MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
1190−------------------------------------------------------------------------------
1191−-- Serially Zipping Streams
1192−------------------------------------------------------------------------------
1194−-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
1195−-- i.e. it produces one element from each stream serially and then zips all
1199−-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1200−-- where s1 = fromFoldable [1, 2]
1201−-- s2 = fromFoldable [3, 4]
1202−-- s3 = fromFoldable [5, 6]
1205−-- [(1,3,5),(2,4,6)]
1208−-- The 'Semigroup' instance of this type works the same way as that of
1212−newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: C.Stream m a}
1213− deriving (Functor, Semigroup, Monoid)
1217−{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
1218−type ZipStream = ZipSerialM
1220−instance IsStream ZipSerialM where
1221− toStream = getZipSerialM
1222− fromStream = ZipSerialM
1224− {-# INLINE consM #-}
1225− {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1226− consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1227− consM = consMSerial
1229− {-# INLINE (|:) #-}
1230− {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1231− (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1234−instance Monad m => Applicative (ZipSerialM m) where
1235− pure = ZipSerialM . C.repeat
1236− m1 <*> m2 = fromStream $ C.zipWith id (toStream m1) (toStream m2)
1238−------------------------------------------------------------------------------
1239−-- Parallely Zipping Streams
1240−------------------------------------------------------------------------------
1242−-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
1243−-- be zipped concurrently.
1246−-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1247−-- where s1 = fromFoldable [1, 2]
1248−-- s2 = fromFoldable [3, 4]
1249−-- s3 = fromFoldable [5, 6]
1252−-- [(1,3,5),(2,4,6)]
1255−-- The 'Semigroup' instance of this type works the same way as that of
1259−newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: C.Stream m a}
1260− deriving (Functor, Semigroup, Monoid)
1262−instance IsStream ZipAsyncM where
1263− toStream = getZipAsyncM
1264− fromStream = ZipAsyncM
1266− {-# INLINE consM #-}
1267− {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1268− consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1269− consM = consMSerial
1271− {-# INLINE (|:) #-}
1272− {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1273− (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1276−instance MonadAsync m => Applicative (ZipAsyncM m) where
1277− pure = ZipAsyncM . C.repeat
1278− m1 <*> m2 = fromStream $ CC.zipAsyncWith id (toStream m1) (toStream m2)
1280−-------------------------------------------------------------------------------
1281−-- Type adapting combinators
1282−-------------------------------------------------------------------------------
1284−-- | Adapt any specific stream type to any other specific stream type.
1287−adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
1288−adapt = fromStream . toStream
1290−-- | Fix the type of a polymorphic stream as 'SerialT'.
1293−serially :: IsStream t => SerialT m a -> t m a
1296−-- | Fix the type of a polymorphic stream as 'WSerialT'.
1299−wSerially :: IsStream t => WSerialT m a -> t m a
1302−-- | Same as 'wSerially'.
1305−{-# DEPRECATED interleaving "Please use wSerially instead." #-}
1306−interleaving :: IsStream t => WSerialT m a -> t m a
1307−interleaving = wSerially
1309−-- | Fix the type of a polymorphic stream as 'AheadT'.
1312−aheadly :: IsStream t => AheadT m a -> t m a
1315−-- | Fix the type of a polymorphic stream as 'AsyncT'.
1318−asyncly :: IsStream t => AsyncT m a -> t m a
1321−-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1324−wAsyncly :: IsStream t => WAsyncT m a -> t m a
1327−-- | Fix the type of a polymorphic stream as 'ParallelT'.
1330−parallely :: IsStream t => ParallelT m a -> t m a
1333−-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
1336−zipSerially :: IsStream t => ZipSerialM m a -> t m a
1339−-- | Same as 'zipSerially'.
1342−{-# DEPRECATED zipping "Please use zipSerially instead." #-}
1343−zipping :: IsStream t => ZipSerialM m a -> t m a
1344−zipping = zipSerially
1346−-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
1349−zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
1352−-- | Same as 'zipAsyncly'.
1355−{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
1356−zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
1357−zippingAsync = zipAsyncly
1359−-------------------------------------------------------------------------------
1360−-- Running Streams, convenience functions specialized to types
1361−-------------------------------------------------------------------------------
1363−-- | Same as @runStream@.
1366−{-# DEPRECATED runStreamT "Please use runStream instead." #-}
1367−runStreamT :: Monad m => SerialT m a -> m ()
1368−runStreamT = runStream
1370−-- | Same as @runStream . wSerially@.
1373−{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
1374−runInterleavedT :: Monad m => InterleavedT m a -> m ()
1375−runInterleavedT = runStream . wSerially
1377−-- | Same as @runStream . asyncly@.
1380−{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}
1381−runAsyncT :: Monad m => AsyncT m a -> m ()
1382−runAsyncT = runStream . asyncly
1384−-- | Same as @runStream . parallely@.
1387−{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
1388−runParallelT :: Monad m => ParallelT m a -> m ()
1389−runParallelT = runStream . parallely
1391−-- | Same as @runStream . zipping@.
1394−{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}
1395−runZipStream :: Monad m => ZipSerialM m a -> m ()
1396−runZipStream = runStream . zipSerially
1398−-- | Same as @runStream . zippingAsync@.
1401−{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}
1402−runZipAsync :: Monad m => ZipAsyncM m a -> m ()
1403−runZipAsync = runStream . zipAsyncly
1405−------------------------------------------------------------------------------
1407−------------------------------------------------------------------------------
1409−-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
1410−-- for more details.
1413−type Serial a = SerialT IO a
1415−-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
1416−-- documentation for more details.
1419−type WSerial a = WSerialT IO a
1421−-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
1422−-- 'AheadT' documentation for more details.
1425−type Ahead a = AheadT IO a
1427−-- | A demand driven left biased parallely composing IO stream of elements of
1428−-- type @a@. See 'AsyncT' documentation for more details.
1431−type Async a = AsyncT IO a
1433−-- | A round robin parallely composing IO stream of elements of type @a@.
1434−-- See 'WAsyncT' documentation for more details.
1437−type WAsync a = WAsyncT IO a
1439−-- | A parallely composing IO stream of elements of type @a@.
1440−-- See 'ParallelT' documentation for more details.
1443−type Parallel a = ParallelT IO a
1445−-- | An IO stream whose applicative instance zips streams serially.
1448−type ZipSerial a = ZipSerialM IO a
1450−-- | An IO stream whose applicative instance zips streams wAsyncly.
1453−type ZipAsync a = ZipAsyncM IO a
1455−------------------------------------------------------------------------------
1457−------------------------------------------------------------------------------
1459−-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
1460−-- container of streams using the specified stream sum operation.
1462−-- @foldWith 'async' $ map return [1..3]@
1465−{-# INLINABLE foldWith #-}
1466−foldWith :: (IsStream t, Foldable f)
1467− => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
1468−foldWith f = foldr f nil
1470−-- | A variant of 'foldMap' that allows you to map a monadic streaming action
1471−-- on a 'Foldable' container and then fold it using the specified stream sum
1474−-- @foldMapWith 'async' return [1..3]@
1477−{-# INLINABLE foldMapWith #-}
1478−foldMapWith :: (IsStream t, Foldable f)
1479− => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1480−foldMapWith f g = foldr (f . g) nil
1482−-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1483−-- monadic streaming function is the last argument.
1486−{-# INLINABLE forEachWith #-}
1487−forEachWith :: (IsStream t, Foldable f)
1488− => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1489−forEachWith f xs g = foldr (f . g) nil xs