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

Flatten a stream of streams to a single stream.

concat = concatMap id

Pre-release

concatMapDivK useTail useHead stream, divides the stream in head and tail, maps a stream generator on the head and maps an action on the tail of a stream. Returns the stream generated by the head. Used for concurrent evaluation of streams using a Channel.

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)

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)

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)

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

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.

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

Reduce a non-empty list with <> The default definition should be sufficient, but this can be overridden for efficiency.

>>> import Data.List.NonEmpty (NonEmpty (..))

>>> sconcat $ "Hello" :| [" ", "Haskell", "!"]
"Hello Haskell!"

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