concat -package:ghc -package:vector -package:dlist package:streamly

Flatten a stream of streams to a single stream.

concat = concatMap id

Pre-release

A variant of fold that allows you to fold a Foldable container of streams using the specified stream sum operation.

concatFoldableWith async $ map return [1..3]

Equivalent to:

concatFoldableWith f = Prelude.foldr f D.nil
concatFoldableWith f = D.concatMapFoldableWith f id

Since: 0.8.0 (Renamed foldWith to concatFoldableWith) Since: 0.1.0 (Streamly)

Like concatMapFoldableWith but with the last two arguments reversed i.e. the monadic streaming function is the last argument. Equivalent to:

concatForFoldableWith f xs g = Prelude.foldr (f . g) D.nil xs
concatForFoldableWith f = flip (D.concatMapFoldableWith f)

Since: 0.8.0 (Renamed forEachWith to concatForFoldableWith) Since: 0.1.0 (Streamly)

Given a stream value in the underlying monad, lift and join the underlying monad with the stream monad.

>>> concatM = Stream.concat . Stream.fromEffect

>>> concatM = Stream.concat . lift    -- requires (MonadTrans t)

>>> concatM = join . lift             -- requires (MonadTrans t, Monad (t m))

See also: concat, sequence Internal

Map a stream producing function on each element of the stream and then flatten the results into a single stream.

>>> concatMap f = Stream.concatMapM (return . f)

>>> concatMap f = Stream.concatMapWith Stream.serial f

>>> concatMap f = Stream.concat . Stream.map f

A variant of foldMap that allows you to map a monadic streaming action on a Foldable container and then fold it using the specified stream merge operation.

concatMapFoldableWith async return [1..3]

Equivalent to:

concatMapFoldableWith f g = Prelude.foldr (f . g) S.nil
concatMapFoldableWith f g xs = S.concatMapWith f g (S.fromFoldable xs)

Since: 0.8.0 (Renamed foldMapWith to concatMapFoldableWith) Since: 0.1.0 (Streamly)

Map a stream producing monadic function on each element of the stream and then flatten the results into a single stream. Since the stream generation function is monadic, unlike concatMap, it can produce an effect at the beginning of each iteration of the inner loop.

concatMapWith mixer generator stream is a two dimensional looping combinator. The generator function is used to generate streams from the elements in the input stream and the mixer function is used to merge those streams. Note we can merge streams concurrently by using a concurrent merge function. Since: 0.7.0 Since: 0.8.0 (signature change)

Combine streams in pairs using a binary stream combinator, then combine the resulting streams in pairs recursively until we get to a single combined stream. For example, you can sort a stream using merge sort like this:

>>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.fromPure $ Stream.fromList [5,1,7,9,2]
[1,2,5,7,9]

Caution: the stream of streams must be finite Pre-release

Like concatMapWith but carries a state which can be used to share information across multiple steps of concat.

concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial

Pre-release

Evaluate the streams in the input stream concurrently and combine them.

>>> parConcat modifier = Stream.parConcatMap modifier id

Same as concatIterate but concurrent. Pre-release

Map each element of the input to a stream and then concurrently evaluate and concatenate the resulting streams. Multiple streams may be evaluated concurrently but earlier streams are perferred. Output from the streams are used as they arrive. Definition:

>>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream

Examples:

>>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs

The following streams finish in 4 seconds:

>>> stream1 = Stream.fromEffect (delay 4)

>>> stream2 = Stream.fromEffect (delay 2)

>>> stream3 = Stream.fromEffect (delay 1)

>>> f id [stream1, stream2, stream3]
1 sec
2 sec
4 sec
[1,2,4]

Limiting threads to 2 schedules the third stream only after one of the first two has finished, releasing a thread:

>>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
...
[2,1,4]

When used with a Single thread it behaves like serial concatMap:

>>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
...
[4,2,1]

>>> stream1 = Stream.fromList [1,2,3]

>>> stream2 = Stream.fromList [4,5,6]

>>> f (Stream.maxThreads 1) [stream1, stream2]
[1,2,3,4,5,6]

Schedule all streams in a round robin fashion over the available threads:

>>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs

>>> stream1 = Stream.fromList [1,2,3]

>>> stream2 = Stream.fromList [4,5,6]

>>> f (Stream.maxThreads 1) [stream1, stream2]
[1,4,2,5,3,6]

Fold a stream of monoid elements by appending them.

mconcat = Stream.fold Fold.mconcat

Pre-release

Make a concurrent stream evaluator from a stream, to be used in withChannelK or toChannelK. Maps a stream generation function on each element of the stream, the evaluation of the map on each element happens concurrently. All the generated streams are merged together in the output of the channel. The scheduling and termination behavior depends on the channel settings. Note that if you queue a stream on the channel using toChannelK, it will be picked up by a worker and the worker would evaluate the entire stream serially and emit the results on the channel. However, if you transform the stream using parConcatMapChanK and queue it on the channel, it parallelizes the function map on each element of the stream. The simplest example is parConcatMapChanK id id which is equivalent to evaluating each element of the stream concurrently. A channel worker evaluating this function would enqueue the tail on the channel's work queue and go on to evaluate the head generating an output stream. The tail is picked up by another worker which does the same and so on.