Streamly.Internal.Data.Stream.Async

For AsyncT streams:

(<>) = Streamly.Prelude.async
(>>=) = flip . Streamly.Prelude.concatMapWith Streamly.Prelude.async

A single Monad bind behaves like a for loop with iterations of the loop executed concurrently a la the async combinator, producing results and side effects of iterations out of order:

>>> :{Stream.toList $ Stream.fromAsync $ do     x <- Stream.fromList [2,1] -- foreach x in stream     Stream.fromEffect $ delay x:}1 sec2 sec[1,2]

Nested monad binds behave like nested for loops with nested iterations executed concurrently, a la the async combinator:

>>> :{Stream.toList $ Stream.fromAsync $ do    x <- Stream.fromList [1,2] -- foreach x in stream    y <- Stream.fromList [2,4] -- foreach y in stream    Stream.fromEffect $ delay (x + y):}3 sec4 sec5 sec6 sec[3,4,5,6]

The behavior can be explained as follows. All the iterations corresponding to the element 1 in the first stream constitute one output stream and all the iterations corresponding to 2 constitute another output stream and these two output streams are merged using async.

Since: 0.1.0 (Streamly)

A demand driven left biased parallely composing IO stream of elements of type a. See AsyncT documentation for more details.

Since: 0.2.0 (Streamly)

XXX we can implement it more efficienty by directly implementing instead of combining streams using async.

Generate a stream asynchronously to keep it buffered, lazily consume from the buffer.

Pre-release

For WAsyncT streams:

(<>) = Streamly.Prelude.wAsync
(>>=) = flip . Streamly.Prelude.concatMapWith Streamly.Prelude.wAsync

A single Monad bind behaves like a for loop with iterations of the loop executed concurrently a la the wAsync combinator, producing results and side effects of iterations out of order:

>>> :{Stream.toList $ Stream.fromWAsync $ do     x <- Stream.fromList [2,1] -- foreach x in stream     Stream.fromEffect $ delay x:}1 sec2 sec[1,2]

Nested monad binds behave like nested for loops with nested iterations executed concurrently, a la the wAsync combinator:

>>> :{Stream.toList $ Stream.fromWAsync $ do    x <- Stream.fromList [1,2] -- foreach x in stream    y <- Stream.fromList [2,4] -- foreach y in stream    Stream.fromEffect $ delay (x + y):}3 sec4 sec5 sec6 sec[3,4,5,6]

The behavior can be explained as follows. All the iterations corresponding to the element 1 in the first stream constitute one WAsyncT output stream and all the iterations corresponding to 2 constitute another WAsyncT output stream and these two output streams are merged using wAsync.

The W in the name stands for wide or breadth wise scheduling in contrast to the depth wise scheduling behavior of AsyncT.

Since: 0.2.0 (Streamly)

A round robin parallely composing IO stream of elements of type a. See WAsyncT documentation for more details.

Since: 0.2.0 (Streamly)

XXX we can implement it more efficienty by directly implementing instead of combining streams using wAsync.