Streamly.Data.StreamK

A newtype wrapper for the StreamK type adding a cross product style monad instance.

A Monad bind behaves like a for loop:

>>> :{
Stream.fold Fold.toList $ StreamK.toStream $ StreamK.unCross $ do
    x <- StreamK.mkCross $ StreamK.fromStream $ Stream.fromList [1,2]
    -- Perform the following actions for each x in the stream
    return x
:}
[1,2]

Nested monad binds behave like nested for loops:

>>> :{
Stream.fold Fold.toList $ StreamK.toStream $ StreamK.unCross $ do
    x <- StreamK.mkCross $ StreamK.fromStream $ Stream.fromList [1,2]
    y <- StreamK.mkCross $ StreamK.fromStream $ Stream.fromList [3,4]
    -- Perform the following actions for each x, for each y
    return (x, y)
:}
[(1,3),(1,4),(2,3),(2,4)]

Unwrap the StreamK type from CrossStreamK newtype.

This is a type level operation with no runtime overhead.

Wrap the StreamK type in a CrossStreamK newtype to enable cross product style applicative and monad instances.

This is a type level operation with no runtime overhead.

A stream that terminates without producing any output or side effect.

>>> Stream.fold Fold.toList (StreamK.toStream StreamK.nil)
[]

A stream that terminates without producing any output, but produces a side effect.

>>> Stream.fold Fold.toList (StreamK.toStream (StreamK.nilM (print "nil")))
"nil"
[]

Pre-release

A right associative prepend operation to add a pure value at the head of an existing stream:

>>> s = 1 `StreamK.cons` 2 `StreamK.cons` 3 `StreamK.cons` StreamK.nil
>>> Stream.fold Fold.toList (StreamK.toStream s)
[1,2,3]

Unlike Streamly.Data.Stream cons StreamK cons can be used recursively:

>>> repeat x = let xs = StreamK.cons x xs in xs
>>> fromFoldable = Prelude.foldr StreamK.cons StreamK.nil

cons is same as the following but more efficient:

>>> cons x xs = return x `StreamK.consM` xs

A right associative prepend operation to add an effectful value at the head of an existing stream::

>>> s = putStrLn "hello" `StreamK.consM` putStrLn "world" `StreamK.consM` StreamK.nil
>>> Stream.fold Fold.drain (StreamK.toStream s)
hello
world

It can be used efficiently with foldr:

>>> fromFoldableM = Prelude.foldr StreamK.consM StreamK.nil

Same as the following but more efficient:

>>> consM x xs = StreamK.fromEffect x `StreamK.append` xs

Convert a fused Stream to StreamK.

For example:

>>> s1 = StreamK.fromStream $ Stream.fromList [1,2]
>>> s2 = StreamK.fromStream $ Stream.fromList [3,4]
>>> Stream.fold Fold.toList $ StreamK.toStream $ s1 `StreamK.append` s2
[1,2,3,4]

Convert a StreamK to a fused Stream.

>>> fromFoldable = Prelude.foldr StreamK.cons StreamK.nil

Construct a stream from a Foldable containing pure values:

Run a ParserK over a StreamK. Please use parseChunks where possible, for better performance.

Similar to parseBreak but works on singular elements.

Run a ParserK over a chunked StreamK and return the parse result and the remaining Stream.

Unlike Streamly.Data.Stream append StreamK append can be used recursively:

>>> cycle xs = let ys = xs `StreamK.append` ys in ys

Note: When joining many streams in a left associative manner earlier streams will get exponential priority than the ones joining later. Because of exponentially high weighting of left streams it can be used with concatMapWith even on a large number of streams.

Merging of n streams can be performed by combining the streams pair wise using mergeMapWith to give O(n * log n) time complexity. If used with concatMapWith it will have O(n^2) performance.

Zipping of n streams can be performed by combining the streams pair wise using mergeMapWith with O(n * log n) time complexity. If used with concatMapWith it will have O(n^2) performance.

Definition:

>>> crossWith f m1 m2 = fmap f m1 `StreamK.crossApply` m2

Note that the second stream is evaluated multiple times.

Perform a concatMap using a specified concat strategy. The first argument specifies a merge or concat function that is used to merge the streams generated by the map function.

For example, interleaving n streams in a left biased manner:

>>> fromList = StreamK.fromStream . Stream.fromList
>>> toList = Stream.toList . StreamK.toStream
>>> lists = fromList [[1,5],[2,6],[3,7],[4,8]]
>>> toList $ StreamK.concatMapWith StreamK.interleave fromList lists
[1,2,5,3,6,4,7,8]

For a fair interleaving example see mergeMapWith.

Combine streams in pairs using a binary combinator, the resulting streams are then combined again in pairs recursively until we get to a single combined stream. The composition would thus form a binary tree.

For example, sorting a stream using merge sort:

>>> fromList = StreamK.fromStream . Stream.fromList
>>> toList = Stream.toList . StreamK.toStream
>>> generate = StreamK.fromPure
>>> combine = StreamK.mergeBy compare
>>> toList $ StreamK.mergeMapWith combine generate (fromList [5,1,7,9,2])
[1,2,5,7,9]

Interleaving n streams in a balanced manner:

>>> lists = fromList [[1,4,7],[2,5,8],[3,6,9]]
>>> toList $ StreamK.mergeMapWith StreamK.interleave fromList lists
[1,3,2,6,4,9,5,7,8]

See unfoldEachInterleave for a much faster fused version of the above example.

Note that if the stream length is not a power of 2, the binary tree composed by mergeMapWith is not balanced, which may or may not be important depending on what you are trying to achieve. This also explains the order of the output in the interleaving example above.

Caution: the stream of streams must be finite

Pre-release

Sort the input stream using a supplied comparison function.

Sorting can be achieved by simply:

>>> sortBy cmp = StreamK.mergeMapWith (StreamK.mergeBy cmp) StreamK.fromPure

However, this combinator uses a parser to first split the input stream into down and up sorted segments and then merges them to optimize sorting when pre-sorted sequences exist in the input stream.

O(n) space

Like Streamly.Data.Stream.handle but with one significant difference, this function observes exceptions from the consumer of the stream as well.

You can also convert StreamK to Stream and use exception handling from Stream module:

>>> handle f s = StreamK.fromStream $ Stream.handle (\e -> StreamK.toStream (f e)) (StreamK.toStream s)

Like Streamly.Data.Stream.bracketIO but with one significant difference, this function observes exceptions from the consumer of the stream as well. Therefore, it cleans up the resource promptly when the consumer encounters an exception.

You can also convert StreamK to Stream and use resource handling from Stream module:

>>> bracketIO bef aft bet = StreamK.fromStream $ Stream.bracketIO bef aft (StreamK.toStream . bet)