@@ -1,1489 +0,0 @@
1−{-# LANGUAGE CPP #-}
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
11−
12−-- |
13−-- Module : Streamly.Streams
14−-- Copyright : (c) 2017 Harendra Kumar
15−--
16−-- License : BSD3
17−-- Maintainer : harendra.kumar@gmail.com
18−-- Stability : experimental
19−-- Portability : GHC
20−--
21−--
22−module Streamly.Streams
23− (
24− IsStream (..)
25− , Streaming -- deprecated
26− , MonadAsync
27−
28− -- * Construction
29− , nil
30− , cons
31− , (.:)
32− , streamBuild
33− , fromCallback
34− , fromSVar
35−
36− -- * Elimination
37− , streamFold
38− , runStream
39− , runStreaming -- deprecated
40− , toSVar
41−
42− -- * Transformation
43− , mkAsync
44− , (|$)
45− , (|&)
46− , (|$.)
47− , (|&.)
48−
49− -- * Merging Streams
50− , serial
51− , wSerial
52− , ahead
53− , async
54− , wAsync
55− , parallel
56− , (<=>) --deprecated
57− , (<|) --deprecated
58−
59− -- * IO Streams
60− , Serial
61− , WSerial
62− , Ahead
63− , Async
64− , WAsync
65− , Parallel
66− , ZipSerial
67− , ZipAsync
68−
69− -- * Stream Transformers
70− , SerialT
71− , StreamT -- deprecated
72− , WSerialT
73− , InterleavedT -- deprecated
74− , AheadT
75− , AsyncT
76− , WAsyncT
77− , ParallelT
78− , ZipStream -- deprecated
79− , ZipSerialM
80− , ZipAsyncM
81−
82− -- * Type Adapters
83− , serially -- deprecated
84− , wSerially
85− , interleaving -- deprecated
86− , aheadly
87− , asyncly
88− , wAsyncly
89− , parallely
90− , zipSerially
91− , zipping -- deprecated
92− , zipAsyncly
93− , zippingAsync -- deprecated
94− , adapt
95−
96− -- * Running Streams
97− , runStreamT -- deprecated
98− , runInterleavedT -- deprecated
99− , runAsyncT -- deprecated
100− , runParallelT -- deprecated
101− , runZipStream -- deprecated
102− , runZipAsync -- deprecated
103−
104− -- * Fold Utilities
105− , foldWith
106− , foldMapWith
107− , forEachWith
108− )
109−where
110−
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)
121−
122−import qualified Streamly.CPS.Stream as C
123−import qualified Streamly.CPS.Concurrent as CC
124−
125−------------------------------------------------------------------------------
126−-- Types that can behave as a Stream
127−------------------------------------------------------------------------------
128−
129−infixr 5 `consM`
130−infixr 5 |:
131−
132−-- | Class of types that can represent a stream of elements of some type 'a' in
133−-- some monad 'm'.
134−--
135−-- @since 0.2.0
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:
141− --
142− -- @
143− -- > toList $ getLine \`consM` getLine \`consM` nil
144− -- hello
145− -- world
146− -- ["hello","world"]
147− -- @
148− --
149− -- /Concurrent (do not use 'parallely' to construct infinite streams)/
150− --
151− -- @since 0.2.0
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 ':'.
155− --
156− -- @
157− -- > toList $ getLine |: getLine |: nil
158− -- hello
159− -- world
160− -- ["hello","world"]
161− -- @
162− --
163− -- @
164− -- let delay = threadDelay 1000000 >> print 1
165− -- runStream $ serially $ delay |: delay |: delay |: nil
166− -- runStream $ parallely $ delay |: delay |: delay |: nil
167− -- @
168− --
169− -- /Concurrent (do not use 'parallely' to construct infinite streams)/
170− --
171− -- @since 0.2.0
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.
175−
176−-- | Same as 'IsStream'.
177−--
178−-- @since 0.1.0
179−{-# DEPRECATED Streaming "Please use IsStream instead." #-}
180−type Streaming = IsStream
181−
182−------------------------------------------------------------------------------
183−-- Constructing a stream
184−------------------------------------------------------------------------------
185−
186−-- | An empty stream.
187−--
188−-- @
189−-- > toList nil
190−-- []
191−-- @
192−--
193−-- @since 0.1.0
194−nil :: IsStream t => t m a
195−nil = fromStream C.nil
196−
197−-- | Constructs a stream by adding a monadic action at the head of an existing
198−-- stream. For example:
199−--
200−-- @
201−-- > toList $ getLine \`consM` getLine \`consM` nil
202−-- hello
203−-- world
204−-- ["hello","world"]
205−-- @
206−--
207−-- @since 0.2.0
208−consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
209−consMSerial m r = fromStream $ C.consM m (toStream r)
210−
211−infixr 5 `cons`
212−
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:
215−--
216−-- @
217−-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
218−-- [1,2,3]
219−-- @
220−--
221−-- @since 0.1.0
222−cons :: IsStream t => a -> t m a -> t m a
223−cons a r = fromStream $ C.cons a (toStream r)
224−
225−infixr 5 .:
226−
227−-- | Operator equivalent of 'cons'.
228−--
229−-- @
230−-- > toList $ 1 .: 2 .: 3 .: nil
231−-- [1,2,3]
232−-- @
233−--
234−-- @since 0.1.1
235−(.:) :: IsStream t => a -> t m a -> t m a
236−(.:) = cons
237−
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)
246− -> (a -> m r)
247− -> m r
248− -> m r)
249− -> t m a
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
253−
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
257−
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
261−
262−------------------------------------------------------------------------------
263−-- Destroying a stream
264−------------------------------------------------------------------------------
265−
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.
269−streamFold
270− :: IsStream t
271− => Maybe (SVar C.Stream m a)
272− -> (a -> t m a -> m r)
273− -> (a -> m r)
274− -> m r
275− -> t m a
276− -> 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
280−
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'@.
284−--
285−-- @since 0.2.0
286−runStream :: Monad m => SerialT m a -> m ()
287−runStream m = go (toStream m)
288− where
289− go m1 =
290− let stop = return ()
291− single _ = return ()
292− yield _ r = go r
293− in (C.runStream m1) Nothing stop single yield
294−
295−-- | Same as 'runStream'
296−--
297−-- @since 0.1.0
298−{-# DEPRECATED runStreaming "Please use runStream instead." #-}
299−runStreaming :: (Monad m, IsStream t) => t m a -> m ()
300−runStreaming = runStream . adapt
301−
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)
306−
307−------------------------------------------------------------------------------
308−-- Transformation
309−------------------------------------------------------------------------------
310−
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'.
317−--
318−-- @since 0.2.0
319−mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
320−mkAsync m = do
321− s <- CC.mkParallel (toStream m)
322− return $ fromStream s
323−
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)
328−
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)
332−
333−infixr 0 |$
334−infixr 0 |$.
335−
336−infixl 1 |&
337−infixl 1 |&.
338−
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
342−-- second delay.
343−--
344−--
345−-- @
346−-- runStream $
347−-- S.mapM (\\x -> threadDelay 1000000 >> print x)
348−-- |$ S.repeatM (threadDelay 1000000 >> return 1)
349−-- @
350−--
351−-- /Concurrent/
352−--
353−-- @since 0.3.0
354−{-# INLINE (|$) #-}
355−(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
356−f |$ x = applyWith f x
357−
358−-- | Parallel reverse function application operator for streams; just like the
359−-- regular reverse function application operator '&' except that it is
360−-- concurrent.
361−--
362−-- @
363−-- runStream $
364−-- S.repeatM (threadDelay 1000000 >> return 1)
365−-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
366−-- @
367−--
368−-- /Concurrent/
369−--
370−-- @since 0.3.0
371−{-# INLINE (|&) #-}
372−(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
373−x |& f = f |$ x
374−
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.
380−--
381−-- @
382−-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
383−-- |$. S.repeatM (threadDelay 1000000 >> return 1)
384−-- @
385−--
386−-- /Concurrent/
387−--
388−-- @since 0.3.0
389−{-# INLINE (|$.) #-}
390−(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
391−f |$. x = runWith f x
392−
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.
395−--
396−-- @
397−-- S.repeatM (threadDelay 1000000 >> return 1)
398−-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
399−-- @
400−--
401−-- /Concurrent/
402−--
403−-- @since 0.3.0
404−{-# INLINE (|&.) #-}
405−(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
406−x |&. f = f |$. x
407−
408−------------------------------------------------------------------------------
409−-- CPP macros for common instances
410−------------------------------------------------------------------------------
411−
412−-- XXX use template haskell instead and include Monoid and IsStream instances
413−-- as well.
414−
415−withLocal :: MonadReader r m => (r -> r) -> C.Stream m a -> C.Stream m a
416−withLocal f m =
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
421−
422−{-
423−-- XXX handle and test cross thread state transfer
424−withCatchError
425− :: MonadError e m
426− => C.Stream m a -> (e -> C.Stream m a) -> C.Stream m a
427−withCatchError m h =
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)
432− in handle $ run m
433−-}
434−
435−#define MONADPARALLEL , MonadAsync m
436−
437−#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
438−instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
439− pure = STREAM . C.singleton; \
440− (<*>) = ap }
441−
442−#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
443−instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
444− liftBase = liftBaseDefault }; \
445− \
446−instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
447− liftIO = lift . liftIO }; \
448− \
449−instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
450− throwM = lift . throwM }; \
451− \
452−{- \
453−instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
454− throwError = lift . throwError; \
455− catchError m h = \
456− fromStream $ withCatchError (toStream m) (\e -> toStream $ h e) }; \
457−-} \
458− \
459−instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
460− ask = lift ask; \
461− local f m = fromStream $ withLocal f (toStream m) }; \
462− \
463−instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
464− get = lift get; \
465− put x = lift (put x); \
466− state k = lift (state k) }
467−
468−------------------------------------------------------------------------------
469−-- SerialT
470−------------------------------------------------------------------------------
471−
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.
476−--
477−-- @
478−-- import Streamly
479−-- import qualified "Streamly.Prelude" as S
480−--
481−-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
482−-- @
483−-- @
484−-- [1,2,3,4]
485−-- @
486−--
487−-- The 'Monad' instance runs the /monadic continuation/ for each
488−-- element of the stream, serially.
489−--
490−-- @
491−-- main = 'runStream' . 'serially' $ do
492−-- x <- return 1 \<\> return 2
493−-- S.once $ print x
494−-- @
495−-- @
496−-- 1
497−-- 2
498−-- @
499−--
500−-- 'SerialT' nests streams serially in a depth first manner.
501−--
502−-- @
503−-- main = 'runStream' . 'serially' $ do
504−-- x <- return 1 \<\> return 2
505−-- y <- return 3 \<\> return 4
506−-- S.once $ print (x, y)
507−-- @
508−-- @
509−-- (1,3)
510−-- (1,4)
511−-- (2,3)
512−-- (2,4)
513−-- @
514−--
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.
520−--
521−-- The 'serially' combinator can be omitted as the default stream type is
522−-- 'SerialT'.
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
525−-- time.
526−--
527−-- @since 0.2.0
528−newtype SerialT m a = SerialT {getSerialT :: C.Stream m a}
529− deriving (Semigroup, Monoid, Functor, MonadTrans)
530−
531−-- |
532−-- @since 0.1.0
533−{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
534−type StreamT = SerialT
535−
536−instance IsStream SerialT where
537− toStream = getSerialT
538− fromStream = SerialT
539−
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
543− consM = consMSerial
544−
545− {-# INLINE (|:) #-}
546− {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
547− (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
548− (|:) = consMSerial
549−
550−------------------------------------------------------------------------------
551−-- Semigroup
552−------------------------------------------------------------------------------
553−
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.
557−--
558−-- @since 0.2.0
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)
562−
563−------------------------------------------------------------------------------
564−-- Monad
565−------------------------------------------------------------------------------
566−
567−instance Monad m => Monad (SerialT m) where
568− return = pure
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
574−
575−------------------------------------------------------------------------------
576−-- Other instances
577−------------------------------------------------------------------------------
578−
579−MONAD_APPLICATIVE_INSTANCE(SerialT,)
580−MONAD_COMMON_INSTANCES(SerialT,)
581−
582−------------------------------------------------------------------------------
583−-- WSerialT
584−------------------------------------------------------------------------------
585−
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.
590−--
591−-- @
592−-- import Streamly
593−-- import qualified "Streamly.Prelude" as S
594−--
595−-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
596−-- @
597−-- @
598−-- [1,3,2,4]
599−-- @
600−--
601−-- Similarly, the 'Monad' instance interleaves the iterations of the
602−-- inner and the outer loop, nesting loops in a breadth first manner.
603−--
604−--
605−-- @
606−-- main = 'runStream' . 'wSerially' $ do
607−-- x <- return 1 \<\> return 2
608−-- y <- return 3 \<\> return 4
609−-- S.once $ print (x, y)
610−-- @
611−-- @
612−-- (1,3)
613−-- (2,3)
614−-- (1,4)
615−-- (2,4)
616−-- @
617−--
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
620−-- stream.
621−--
622−-- @since 0.2.0
623−newtype WSerialT m a = WSerialT {getWSerialT :: C.Stream m a}
624− deriving (Functor, MonadTrans)
625−
626−-- |
627−-- @since 0.1.0
628−{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
629−type InterleavedT = WSerialT
630−
631−instance IsStream WSerialT where
632− toStream = getWSerialT
633− fromStream = WSerialT
634−
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
638− consM = consMSerial
639−
640− {-# INLINE (|:) #-}
641− {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
642− (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
643− (|:) = consMSerial
644−
645−------------------------------------------------------------------------------
646−-- Semigroup
647−------------------------------------------------------------------------------
648−
649−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
650−-- Interleaves two streams, yielding one element from each stream alternately.
651−--
652−-- @since 0.2.0
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)
656−
657−instance Semigroup (WSerialT m a) where
658− (<>) = wSerial
659−
660−infixr 5 <=>
661−
662−-- | Same as 'wSerial'.
663−--
664−-- @since 0.1.0
665−{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
666−{-# INLINE (<=>) #-}
667−(<=>) :: IsStream t => t m a -> t m a -> t m a
668−(<=>) = wSerial
669−
670−------------------------------------------------------------------------------
671−-- Monoid
672−------------------------------------------------------------------------------
673−
674−instance Monoid (WSerialT m a) where
675− mempty = nil
676− mappend = (<>)
677−
678−------------------------------------------------------------------------------
679−-- Monad
680−------------------------------------------------------------------------------
681−
682−instance Monad m => Monad (WSerialT m) where
683− return = pure
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
689−
690−------------------------------------------------------------------------------
691−-- Other instances
692−------------------------------------------------------------------------------
693−
694−MONAD_APPLICATIVE_INSTANCE(WSerialT,)
695−MONAD_COMMON_INSTANCES(WSerialT,)
696−
697−------------------------------------------------------------------------------
698−-- AheadT
699−------------------------------------------------------------------------------
700−
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.
706−--
707−-- @
708−-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
709−-- @
710−-- @
711−-- [1,2,3,4]
712−-- @
713−--
714−-- Any exceptions generated by a constituent stream are propagated to the
715−-- output stream.
716−--
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
719−-- 'SerialT'.
720−--
721−-- @
722−-- import "Streamly"
723−-- import qualified "Streamly.Prelude" as S
724−-- import Control.Concurrent
725−--
726−-- main = 'runStream' . 'aheadly' $ do
727−-- n <- return 3 \<\> return 2 \<\> return 1
728−-- S.once $ do
729−-- threadDelay (n * 1000000)
730−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
731−-- @
732−-- @
733−-- ThreadId 40: Delay 1
734−-- ThreadId 39: Delay 2
735−-- ThreadId 38: Delay 3
736−-- @
737−--
738−-- All iterations may run in the same thread if they do not block.
739−--
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.
743−--
744−-- @since 0.3.0
745−newtype AheadT m a = AheadT {getAheadT :: C.Stream m a}
746− deriving (Functor, MonadTrans)
747−
748−instance IsStream AheadT where
749− toStream = getAheadT
750− fromStream = AheadT
751−
752− {-# INLINE consM #-}
753− {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
754− consM m r = fromStream $ CC.consMAhead m (toStream r)
755−
756− {-# INLINE (|:) #-}
757− {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
758− (|:) = consM
759−
760−------------------------------------------------------------------------------
761−-- Semigroup
762−------------------------------------------------------------------------------
763−
764−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
765−-- Merges two streams sequentially but with concurrent lookahead.
766−--
767−-- @since 0.3.0
768−{-# INLINE ahead #-}
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)
771−
772−instance MonadAsync m => Semigroup (AheadT m a) where
773− (<>) = ahead
774−
775−------------------------------------------------------------------------------
776−-- Monoid
777−------------------------------------------------------------------------------
778−
779−instance MonadAsync m => Monoid (AheadT m a) where
780− mempty = nil
781− mappend = (<>)
782−
783−------------------------------------------------------------------------------
784−-- Monad
785−------------------------------------------------------------------------------
786−
787−{-# INLINE aheadbind #-}
788−aheadbind
789− :: MonadAsync m
790− => C.Stream m a
791− -> (a -> C.Stream m b)
792− -> C.Stream m b
793−aheadbind m f = go m
794− where
795− go (C.Stream g) =
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
801−
802−instance MonadAsync m => Monad (AheadT m) where
803− return = pure
804− (AheadT m) >>= f = AheadT $ aheadbind m (getAheadT . f)
805−
806−------------------------------------------------------------------------------
807−-- Other instances
808−------------------------------------------------------------------------------
809−
810−MONAD_APPLICATIVE_INSTANCE(AheadT,MONADPARALLEL)
811−MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
812−
813−------------------------------------------------------------------------------
814−-- AsyncT
815−------------------------------------------------------------------------------
816−
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.
822−--
823−-- @
824−-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
825−-- @
826−-- @
827−-- [1,2,3,4]
828−-- @
829−--
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.
836−--
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
840−-- consumer.
841−--
842−-- @
843−-- import "Streamly"
844−-- import qualified "Streamly.Prelude" as S
845−-- import Control.Concurrent
846−--
847−-- main = 'runStream' . 'asyncly' $ do
848−-- n <- return 3 \<\> return 2 \<\> return 1
849−-- S.once $ do
850−-- threadDelay (n * 1000000)
851−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
852−-- @
853−-- @
854−-- ThreadId 40: Delay 1
855−-- ThreadId 39: Delay 2
856−-- ThreadId 38: Delay 3
857−-- @
858−--
859−-- All iterations may run in the same thread if they do not block.
860−--
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.
864−--
865−-- @since 0.1.0
866−newtype AsyncT m a = AsyncT {getAsyncT :: C.Stream m a}
867− deriving (Functor, MonadTrans)
868−
869−instance IsStream AsyncT where
870− toStream = getAsyncT
871− fromStream = AsyncT
872−
873− {-# INLINE consM #-}
874− {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
875− consM m r = fromStream $ CC.consMAsync m (toStream r)
876−
877− {-# INLINE (|:) #-}
878− {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
879− (|:) = consM
880−
881−------------------------------------------------------------------------------
882−-- Semigroup
883−------------------------------------------------------------------------------
884−
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.
888−--
889−-- @since 0.2.0
890−{-# INLINE async #-}
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)
893−
894−-- | Same as 'async'.
895−--
896−-- @since 0.1.0
897−{-# DEPRECATED (<|) "Please use 'async' instead." #-}
898−{-# INLINE (<|) #-}
899−(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
900−(<|) = async
901−
902−instance MonadAsync m => Semigroup (AsyncT m a) where
903− (<>) = async
904−
905−------------------------------------------------------------------------------
906−-- Monoid
907−------------------------------------------------------------------------------
908−
909−instance MonadAsync m => Monoid (AsyncT m a) where
910− mempty = nil
911− mappend = (<>)
912−
913−------------------------------------------------------------------------------
914−-- Monad
915−------------------------------------------------------------------------------
916−
917−{-# INLINE parbind #-}
918−parbind
919− :: (forall c. C.Stream m c -> C.Stream m c -> C.Stream m c)
920− -> C.Stream m a
921− -> (a -> C.Stream m b)
922− -> C.Stream m b
923−parbind par m f = go m
924− where
925− go (C.Stream g) =
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
931−
932−instance MonadAsync m => Monad (AsyncT m) where
933− return = pure
934− (AsyncT m) >>= f = AsyncT $ parbind CC.async m (getAsyncT . f)
935−
936−------------------------------------------------------------------------------
937−-- Other instances
938−------------------------------------------------------------------------------
939−
940−MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
941−MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
942−
943−------------------------------------------------------------------------------
944−-- WAsyncT
945−------------------------------------------------------------------------------
946−
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.
951−--
952−-- @
953−-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
954−-- @
955−-- @
956−-- [1,3,2,4]
957−-- @
958−--
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.
965−--
966−-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
967−-- concurrently using a round robin scheduling.
968−--
969−-- @
970−-- import "Streamly"
971−-- import qualified "Streamly.Prelude" as S
972−-- import Control.Concurrent
973−--
974−-- main = 'runStream' . 'wAsyncly' $ do
975−-- n <- return 3 \<\> return 2 \<\> return 1
976−-- S.once $ do
977−-- threadDelay (n * 1000000)
978−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
979−-- @
980−-- @
981−-- ThreadId 40: Delay 1
982−-- ThreadId 39: Delay 2
983−-- ThreadId 38: Delay 3
984−-- @
985−--
986−-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
987−-- concurrently, unconditionally.
988−--
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
991−-- stream.
992−--
993−-- @since 0.2.0
994−newtype WAsyncT m a = WAsyncT {getWAsyncT :: C.Stream m a}
995− deriving (Functor, MonadTrans)
996−
997−instance IsStream WAsyncT where
998− toStream = getWAsyncT
999− fromStream = WAsyncT
1000−
1001− {-# INLINE consM #-}
1002− {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1003− consM m r = fromStream $ CC.consMWAsync m (toStream r)
1004−
1005− {-# INLINE (|:) #-}
1006− {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1007− (|:) = consM
1008−
1009−------------------------------------------------------------------------------
1010−-- Semigroup
1011−------------------------------------------------------------------------------
1012−
1013−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
1014−-- Merges two streams concurrently choosing elements from both fairly.
1015−--
1016−-- @since 0.2.0
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)
1020−
1021−instance MonadAsync m => Semigroup (WAsyncT m a) where
1022− (<>) = wAsync
1023−
1024−------------------------------------------------------------------------------
1025−-- Monoid
1026−------------------------------------------------------------------------------
1027−
1028−instance MonadAsync m => Monoid (WAsyncT m a) where
1029− mempty = nil
1030− mappend = (<>)
1031−
1032−------------------------------------------------------------------------------
1033−-- Monad
1034−------------------------------------------------------------------------------
1035−
1036−instance MonadAsync m => Monad (WAsyncT m) where
1037− return = pure
1038− (WAsyncT m) >>= f =
1039− WAsyncT $ parbind CC.wAsync m (getWAsyncT . f)
1040−
1041−------------------------------------------------------------------------------
1042−-- Other instances
1043−------------------------------------------------------------------------------
1044−
1045−MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
1046−MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
1047−
1048−------------------------------------------------------------------------------
1049−-- ParallelT
1050−------------------------------------------------------------------------------
1051−
1052−-- | Async composition with simultaneous traversal of all streams.
1053−--
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.
1059−--
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.
1069−--
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
1081−-- is full.
1082−--
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.
1095−--
1096−-- @
1097−-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
1098−-- @
1099−-- @
1100−-- [1,3,2,4]
1101−-- @
1102−--
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.
1105−--
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.
1112−--
1113−-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
1114−-- of the loop concurrently.
1115−--
1116−-- @
1117−-- import "Streamly"
1118−-- import qualified "Streamly.Prelude" as S
1119−-- import Control.Concurrent
1120−--
1121−-- main = 'runStream' . 'parallely' $ do
1122−-- n <- return 3 \<\> return 2 \<\> return 1
1123−-- S.once $ do
1124−-- threadDelay (n * 1000000)
1125−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1126−-- @
1127−-- @
1128−-- ThreadId 40: Delay 1
1129−-- ThreadId 39: Delay 2
1130−-- ThreadId 38: Delay 3
1131−-- @
1132−--
1133−-- Note that parallel composition can only combine a finite number of
1134−-- streams as it needs to retain state for each unfinished stream.
1135−--
1136−-- @since 0.1.0
1137−newtype ParallelT m a = ParallelT {getParallelT :: C.Stream m a}
1138− deriving (Functor, MonadTrans)
1139−
1140−instance IsStream ParallelT where
1141− toStream = getParallelT
1142− fromStream = ParallelT
1143−
1144− {-# INLINE consM #-}
1145− {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1146− consM m r = fromStream $ CC.consMParallel m (toStream r)
1147−
1148− {-# INLINE (|:) #-}
1149− {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1150− (|:) = consM
1151−
1152−------------------------------------------------------------------------------
1153−-- Semigroup
1154−------------------------------------------------------------------------------
1155−
1156−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
1157−-- Merges two streams concurrently.
1158−--
1159−-- @since 0.2.0
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)
1163−
1164−instance MonadAsync m => Semigroup (ParallelT m a) where
1165− (<>) = parallel
1166−
1167−------------------------------------------------------------------------------
1168−-- Monoid
1169−------------------------------------------------------------------------------
1170−
1171−instance MonadAsync m => Monoid (ParallelT m a) where
1172− mempty = nil
1173− mappend = (<>)
1174−
1175−------------------------------------------------------------------------------
1176−-- Monad
1177−------------------------------------------------------------------------------
1178−
1179−instance MonadAsync m => Monad (ParallelT m) where
1180− return = pure
1181− (ParallelT m) >>= f = ParallelT $ parbind CC.parallel m (getParallelT . f)
1182−
1183−------------------------------------------------------------------------------
1184−-- Other instances
1185−------------------------------------------------------------------------------
1186−
1187−MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
1188−MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
1189−
1190−------------------------------------------------------------------------------
1191−-- Serially Zipping Streams
1192−------------------------------------------------------------------------------
1193−
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
1196−-- those elements.
1197−--
1198−-- @
1199−-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1200−-- where s1 = fromFoldable [1, 2]
1201−-- s2 = fromFoldable [3, 4]
1202−-- s3 = fromFoldable [5, 6]
1203−-- @
1204−-- @
1205−-- [(1,3,5),(2,4,6)]
1206−-- @
1207−--
1208−-- The 'Semigroup' instance of this type works the same way as that of
1209−-- 'SerialT'.
1210−--
1211−-- @since 0.2.0
1212−newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: C.Stream m a}
1213− deriving (Functor, Semigroup, Monoid)
1214−
1215−-- |
1216−-- @since 0.1.0
1217−{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
1218−type ZipStream = ZipSerialM
1219−
1220−instance IsStream ZipSerialM where
1221− toStream = getZipSerialM
1222− fromStream = ZipSerialM
1223−
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
1228−
1229− {-# INLINE (|:) #-}
1230− {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1231− (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1232− (|:) = consMSerial
1233−
1234−instance Monad m => Applicative (ZipSerialM m) where
1235− pure = ZipSerialM . C.repeat
1236− m1 <*> m2 = fromStream $ C.zipWith id (toStream m1) (toStream m2)
1237−
1238−------------------------------------------------------------------------------
1239−-- Parallely Zipping Streams
1240−------------------------------------------------------------------------------
1241−
1242−-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
1243−-- be zipped concurrently.
1244−--
1245−-- @
1246−-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1247−-- where s1 = fromFoldable [1, 2]
1248−-- s2 = fromFoldable [3, 4]
1249−-- s3 = fromFoldable [5, 6]
1250−-- @
1251−-- @
1252−-- [(1,3,5),(2,4,6)]
1253−-- @
1254−--
1255−-- The 'Semigroup' instance of this type works the same way as that of
1256−-- 'SerialT'.
1257−--
1258−-- @since 0.2.0
1259−newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: C.Stream m a}
1260− deriving (Functor, Semigroup, Monoid)
1261−
1262−instance IsStream ZipAsyncM where
1263− toStream = getZipAsyncM
1264− fromStream = ZipAsyncM
1265−
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
1270−
1271− {-# INLINE (|:) #-}
1272− {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1273− (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1274− (|:) = consMSerial
1275−
1276−instance MonadAsync m => Applicative (ZipAsyncM m) where
1277− pure = ZipAsyncM . C.repeat
1278− m1 <*> m2 = fromStream $ CC.zipAsyncWith id (toStream m1) (toStream m2)
1279−
1280−-------------------------------------------------------------------------------
1281−-- Type adapting combinators
1282−-------------------------------------------------------------------------------
1283−
1284−-- | Adapt any specific stream type to any other specific stream type.
1285−--
1286−-- @since 0.1.0
1287−adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
1288−adapt = fromStream . toStream
1289−
1290−-- | Fix the type of a polymorphic stream as 'SerialT'.
1291−--
1292−-- @since 0.1.0
1293−serially :: IsStream t => SerialT m a -> t m a
1294−serially = adapt
1295−
1296−-- | Fix the type of a polymorphic stream as 'WSerialT'.
1297−--
1298−-- @since 0.2.0
1299−wSerially :: IsStream t => WSerialT m a -> t m a
1300−wSerially = adapt
1301−
1302−-- | Same as 'wSerially'.
1303−--
1304−-- @since 0.1.0
1305−{-# DEPRECATED interleaving "Please use wSerially instead." #-}
1306−interleaving :: IsStream t => WSerialT m a -> t m a
1307−interleaving = wSerially
1308−
1309−-- | Fix the type of a polymorphic stream as 'AheadT'.
1310−--
1311−-- @since 0.3.0
1312−aheadly :: IsStream t => AheadT m a -> t m a
1313−aheadly = adapt
1314−
1315−-- | Fix the type of a polymorphic stream as 'AsyncT'.
1316−--
1317−-- @since 0.1.0
1318−asyncly :: IsStream t => AsyncT m a -> t m a
1319−asyncly = adapt
1320−
1321−-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1322−--
1323−-- @since 0.2.0
1324−wAsyncly :: IsStream t => WAsyncT m a -> t m a
1325−wAsyncly = adapt
1326−
1327−-- | Fix the type of a polymorphic stream as 'ParallelT'.
1328−--
1329−-- @since 0.1.0
1330−parallely :: IsStream t => ParallelT m a -> t m a
1331−parallely = adapt
1332−
1333−-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
1334−--
1335−-- @since 0.2.0
1336−zipSerially :: IsStream t => ZipSerialM m a -> t m a
1337−zipSerially = adapt
1338−
1339−-- | Same as 'zipSerially'.
1340−--
1341−-- @since 0.1.0
1342−{-# DEPRECATED zipping "Please use zipSerially instead." #-}
1343−zipping :: IsStream t => ZipSerialM m a -> t m a
1344−zipping = zipSerially
1345−
1346−-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
1347−--
1348−-- @since 0.2.0
1349−zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
1350−zipAsyncly = adapt
1351−
1352−-- | Same as 'zipAsyncly'.
1353−--
1354−-- @since 0.1.0
1355−{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
1356−zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
1357−zippingAsync = zipAsyncly
1358−
1359−-------------------------------------------------------------------------------
1360−-- Running Streams, convenience functions specialized to types
1361−-------------------------------------------------------------------------------
1362−
1363−-- | Same as @runStream@.
1364−--
1365−-- @since 0.1.0
1366−{-# DEPRECATED runStreamT "Please use runStream instead." #-}
1367−runStreamT :: Monad m => SerialT m a -> m ()
1368−runStreamT = runStream
1369−
1370−-- | Same as @runStream . wSerially@.
1371−--
1372−-- @since 0.1.0
1373−{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
1374−runInterleavedT :: Monad m => InterleavedT m a -> m ()
1375−runInterleavedT = runStream . wSerially
1376−
1377−-- | Same as @runStream . asyncly@.
1378−--
1379−-- @since 0.1.0
1380−{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}
1381−runAsyncT :: Monad m => AsyncT m a -> m ()
1382−runAsyncT = runStream . asyncly
1383−
1384−-- | Same as @runStream . parallely@.
1385−--
1386−-- @since 0.1.0
1387−{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
1388−runParallelT :: Monad m => ParallelT m a -> m ()
1389−runParallelT = runStream . parallely
1390−
1391−-- | Same as @runStream . zipping@.
1392−--
1393−-- @since 0.1.0
1394−{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}
1395−runZipStream :: Monad m => ZipSerialM m a -> m ()
1396−runZipStream = runStream . zipSerially
1397−
1398−-- | Same as @runStream . zippingAsync@.
1399−--
1400−-- @since 0.1.0
1401−{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}
1402−runZipAsync :: Monad m => ZipAsyncM m a -> m ()
1403−runZipAsync = runStream . zipAsyncly
1404−
1405−------------------------------------------------------------------------------
1406−-- IO Streams
1407−------------------------------------------------------------------------------
1408−
1409−-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
1410−-- for more details.
1411−--
1412−-- @since 0.2.0
1413−type Serial a = SerialT IO a
1414−
1415−-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
1416−-- documentation for more details.
1417−--
1418−-- @since 0.2.0
1419−type WSerial a = WSerialT IO a
1420−
1421−-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
1422−-- 'AheadT' documentation for more details.
1423−--
1424−-- @since 0.3.0
1425−type Ahead a = AheadT IO a
1426−
1427−-- | A demand driven left biased parallely composing IO stream of elements of
1428−-- type @a@. See 'AsyncT' documentation for more details.
1429−--
1430−-- @since 0.2.0
1431−type Async a = AsyncT IO a
1432−
1433−-- | A round robin parallely composing IO stream of elements of type @a@.
1434−-- See 'WAsyncT' documentation for more details.
1435−--
1436−-- @since 0.2.0
1437−type WAsync a = WAsyncT IO a
1438−
1439−-- | A parallely composing IO stream of elements of type @a@.
1440−-- See 'ParallelT' documentation for more details.
1441−--
1442−-- @since 0.2.0
1443−type Parallel a = ParallelT IO a
1444−
1445−-- | An IO stream whose applicative instance zips streams serially.
1446−--
1447−-- @since 0.2.0
1448−type ZipSerial a = ZipSerialM IO a
1449−
1450−-- | An IO stream whose applicative instance zips streams wAsyncly.
1451−--
1452−-- @since 0.2.0
1453−type ZipAsync a = ZipAsyncM IO a
1454−
1455−------------------------------------------------------------------------------
1456−-- Fold Utilities
1457−------------------------------------------------------------------------------
1458−
1459−-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
1460−-- container of streams using the specified stream sum operation.
1461−--
1462−-- @foldWith 'async' $ map return [1..3]@
1463−--
1464−-- @since 0.1.0
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
1469−
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
1472−-- operation.
1473−--
1474−-- @foldMapWith 'async' return [1..3]@
1475−--
1476−-- @since 0.1.0
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
1481−
1482−-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1483−-- monadic streaming function is the last argument.
1484−--
1485−-- @since 0.1.0
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