git.iko.soy/streamly
on master help sign in
Update doc for Streamly module
754b95bHarendra Kumar
src/Streamly.hs+137−56
compare with master
unifiedsplit
@@ -6,64 +6,40 @@
6-- Maintainer : harendra.kumar@gmail.com
6-- Maintainer : harendra.kumar@gmail.com
7-- Stability : experimental
7-- Stability : experimental
8-- Portability : GHC
8-- Portability : GHC
9−--
10−-- 'AsyncT' is a monad transformer that extends the product style composition
11−-- of monads to streams; it is a functional porgramming equivalent to nested
12−-- loops in imperative programming. This style of composition allows composing
13−-- each element in one stream with each element in the other stream,
14−-- generalizing the monadic product of single elements. For example, you can
15−-- think of the IO Monad as a special case of the more general @AsyncT IO@
16−-- monad; with single element streams. List transformers and logic programming
17−-- monads also provide a similar product style composition of streams, however
18−-- 'AsyncT' generalizes it with the time dimension; allowing streams to be
19−-- composed in an asynchronous and concurrent fashion in many different ways.
20−--
21−-- The power of this seemingly simple addition of asynchronicity and
22−-- concurrency to product style streaming composition should not be
23−-- underestimated. It unifies a number of disparate abstractions into one
24−-- powerful and elegant abstraction. A wide variety of programming problems
25−-- can be solved elegantly with this abstraction. In particular, it unifies
26−-- three major programming domains namely non-deterministic (logic)
27−-- programming, concurrent programming and functional reactive programming. In
28−-- other words, you can do everything with this one abstraction that you could
29−-- with list transformers (e.g.
30−-- <https://hackage.haskell.org/package/list-t list-t>), logic programming
31−-- monads (e.g. <https://hackage.haskell.org/package/logict logict>),
32−-- streaming libraries (a lot of what
33−-- <https://hackage.haskell.org/package/conduit conduit> or
34−-- <https://hackage.haskell.org/package/pipes pipes> can do), concurrency
35−-- libraries (e.g. <https://hackage.haskell.org/package/async async>) and FRP
36−-- libraries (e.g. <https://hackage.haskell.org/package/Yampa Yampa> or
37−-- <https://hackage.haskell.org/package/reflex reflex>).
38−--
39−-- When it comes to streaming composition, if @conduit@ or @pipes@ are arrows
40−-- then streamly is monad. Streaming libraries like conduit or pipes provide
41−-- a categorical composition (sum) of streams whereas streamly provides a more
42−-- general monadic composition (product) of streams, with concurrency.
43−-- Streamly interworks with major streaming libraries providing a way to use
44−-- both types of compositions and using the right tool at the right place.
45−-- When it comes to concurrency, streamly may be comared with imperative style
46−-- concurrency frameworks like
47−-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> or
48−-- <https://en.wikipedia.org/wiki/Cilk Cilk>.
49−--
50−-- For more details please see the 'Streamly.Tutorial' and 'Streamly.Examples'.
51
9
52−-- A simple inline example here illustrating applicative, monad and alternative
53−-- compositions.
54module Streamly
10module Streamly
11+ (
12+ -- * Background
13+ -- $background
14+
15+ -- * Overview
16+ -- $overview
17+
55− ( MonadAsync
18+ MonadAsync
56 , Streaming
19 , Streaming
57
20
58 -- * Product Style Composition
21 -- * Product Style Composition
22+ -- $product
59 , StreamT
23 , StreamT
60 , InterleavedT
24 , InterleavedT
61 , AsyncT
25 , AsyncT
62 , ParallelT
26 , ParallelT
27+
28+ -- * Zip Style Composition
29+ -- $zipping
63 , ZipStream
30 , ZipStream
64 , ZipAsync
31 , ZipAsync
65
32
33+ -- * Sum Style Composition
34+ -- $sum
35+ , (<=>)
36+ , (<|)
37+
38+ -- * Transformation
39+ , async
40+
66 -- * Stream Type Adapters
41 -- * Stream Type Adapters
42+ -- $adapters
67 , serially
43 , serially
68 , interleaving
44 , interleaving
69 , asyncly
45 , asyncly
@@ -81,15 +57,8 @@ module Streamly
81 , runZipStream
57 , runZipStream
82 , runZipAsync
58 , runZipAsync
83
59
84− -- * Transformation
85− , async
86−
87− -- * Sum Style Composition
88− -- $monoidal
89− , (<=>)
90− , (<|)
91−
92 -- * Fold Utilities
60 -- * Fold Utilities
61+ -- $foldutils
93 , foldWith
62 , foldWith
94 , foldMapWith
63 , foldMapWith
95 , forEachWith
64 , forEachWith
@@ -111,7 +80,119 @@ import Control.Monad (MonadPlus(..))
111import Control.Monad.IO.Class (MonadIO (..))
80import Control.Monad.IO.Class (MonadIO (..))
112import Control.Monad.Trans.Class (MonadTrans (..))
81import Control.Monad.Trans.Class (MonadTrans (..))
113
82
114−-- $monoidal
83+-- $background
84+--
85+-- Streamly provides a monad transformer that extends the product style
86+-- composition of monads to streams of many elements of the same type; it is a
87+-- functional programming equivalent of nested loops from imperative
88+-- programming. Composing each element in one stream with each element in the
89+-- other stream generalizes the monadic product of single elements. You can
90+-- think of the IO monad as a special case of the more general @StreamT IO@
91+-- monad; with single element streams. List transformers and logic programming
92+-- monads also provide a similar product style composition of streams, however
93+-- streamly generalizes it with the time dimension; allowing streams to be
94+-- composed in an asynchronous and concurrent fashion in many different ways.
95+-- It also provides multiple alternative ways of composing streams e.g.
96+-- serial, interleaved or concurrent.
97+--
98+-- The seemingly simple addition of asynchronicity and concurrency to product
99+-- style streaming composition unifies a number of disparate abstractions into
100+-- one powerful and elegant abstraction. A wide variety of programming
101+-- problems can be solved elegantly with this abstraction. In particular, it
102+-- unifies three major programming domains namely non-deterministic (logic)
103+-- programming, concurrent programming and functional reactive programming. In
104+-- other words, you can do everything with this one abstraction that you could
105+-- with list transformers (e.g.
106+-- <https://hackage.haskell.org/package/list-t list-t>), logic programming
107+-- monads (e.g. <https://hackage.haskell.org/package/logict logict>),
108+-- streaming libraries (a lot of what
109+-- <https://hackage.haskell.org/package/conduit conduit> or
110+-- <https://hackage.haskell.org/package/pipes pipes> can do), concurrency
111+-- libraries (e.g. <https://hackage.haskell.org/package/async async>) and FRP
112+-- libraries (e.g. <https://hackage.haskell.org/package/Yampa Yampa> or
113+-- <https://hackage.haskell.org/package/reflex reflex>).
114+
115+-- $overview
116+--
117+-- Streamly provides six distinct stream types i.e. 'StreamT', 'InterleavedT',
118+-- 'AsyncT' and 'ParallelT', 'ZipStream' and 'ZipAsync', each representing a
119+-- stream of elements. All these types have the same underlying representation
120+-- and can be adapted from one to another using type adaptor combinators
121+-- described later. Each of these types belongs to the 'Streaming' type class
122+-- which helps converting the specific type to and from the underlying generic
123+-- stream type.
124+--
125+-- The types 'StreamT', 'InterleavedT', 'AsyncT' and 'ParallelT' are 'Monad'
126+-- transformers with the monadic bind operation combining streams in a product
127+-- style in much the same way as a list monad or a list transformer i.e. each
128+-- element from one stream is combined with every element of the other stream.
129+-- However, the applicative and monadic composition of these types differ in
130+-- terms of the ordering and time sequence in which the elements from two
131+-- streams are combined. 'StreamT' and 'InterleavedT' compose streams serially
132+-- whereas 'AsyncT' and 'ParallelT' are their concurrent counterparts. See the
133+-- documentation of the respective types for more details.
134+--
135+-- The types 'ZipStream' and 'ZipAsync' provide 'Applicative' instances to zip
136+-- two streams together i.e. each element in one stream is combined with the
137+-- corresponding element in the other stream. 'ZipStream' generates the streams
138+-- being zipped serially whereas 'ZipAsync' produces both the elements being
139+-- zipped concurrently.
140+--
141+-- Two streams of the same type can be combined using a sum style composition
142+-- to generate a stream of the same type where the output stream would contain
143+-- all elements of both the streams. However, the sequence in which the
144+-- elements in the resulting stream are produced depends on the combining
145+-- operator. Four distinct sum style operators, '<>', '<=>', '<|' and '<|>'
146+-- combine two streams in different ways, each corresponding to the one of the
147+-- four ways of combining monadically. See the respective section below for
148+-- more details.
149+--
150+-- Concurrent composition types 'AsyncT', 'ParallelT', 'ZipAsync' and
151+-- concurrent composition operators '<|' and '<|>' require the underlying monad
152+-- of the streaming monad transformer to be 'MonadAsync'.
153+--
154+-- For more details please see the 'Streamly.Tutorial' and 'Streamly.Examples'.
155+
156+-- A simple inline example here illustrating applicative, monad and alternative
157+-- compositions.
158+
159+-- $product
160+--
161+-- Streams that compose serially or non-concurrently come in two flavors i.e.
162+-- 'StreamT' and 'InterleavedT'. Both of these serial flavors have
163+-- corresponding concurrent equivalents, those are 'AsyncT' and 'ParallelT'
164+-- respectively.
165+
166+-- $zipping
167+--
168+-- 'ZipStream' and 'ZipAsync', provide 'Applicative' instances for zipping the
169+-- corresponding elements of two streams together. Note that these types do not
170+-- provide 'Monad' instances.
171+
172+-- $sum
173+--
174+-- Just like product style composition there are four distinct ways to combine
175+-- streams in sum style each directly corresponding to one of the product style
176+-- composition. The standard semigroup append '<>' operator appends two
177+-- streams serially, this style corresponds to the 'StreamT' style of monadic
178+-- composition. The standard 'Alternative' operator '<|>' fairly interleaves
179+-- two streams in parallel, this corresponds to the 'ParallelT' style. Two
180+-- additional sum style composition operators that streamly introduces are
181+-- described below.
182+
183+-- $adapters
184+--
185+-- Code using streamly is written such that it is agnostic of any specific
186+-- streaming type. We use a polymorphic type with the 'Streaming' class
187+-- constraint. Finally when running the monad we can specify the actual type
188+-- that we want to use to interpret the code. However, in certain cases we may
189+-- want to use a specific type to force a certain type of composition. Since
190+-- they have the same underlying type, streams can be converted from one type
191+-- to another freely. These combinators can be used to convert the streaming
192+-- types at will. If you get an ambiguous type error you perhaps forgot to
193+-- use a sepcific type before running the monad.
194+
195+-- $foldutils
115--
196--
116−-- These combinators can be used in place of 'Monoid' ('<>') or 'Alternative'
197+-- These are some convenience functions to fold any 'Foldable' container using
117−-- ('<|>') composition to fold streams in alternate ways.
198+-- one of the sum composition operators to convert it into a streamly stream.
src/Streamly/Core.hs+3−1
@@ -156,7 +156,9 @@ newtype Stream m a =
156 -> m r
156 -> m r
157 }
157 }
158
158
159−-- | A monad that can perform asynchronous/concurrent IO operations.
159+-- | A monad that can perform asynchronous/concurrent IO operations. Streams
160+-- that can be composed concurrently require the underlying monad to be
161+-- 'MonadAsync'.
160type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
162type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
161
163
162scons :: a -> Maybe (Stream m a) -> Stream m a
164scons :: a -> Maybe (Stream m a) -> Stream m a
src/Streamly/Streams.hs+41−78
@@ -18,8 +18,6 @@
18--
18--
19module Streamly.Streams
19module Streamly.Streams
20 (
20 (
21− -- * Product Style Composition
22− -- $product
23 Streaming (..)
21 Streaming (..)
24 , MonadAsync
22 , MonadAsync
25
23
@@ -99,48 +97,6 @@ import Data.Semigroup (Semigroup(..))
99import Prelude hiding (drop, take, zipWith)
97import Prelude hiding (drop, take, zipWith)
100import Streamly.Core
98import Streamly.Core
101
99
102−-- $product
103−--
104−-- Different styles of streams represented by different types dictate the
105−-- manner in which a stream is processed in monadic or applicative composition.
106−-- Each type differs in the processing order of the elements or processing
107−-- parallelism. The types have the same underlying representation and
108−-- therefore can be converted back and forth easily to switch the composition
109−-- style at any point.
110−--
111−-- 'StreamT' and 'InterleavedT' process the streams serially but differ in the
112−-- order in which the elements of the streams are processed. 'StreamT' goes
113−-- depth first that is it combine one element of the first stream to every
114−-- element of the second stream before it comes back to process the second
115−-- element of the first stream. 'InterleavedT' takes turns when processing
116−-- elements from each stream and therefore all iterations finish at the same
117−-- time rather than one iteration at a time as in the case of 'StreamT'.
118−-- Depending on the cache behavior or IO behavior 'InterleavedT' might be
119−-- slightly more efficient than 'StreamT' or vice-versa.
120−--
121−-- 'AsyncT' and 'ParallelT' can process the individual elements of the streams
122−-- in parallel. 'AsyncT' is the parallel version of 'StreamT', it follows the
123−-- order of 'StreamT', processing multiple elements of a stream concurrently.
124−-- 'ParallelT' follows the order of 'InterleavedT', processing and interleaving
125−-- both streams concurrently. Also, note that 'fmap' on these streams would
126−-- also work in a concurrent fashion i.e. multiple elements in a stream or
127−-- multiple streams can be mapped concurrently.
128−--
129−-- Zipping types have a different style of composition altogether, they zip the
130−-- streams rather than multiplying them. These types provide only Applicative
131−-- composition and no Monad instances. 'ZipStream' zips serially i.e. it
132−-- produces one element from each stream in a serial manner and then zips the
133−-- two items. 'ZipAsync' on the other hand zips in parallel, it produces one
134−-- element from each stream concurrently and then zips the two.
135−--
136−-- It may help develop a better understanding if we think in terms of stream
137−-- generation (production) side parallelism versus stream processing
138−-- (consumption) side parallelism. The sum style operators (e.g. '<|' or '<|>')
139−-- generate streams in parallel and then combine the results into a unified
140−-- stream. On the other hand the product style composition (e.g. 'AsyncT' or
141−-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are
142−-- processed when it is consumed.
143−
144------------------------------------------------------------------------------
100------------------------------------------------------------------------------
145-- Types that can behave as a Stream
101-- Types that can behave as a Stream
146------------------------------------------------------------------------------
102------------------------------------------------------------------------------
@@ -231,12 +187,19 @@ async m = do
231 return $ fromSVar sv
187 return $ fromSVar sv
232
188
233------------------------------------------------------------------------------
189------------------------------------------------------------------------------
234−-- StreamT iterates serially in the Monad and Applicative compositions
190+-- StreamT
235------------------------------------------------------------------------------
191------------------------------------------------------------------------------
236
192
193+-- | The 'Monad' instance of 'StreamT' combines two streams serially in the
194+-- most straightforward manner, just like the nested 'for' loops in imperative
195+-- paradigm. It moves to the next element in the first stream only after the
196+-- first element has been combined with every element in the second stream. In
197+-- other words, 'StreamT' nests loops in a depth first manner.
198+--
237−-- | Note that 'StreamT' style composition works well operationally for
199+-- Note that 'StreamT' style composition works well operationally even when the
238−-- infinite streams. However if the second stream is infinite we will never
200+-- streams being composed are infinite. However, if the second stream is
239−-- come back to the second iteration of the first stream.
201+-- infinite we will never come back to the second iteration of the first
202+-- stream.
240
203
241newtype StreamT m a = StreamT {getStreamT :: Stream m a}
204newtype StreamT m a = StreamT {getStreamT :: Stream m a}
242 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
205 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
@@ -261,8 +224,6 @@ instance Streaming StreamT where
261-- Monad
224-- Monad
262------------------------------------------------------------------------------
225------------------------------------------------------------------------------
263
226
264−-- | Execute a monadic action sequentially for each element in the 'AsyncT'
265−-- stream, i.e. an iteration finishes completely before the next one starts.
266instance Monad m => Monad (StreamT m) where
227instance Monad m => Monad (StreamT m) where
267 return = pure
228 return = pure
268 (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp yld ->
229 (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp yld ->
@@ -336,13 +297,16 @@ instance (Monad m, Floating a) => Floating (StreamT m a) where
336 logBase = liftA2 logBase
297 logBase = liftA2 logBase
337
298
338------------------------------------------------------------------------------
299------------------------------------------------------------------------------
339−-- InterleavedT interleaves the iterations in the Monad and Applicative
300+-- InterleavedT
340−-- compositions.
341------------------------------------------------------------------------------
301------------------------------------------------------------------------------
342
302
303+-- | The 'Monad' instance of 'InterleavedT' fairly interleaves the iterations
304+-- of the inner and the outer loop, nesting loops in a breadth first manner.
305+--
343−-- | Note that 'InterleavedT' style composition cannot work well operationally
306+-- Note that 'InterleavedT' style composition cannot work well operationally on
344−-- on infinite streams. It will keep accumulating the elements of the stream
307+-- infinite streams. Since the next iteration of the outer loop is started even
308+-- before the first one has finsihed, it will keep accumulating space resulting
345−-- resulting in continuous increase in space consumption.
309+-- in a continuous increase in space consumption.
346--
310--
347newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
311newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
348 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
312 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
@@ -358,8 +322,6 @@ instance Streaming InterleavedT where
358 toStream = getInterleavedT
322 toStream = getInterleavedT
359 fromStream = InterleavedT
323 fromStream = InterleavedT
360
324
361−-- | Execute a monadic action for each element in the stream, in a fairly
362−-- interleaved manner i.e. iterations yield alternately.
363instance Monad m => Monad (InterleavedT m) where
325instance Monad m => Monad (InterleavedT m) where
364 return = pure
326 return = pure
365 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
327 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
@@ -434,16 +396,18 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where
434 logBase = liftA2 logBase
396 logBase = liftA2 logBase
435
397
436------------------------------------------------------------------------------
398------------------------------------------------------------------------------
437−-- AsyncT runs the iterations asynchronously in the Monad and Applicative
399+-- AsyncT
438−-- compositions. More iterations are started in parallel on demand. It can work
439−-- with infinite streams.
440------------------------------------------------------------------------------
400------------------------------------------------------------------------------
441
401
442−-- | Just like 'StreamT', 'AsyncT' style composition works well operationally
402+-- | 'AsyncT' is the concurrent equivalent of 'StreamT'. The type 'AsyncT'
443−-- for infinite streams. However, if the second stream is infinite we will
403+-- follows the same composition scheme as 'StreamT' except that it can run
444−-- never be able to finish the first stream.
404+-- multiple iterations concurrently. More concurrent iterations are started
405+-- only if the previous iterations are not able to produce enough output for
406+-- the consumer.
445--
407--
446−-- Note that 'fmap' on 'AsyncT' would map on multiple elements concurrently.
408+-- Just like 'StreamT', 'AsyncT' style composition works well operationally
409+-- for infinite streams. However, if the second stream is infinite we will
410+-- never be able to start the second iteration of the first stream.
447
411
448newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
412newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
449 deriving (Semigroup, Monoid, MonadTrans)
413 deriving (Semigroup, Monoid, MonadTrans)
@@ -476,9 +440,6 @@ parbind par m f = go m
476 yield a (Just r) = run $ f a `par` (go r)
440 yield a (Just r) = run $ f a `par` (go r)
477 in g Nothing stp yield
441 in g Nothing stp yield
478
442
479−-- | Execute a monadic action for each element in the stream, running
480−-- iterations in parallel, but giving preference to iterations started
481−-- earlier.
482instance MonadAsync m => Monad (AsyncT m) where
443instance MonadAsync m => Monad (AsyncT m) where
483 return = pure
444 return = pure
484 (AsyncT m) >>= f = AsyncT $ parbind par m g
445 (AsyncT m) >>= f = AsyncT $ parbind par m g
@@ -548,17 +509,20 @@ instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where
548 logBase = liftA2 logBase
509 logBase = liftA2 logBase
549
510
550------------------------------------------------------------------------------
511------------------------------------------------------------------------------
551−-- ParallelT runs all iterations in parallel in the Monad and Applicative
512+-- ParallelT
552−-- compositions. Iterations are scheduled in a round-robin fashion.
553------------------------------------------------------------------------------
513------------------------------------------------------------------------------
554
514
515+-- | 'ParallelT' is the concurrent equivalent of 'InterleavedT'. It follows the
516+-- same composition scheme as 'InterleavedT' i.e. fairly interleaving the inner
517+-- and outer loops in a round robin fashion except that it can run multiple
518+-- iterations concurrently.
519+--
555−-- | Note that, just like 'InterleavedT', 'ParallelT' style composition cannot
520+-- Note that, just like 'InterleavedT', 'ParallelT' style composition cannot
556−-- work well operationally on infinite streams. It will keep accumulating the
521+-- work well operationally on infinite streams. Because earlier iterations do
522+-- not finish before the new ones are started, it will keep on accumulating the
557-- elements of the stream resulting in continuous increase in space
523-- elements of the stream resulting in continuous increase in space
558-- consumption.
524-- consumption.
559--
525--
560−-- Note that 'fmap' on 'ParallelT' would map on both streams concurrently.
561−--
562newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
526newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
563 deriving (Semigroup, Monoid, MonadTrans)
527 deriving (Semigroup, Monoid, MonadTrans)
564
528
@@ -575,9 +539,6 @@ instance Streaming ParallelT where
575 toStream = getParallelT
539 toStream = getParallelT
576 fromStream = ParallelT
540 fromStream = ParallelT
577
541
578−-- | Execute a monadic action for each element in the stream, running
579−-- iterations in a fairly parallel manner, i.e. all iterations are equally
580−-- likely to run.
581instance MonadAsync m => Monad (ParallelT m) where
542instance MonadAsync m => Monad (ParallelT m) where
582 return = pure
543 return = pure
583 (ParallelT m) >>= f = ParallelT $ parbind par m g
544 (ParallelT m) >>= f = ParallelT $ parbind par m g
@@ -870,7 +831,8 @@ infixr 5 <=>
870-- | Sequential interleaved composition, in contrast to '<>' this operator
831-- | Sequential interleaved composition, in contrast to '<>' this operator
871-- fairly interleaves the two streams instead of appending them; yielding one
832-- fairly interleaves the two streams instead of appending them; yielding one
872-- element from each stream alternately. Unlike '<>' it cannot be used to fold
833-- element from each stream alternately. Unlike '<>' it cannot be used to fold
873−-- an infinite container of streams.
834+-- an infinite container of streams. This operator corresponds to the
835+-- 'InterleavedT' type for product style composition.
874{-# INLINE (<=>) #-}
836{-# INLINE (<=>) #-}
875(<=>) :: Streaming t => t m a -> t m a -> t m a
837(<=>) :: Streaming t => t m a -> t m a -> t m a
876m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
838m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
@@ -880,7 +842,8 @@ m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
880-- them. It keeps yielding from the stream on the left as long as it can. If
842-- them. It keeps yielding from the stream on the left as long as it can. If
881-- the left stream blocks or cannot keep up with the pace of the consumer it
843-- the left stream blocks or cannot keep up with the pace of the consumer it
882-- can yield from the stream on the right in parallel. Unlike '<|>' it can be
844-- can yield from the stream on the right in parallel. Unlike '<|>' it can be
883−-- used to fold infinite containers of streams.
845+-- used to fold infinite containers of streams. This operator corresponds to
846+-- the 'AsyncT' type for product style composition.
884{-# INLINE (<|) #-}
847{-# INLINE (<|) #-}
885(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
848(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
886m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
849m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)