A stream consists of a step function that generates the next step given a current state, and the current state.

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

>>> Stream.toList Stream.nil
[]

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

>>> nilM action = Stream.before action Stream.nil
>>> Stream.fold Fold.toList (Stream.nilM (print "nil"))
"nil"
[]

Pre-release

WARNING! O(n^2) time complexity wrt number of elements. Use the O(n) complexity StreamK.cons unless you want to statically fuse just a few elements.

Fuse a pure value at the head of an existing stream::

>>> s = 1 `Stream.cons` Stream.fromList [2,3]
>>> Stream.toList s
[1,2,3]

Definition:

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

Like cons but fuses an effect instead of a pure value.

Build a stream by unfolding a pure step function step starting from a seed s. The step function returns the next element in the stream and the next seed value. When it is done it returns Nothing and the stream ends. For example,

>>> :{
let f b =
        if b > 2
        then Nothing
        else Just (b, b + 1)
in Stream.toList $ Stream.unfoldr f 0
:}
[0,1,2]

Build a stream by unfolding a monadic step function starting from a seed. The step function returns the next element in the stream and the next seed value. When it is done it returns Nothing and the stream ends. For example,

>>> :{
let f b =
        if b > 2
        then return Nothing
        else return (Just (b, b + 1))
in Stream.toList $ Stream.unfoldrM f 0
:}
[0,1,2]

Create a singleton stream from a pure value.

>>> fromPure a = a `Stream.cons` Stream.nil
>>> fromPure = pure
>>> fromPure = Stream.fromEffect . pure

Create a singleton stream from a monadic action.

>>> fromEffect m = m `Stream.consM` Stream.nil
>>> fromEffect = Stream.sequence . Stream.fromPure

>>> Stream.fold Fold.drain $ Stream.fromEffect (putStrLn "hello")
hello

Generate an infinite stream with x as the first element and each successive element derived by applying the function f on the previous element.

>>> Stream.toList $ Stream.take 5 $ Stream.iterate (+1) 1
[1,2,3,4,5]

Generate an infinite stream with the first element generated by the action m and each successive element derived by applying the monadic function f on the previous element.

>>> :{
Stream.iterateM (\x -> print x >> return (x + 1)) (return 0)
    & Stream.take 3
    & Stream.toList
:}
0
1
[0,1,2]

Generate an infinite stream by repeating a pure value.

>>> repeat = Stream.iterate id
>>> repeat x = Stream.repeatM (pure x)

>>> repeatM act = Stream.iterateM (const act) act
>>> repeatM = Stream.sequence . Stream.repeat

Generate a stream by repeatedly executing a monadic action forever.

>>> :{
repeatAction =
       Stream.repeatM (threadDelay 1000000 >> print 1)
     & Stream.take 10
     & Stream.fold Fold.drain
:}

>>> replicate n = Stream.take n . Stream.repeat
>>> replicate n x = Stream.replicateM n (pure x)

Generate a stream of length n by repeating a value n times.

>>> replicateM n = Stream.sequence . Stream.replicate n

Generate a stream by performing a monadic action n times.

Types that can be enumerated as a stream. The operations in this type class are equivalent to those in the Enum type class, except that these generate a stream instead of a list. Use the functions in Streamly.Internal.Data.Stream.Enumeration module to define new instances.

enumerate = enumerateFrom minBound

Enumerate a Bounded type from its minBound to maxBound

>>> enumerateTo = Stream.enumerateFromTo minBound

Enumerate a Bounded type from its minBound to specified value.

Construct a stream from a list of pure values.

Convert an Unfold into a stream by supplying it an input seed.

>>> s = Stream.unfold Unfold.replicateM (3, putStrLn "hello")
>>> Stream.fold Fold.drain s
hello
hello
hello

Decompose a stream into its head and tail. If the stream is empty, returns Nothing. If the stream is non-empty, returns Just (a, ma), where a is the head of the stream and ma its tail.

Properties:

>>> Nothing <- Stream.uncons Stream.nil
>>> Just ("a", t) <- Stream.uncons (Stream.cons "a" Stream.nil)

This can be used to consume the stream in an imperative manner one element at a time, as it just breaks down the stream into individual elements and we can loop over them as we deem fit. For example, this can be used to convert a streamly stream into other stream types.

All the folds in this module can be expressed in terms of uncons, however, this is generally less efficient than specific folds because it takes apart the stream one element at a time, therefore, does not take adavantage of stream fusion.

foldBreak is a more general way of consuming a stream piecemeal.

>>> :{
uncons xs = do
    r <- Stream.foldBreak Fold.one xs
    return $ case r of
        (Nothing, _) -> Nothing
        (Just h, t) -> Just (h, t)
:}

Fold a stream using the supplied left Fold and reducing the resulting expression strictly at each step. The behavior is similar to foldl'. A Fold can terminate early without consuming the full stream. See the documentation of individual Folds for termination behavior.

Definitions:

>>> fold f = fmap fst . Stream.foldBreak f
>>> fold f = Stream.parse (Parser.fromFold f)

Example:

>>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
5050

Like fold but also returns the remaining stream. The resulting stream would be nil if the stream finished before the fold.

Parse a stream using the supplied Parser.

Parsers (See Streamly.Internal.Data.Parser) are more powerful folds that add backtracking and error functionality to terminating folds. Unlike folds, parsers may not always result in a valid output, they may result in an error. For example:

>>> Stream.parse (Parser.takeEQ 1 Fold.drain) Stream.nil
Left (ParseError "takeEQ: Expecting exactly 1 elements, input terminated on 0")

Note: parse p is not the same as head . parseMany p on an empty stream.

Parse a stream using the supplied Parser.

Like parse but includes stream position information in the error messages.

>>> Stream.parsePos (Parser.takeEQ 2 Fold.drain) (Stream.fromList [1])
Left (ParseErrorPos 1 "takeEQ: Expecting exactly 2 elements, input terminated on 1")

Like parseBreak but includes stream position information in the error messages.

Right associative/lazy pull fold. foldrM build final stream constructs an output structure using the step function build. build is invoked with the next input element and the remaining (lazy) tail of the output structure. It builds a lazy output expression using the two. When the "tail structure" in the output expression is evaluated it calls build again thus lazily consuming the input stream until either the output expression built by build is free of the "tail" or the input is exhausted in which case final is used as the terminating case for the output structure. For more details see the description in the previous section.

Example, determine if any element is odd in a stream:

>>> s = Stream.fromList (2:4:5:undefined)
>>> step x xs = if odd x then return True else xs
>>> Stream.foldrM step (return False) s
True

>>> import Control.Monad (join)
>>> foldrM f z = join . Stream.foldr f z

Right fold, lazy for lazy monads and pure streams, and strict for strict monads.

Please avoid using this routine in strict monads like IO unless you need a strict right fold. This is provided only for use in lazy monads (e.g. Identity) or pure streams. Note that with this signature it is not possible to implement a lazy foldr when the monad m is strict. In that case it would be strict in its accumulator and therefore would necessarily consume all its input.

>>> foldr f z = Stream.foldrM (\a b -> f a <$> b) (return z)

Note: This is similar to Fold.foldr' (the right fold via left fold), but could be more efficient.

Definitions:

>>> toList = Stream.foldr (:) []
>>> toList = Stream.fold Fold.toList

Convert a stream into a list in the underlying monad. The list can be consumed lazily in a lazy monad (e.g. Identity). In a strict monad (e.g. IO) the whole list is generated and buffered before it can be consumed.

Warning! working on large lists accumulated as buffers in memory could be very inefficient, consider using Streamly.Data.Array instead.

Note that this could a bit more efficient compared to Stream.fold Fold.toList, and it can fuse with pure list consumers.

>>> sequence = Stream.mapM id

Replace the elements of a stream of monadic actions with the outputs of those actions.

>>> s = Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
>>> Stream.fold Fold.drain $ Stream.sequence s
abc

>>> mapM f = Stream.sequence . fmap f

Apply a monadic function to each element of the stream and replace it with the output of the resulting action.

>>> s = Stream.fromList ["a", "b", "c"]
>>> Stream.fold Fold.drain $ Stream.mapM putStr s
abc

This is functional equivalent of an imperative loop.

Apply a monadic function to each element flowing through the stream and discard the results.

>>> s = Stream.enumerateFromTo 1 2
>>> Stream.fold Fold.drain $ Stream.trace print s
1
2

Compare with tap.

Tap the data flowing through a stream into a Fold. For example, you may add a tap to log the contents flowing through the stream. The fold is used only for effects, its result is discarded.

                  Fold m a b
                      |
-----stream m a ---------------stream m a-----

>>> s = Stream.enumerateFromTo 1 2
>>> Stream.fold Fold.drain $ Stream.tap (Fold.drainMapM print) s
1
2

Compare with trace.

Introduce a delay of specified seconds between elements of the stream.

Definition:

>>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
>>> delay = Stream.intersperseM_ . sleep

Example:

>>> input = Stream.enumerateFromTo 1 3
>>> Stream.fold (Fold.drainMapM print) $ Stream.delay 1 input
1
2
3

Strict left scan. Scan a stream using the given fold. Scan includes the initial (default) value of the accumulator as well as the final value. Compare with postscan which omits the initial value.

>>> s = Stream.fromList [1..10]
>>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scanl Scanl.sum s
[0,1,3,6]

See also: usingStateT

Postscan a stream using the given fold. A postscan omits the initial (default) value of the accumulator and includes the final value.

>>> Stream.toList $ Stream.postscanl Scanl.latest (Stream.fromList [])
[]

Compare with scan which includes the initial value as well:

>>> Stream.toList $ Stream.scanl Scanl.latest (Stream.fromList [])
[Nothing]

The following example extracts the input stream up to a point where the running average of elements is no more than 10:

>>> import Data.Maybe (fromJust)
>>> let avg = Scanl.teeWith (/) Scanl.sum (fmap fromIntegral Scanl.length)
>>> s = Stream.enumerateFromTo 1.0 100.0
>>> :{
 Stream.fold Fold.toList
  $ fmap (fromJust . fst)
  $ Stream.takeWhile (\(_,x) -> x <= 10)
  $ Stream.postscanl (Scanl.tee Scanl.latest avg) s
:}
[1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]

>>> f = Scanl.mkScanl (\(i, _) x -> (i + 1, x)) (-1,undefined)
>>> indexed = Stream.postscanl f
>>> indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
>>> indexedR n = fmap (\(i, a) -> (n - i, a)) . indexed

Pair each element in a stream with its index, starting from index 0.

>>> Stream.fold Fold.toList $ Stream.indexed $ Stream.fromList "hello"
[(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]

insertBy cmp elem stream inserts elem before the first element in stream that is less than elem when compared using cmp.

>>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)

>>> input = Stream.fromList [1,3,5]
>>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
[1,2,3,5]

Effectful variant of intersperse. Insert an effect and its output between successive elements of a stream. It does nothing if stream has less than two elements.

Definition:

>>> intersperseM x = Stream.interleaveSepBy (Stream.repeatM x)

Perform a side effect between two successive elements of a stream. It does nothing if the stream has less than two elements.

>>> f x y = Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ x $ Stream.fromList y
>>> f (putChar '.') "abc"
a.b.c
>>> f (putChar '.') "a"
a

Pre-release

Insert a pure value between successive elements of a stream. It does nothing if stream has less than two elements.

Definition:

>>> intersperse x = Stream.intersperseM (return x)
>>> intersperse x = Stream.unfoldEachSepBy x Unfold.identity
>>> intersperse x = Stream.unfoldEachSepBySeq x Unfold.identity
>>> intersperse x = Stream.interleaveSepBy (Stream.repeat x)

Example:

>>> f x y = Stream.toList $ Stream.intersperse x $ Stream.fromList y
>>> f ',' "abc"
"a,b,c"
>>> f ',' "a"
"a"

Map a Maybe returning function to a stream, filter out the Nothing elements, and return a stream of values extracted from Just.

Equivalent to:

>>> mapMaybe f = Stream.catMaybes . fmap f

Like mapMaybe but maps a monadic function.

Equivalent to:

>>> mapMaybeM f = Stream.catMaybes . Stream.mapM f

>>> mapM f = Stream.mapMaybeM (\x -> Just <$> f x)

Include only those elements that pass a predicate.

>>> filter p = Stream.filterM (return . p)
>>> filter p = Stream.mapMaybe (\x -> if p x then Just x else Nothing)
>>> filter p = Stream.postscanlMaybe (Scanl.filtering p)

Same as filter but with a monadic predicate.

>>> f p x = p x >>= \r -> return $ if r then Just x else Nothing
>>> filterM p = Stream.mapMaybeM (f p)

In a stream of Maybes, discard Nothings and unwrap Justs.

>>> catMaybes = Stream.mapMaybe id
>>> catMaybes = fmap fromJust . Stream.filter isJust

Pre-release

Discard Rights and unwrap Lefts in an Either stream.

>>> catLefts = fmap (fromLeft undefined) . Stream.filter isLeft

Pre-release

Discard Lefts and unwrap Rights in an Either stream.

>>> catRights = fmap (fromRight undefined) . Stream.filter isRight

Pre-release

Remove the either wrapper and flatten both lefts and as well as rights in the output stream.

>>> catEithers = fmap (either id id)

Pre-release

Take first n elements from the stream and discard the rest.

End the stream as soon as the predicate fails on an element.

Same as takeWhile but with a monadic predicate.

Discard first n elements from the stream and take the rest.

Drop elements in the stream as long as the predicate succeeds and then take the rest of the stream.

Same as dropWhile but with a monadic predicate.

WARNING! O(n^2) time complexity wrt number of streams. Suitable for statically fusing a small number of streams. Use the O(n) complexity StreamK.append otherwise.

Fuses two streams sequentially, yielding all elements from the first stream, and then all elements from the second stream.

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

WARNING! O(n^2) time complexity wrt number of streams. Suitable for statically fusing a small number of streams. Use the O(n) complexity StreamK.interleave otherwise.

Interleaves two streams, yielding one element from each stream alternately, starting from the first stream. When one stream is exhausted, all the remaining elements of the other stream are emitted in the output stream.

Both the streams are completely exhausted.

(a b c) (. . .) => a . b . c .
(a b c) (. .  ) => a . b . c
(a b  ) (. . .) => a . b .  .

Examples:

>>> f x y = Stream.toList $ Stream.interleave (Stream.fromList x) (Stream.fromList y)
>>> f "abc" "..."
"a.b.c."
>>> f "abc" ".."
"a.b.c"
>>> f "ab" "..."
"a.b.."

WARNING! O(n^2) time complexity wrt number of streams. Suitable for statically fusing a small number of streams. Use the O(n) complexity StreamK.mergeBy otherwise.

Merge two streams using a comparison function. The head elements of both the streams are compared and the smaller of the two elements is emitted, if both elements are equal then the element from the first stream is used first.

If the streams are sorted in ascending order, the resulting stream would also remain sorted in ascending order.

>>> s1 = Stream.fromList [1,3,5]
>>> s2 = Stream.fromList [2,4,6,8]
>>> Stream.fold Fold.toList $ Stream.mergeBy compare s1 s2
[1,2,3,4,5,6,8]

Like mergeBy but with a monadic comparison function.

Example, to merge two streams randomly:

> randomly _ _ = randomIO >>= x -> return $ if x then LT else GT
> Stream.toList $ Stream.mergeByM randomly (Stream.fromList [1,1,1,1]) (Stream.fromList [2,2,2,2])
[2,1,2,2,2,1,1,1]

Example, merge two streams in a proportion of 2:1:

>>> :set -fno-warn-unrecognised-warning-flags
>>> :set -fno-warn-x-partial
>>> :{
do
 let s1 = Stream.fromList [1,1,1,1,1,1]
     s2 = Stream.fromList [2,2,2]
 let proportionately m n = do
      ref <- newIORef $ cycle $ Prelude.concat [Prelude.replicate m LT, Prelude.replicate n GT]
      return $ \_ _ -> do
         r <- readIORef ref
         writeIORef ref $ Prelude.tail r
         return $ Prelude.head r
 f <- proportionately 2 1
 xs <- Stream.fold Fold.toList $ Stream.mergeByM f s1 s2
 print xs
:}
[1,1,2,1,1,2,1,1,2]

WARNING! O(n^2) time complexity wrt number of streams. Suitable for statically fusing a small number of streams. Use the O(n) complexity StreamK.zipWith otherwise.

Stream a is evaluated first, followed by stream b, the resulting elements a and b are then zipped using the supplied zip function and the result c is yielded to the consumer.

If stream a or stream b ends, the zipped stream ends. If stream b ends first, the element a from previous evaluation of stream a is discarded.

>>> s1 = Stream.fromList [1,2,3]
>>> s2 = Stream.fromList [4,5,6]
>>> Stream.fold Fold.toList $ Stream.zipWith (+) s1 s2
[5,7,9]

Like zipWith but using a monadic zipping function.

Definition:

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

Note that the second stream is evaluated multiple times.

Also see "Streamly.Data.Unfold.crossWith" for fast fusible static cross product option.

Given a Stream m a and Stream m b generate a stream with all possible combinations of the tuple (a, b).

Definition:

>>> cross = Stream.crossWith (,)

The second stream is evaluated multiple times. If that is not desired it can be cached in an Array and then generated from the array before calling this function. Caching may also improve performance if the stream is expensive to evaluate.

Time: O(m x n)

Pre-release

Like cross but interleaves the outer and inner loops fairly. See fairConcatFor for more details.

unfoldEach unfold stream uses unfold to map the input stream elements to streams and then flattens the generated streams into a single output stream.

Like concatMap but uses an Unfold for stream generation. Unlike concatMap this can fuse the Unfold code with the inner loop and therefore provide many times better performance.

>>> concatMap f = Stream.unfoldEach (Unfold.lmap f Unfold.fromStream)

Here is an example of a two level nested loop much faster than concatMap based nesting.

>>> :{
outerLoop =
  flip Stream.mapM (Stream.fromList [1,2,3]) $ \x -> do
      liftIO $ putStrLn (show x)
      return x
innerUnfold = Unfold.carry $ Unfold.lmap (const [4,5,6]) Unfold.fromList
innerLoop =
     flip Unfold.mapM innerUnfold $ \(x, y) -> do
         when (x == 1) $ liftIO $ putStrLn (show y)
         pure $ (x, y)
:}

>>> Stream.toList $ Stream.unfoldEach innerLoop outerLoop
1
4
5
6
2
3
[(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]

Like unfoldEach but interleaves the resulting streams in a breadth first manner instead of appending them. Unfolds each element in the input stream to a stream and then interleave the resulting streams.

>>> lists = Stream.fromList [[1,4,7],[2,5,8],[3,6,9]]
>>> Stream.toList $ Stream.bfsUnfoldEach Unfold.fromList lists
[1,2,3,4,5,6,7,8,9]

CAUTION! Do not use on infinite streams.

See fairConcatFor for more details. This is similar except that this uses unfolds, therefore, it is much faster due to fusion.

>>> :{
outerLoop = Stream.fromList [1,2,3]
innerLoop = Unfold.carry $ Unfold.lmap (const [4,5,6]) Unfold.fromList
:}

>>> Stream.toList $ Stream.fairUnfoldEach innerLoop outerLoop
[(1,4),(1,5),(2,4),(1,6),(2,5),(3,4),(2,6),(3,5),(3,6)]

Unfold each element of the stream, separate the successive unfolds by a sequence generated by unfolding the supplied value.

Definition:

>>> unfoldEachSepBySeq a u = Stream.unfoldEach u . Stream.intersperse a
>>> unfoldEachSepBySeq a u = Stream.intercalateSepBy u (Stream.repeat a) u

Idioms:

>>> intersperse x = Stream.unfoldEachSepBySeq x Unfold.identity
>>> unwords = Stream.unfoldEachSepBySeq " " Unfold.fromList

Usage:

>>> input = Stream.fromList ["abc", "def", "ghi"]
>>> Stream.toList $ Stream.unfoldEachSepBySeq " " Unfold.fromList input
"abc def ghi"

Unfold each element of the stream, end each unfold by a sequence generated by unfolding the supplied value.

Definition:

>>> unfoldEachEndBySeq a u = Stream.unfoldEach u . Stream.intersperseEndByM a
>>> unfoldEachEndBySeq a u = Stream.intercalateEndBy u (Stream.repeat a) u

Idioms:

>>> intersperseEndByM x = Stream.unfoldEachEndBySeq x Unfold.identity
>>> unlines = Stream.unfoldEachEndBySeq "\n" Unfold.fromList

Usage:

>>> input = Stream.fromList ["abc", "def", "ghi"]
>>> Stream.toList $ Stream.unfoldEachEndBySeq "\n" Unfold.fromList input
"abc\ndef\nghi\n"

Flatten a stream generated by an effect i.e. concat the effect monad with the stream monad.

>>> concatEffect = Stream.concat . Stream.fromEffect
>>> concatEffect eff = Stream.concatMapM (\() -> eff) (Stream.fromPure ())

See also: concat, sequence

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

>>> concatMap f = Stream.concat . fmap f
>>> concatMap f = Stream.concatMapM (return . f)
>>> concatMap f = Stream.unfoldEach (Unfold.lmap f Unfold.fromStream)

See argument flipped version concatFor for more detailed documentation.

NOTE: We recommend using unfoldEach or unfoldCross instead of concatMap especially in performance critical code. unfoldEach is much faster than concatMap and matches its expressive power in terms of generating dependent inner streams, there is one important distinction though: the nesting structure when using unfoldEach is fixed statically in the code. In contrast, concatMap allows dynamic and arbitrary nesting through monadic composition. This means that deeply nested or programmatically determined levels of nesting are easier to express and compose with concatMap, though often at the cost of performance and fusion.

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.

See unfoldEach for a faster alternative.

See fairConcatFor for documentation.

Map a stream generating function on each element of a stream and concatenate the results. This is the same as the bind function of the monad instance. It is just a flipped concatMap but more convenient to use for nested use case, feels like an imperative for loop. It is in fact equivalent to concat . for.

>>> concatFor = flip Stream.concatMap

A concatenating for loop:

>>> :{
Stream.toList $
    Stream.concatFor (Stream.fromList [1,2,3]) $ \x ->
      Stream.fromPure x
:}
[1,2,3]

Use unfoldEach instead of concatFor where possible, unfoldEach is much faster due to fusion.

Nested concatenating for loops:

>>> :{
Stream.toList $
    Stream.concatFor (Stream.fromList [1,2,3]) $ \x ->
     Stream.concatFor (Stream.fromList [4,5,6]) $ \y ->
      Stream.fromPure (x, y)
:}
[(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]

If total iterations are kept the same, each increase in the nesting level increases the cost by roughly 2 times.

For significantly faster multi-level nesting, prefer using the better fusible, applicative-like crossWith over concatFor where possible.

concatFor is monad-like: it allows expressing dependencies between the outer and the inner loops of the nesting, it means that the stream generated by the inner loop is dynamically governed by the outer loop. This expressive power comes at a significant performance cost.

NOTE: We recommend using unfoldEach or unfoldCross instead of concatFor especially in performance critical code. unfoldEach is much faster than concatFor and matches its expressive power in terms of generating dependent inner streams, there is one important distinction though: the nesting structure when using unfoldEach is fixed statically in the code. In contrast, concatFor allows dynamic and arbitrary nesting through monadic composition. This means that deeply nested or programmatically determined levels of nesting are easier to express and compose with concatFor, though often at the cost of performance and fusion.

fairConcatFor is like concatFor but traverses the depth and breadth of nesting equally. Therefore, the outer and the inner loops in a nested loop get equal priority. It can be used to nest infinite streams without starving outer streams due to inner ones.

Given a stream of three streams:

1. [1,2,3]
2. [4,5,6]
3. [7,8,9]

Here, outer loop is the stream of streams and the inner loops are the individual streams. The traversal sweeps the diagonals in the above grid to give equal chance to outer and inner loops. The resulting stream is (1),(2,4),(3,5,7),(6,8),(9), diagonals are parenthesized for emphasis.

Looping

A single stream case is equivalent to concatFor:

>>> Stream.toList $ Stream.fairConcatFor (Stream.fromList [1,2]) $ \x -> Stream.fromPure x
[1,2]

Fair Nested Looping

Multiple streams nest like for loops. The result is a cross product of the streams. However, the ordering of the results of the cross product is such that each stream gets consumed equally. In other words, inner iterations of a nested loop get the same priority as the outer iterations. Inner iterations do not finish completely before the outer iterations start.

>>> :{
Stream.toList $ do
    Stream.fairConcatFor (Stream.fromList [1,2,3]) $ \x ->
     Stream.fairConcatFor (Stream.fromList [4,5,6]) $ \y ->
      Stream.fromPure (x, y)
:}
[(1,4),(1,5),(2,4),(1,6),(2,5),(3,4),(2,6),(3,5),(3,6)]

Nesting Infinite Streams

Example with infinite streams. Print all pairs in the cross product with sum less than a specified number.

>>> :{
Stream.toList
 $ Stream.takeWhile (\(x,y) -> x + y < 6)
 $ Stream.fairConcatFor (Stream.fromList [1..]) $ \x ->
    Stream.fairConcatFor (Stream.fromList [1..]) $ \y ->
     Stream.fromPure (x, y)
:}
[(1,1),(1,2),(2,1),(1,3),(2,2),(3,1),(1,4),(2,3),(3,2),(4,1)]

How the nesting works?

If we look at the cross product of [1,2,3], [4,5,6], the streams being combined using fairConcatFor are the following sequential loop iterations:

(1,4) (1,5) (1,6) -- first iteration of the outer loop
(2,4) (2,5) (2,6) -- second iteration of the outer loop
(3,4) (3,5) (3,6) -- third iteration of the outer loop

The result is a triangular or diagonal traversal of these iterations:

[(1,4),(1,5),(2,4),(1,6),(2,5),(3,4),(2,6),(3,5),(3,6)]

Non-Termination Cases

If one of the two interleaved streams does not produce an output at all and continues forever then the other stream will never get scheduled. This is because a stream is unscheduled only after it produces an output. This can lead to non-terminating programs, an example is provided below.

>>> :{
oddsIf x = Stream.fromList (if x then [1,3..] else [2,4..])
filterEven x = if even x then Stream.fromPure x else Stream.nil
:}

>>> :{
evens =
    Stream.fairConcatFor (Stream.fromList [True,False]) $ \r ->
     Stream.concatFor (oddsIf r) filterEven
:}

The evens function does not terminate because, when r is True, the nested concatFor is a non-productive infinite loop, therefore, the outer loop never gets a chance to generate the False value.

But the following refactoring of the above code works as expected:

>>> :{
mixed =
     Stream.fairConcatFor (Stream.fromList [True,False]) $ \r ->
         Stream.concatFor (oddsIf r) Stream.fromPure
:}

>>> evens = Stream.fairConcatFor mixed filterEven
>>> Stream.toList $ Stream.take 3 $ evens
[2,4,6]

This works because in mixed both the streams being interleaved are productive.

Care should be taken how you write your program, keep in mind the scheduling implications. To avoid such scheduling problems in serial interleaving, you can use fairSchedFor or concurrent scheduling i.e. parFairConcatFor. Due to concurrent scheduling the other branch will make progress even if one is an infinite loop producing nothing.

Logic Programming

Streamly provides all operations for logic programming. It provides functionality equivalent to LogicT type from the logict package. The MonadLogic operations can be implemented using the available stream operations. For example, uncons is msplit, interleave corresponds to the interleave operation of MonadLogic, fairConcatFor is the fair bind (>>-) operation. fairSchedFor is an even better alternative for fair bind, it guarantees that non-productive infinite streams cannot block progress.

Related Operations

See also "Streamly.Internal.Data.StreamK.fairConcatFor".

Like concatFor but maps an effectful function. It allows conveniently mixing monadic effects with streams.

>>> import Control.Monad.IO.Class (liftIO)
>>> :{
Stream.toList $
    Stream.concatForM (Stream.fromList [1,2,3]) $ \x -> do
      liftIO $ putStrLn (show x)
      pure $ Stream.fromPure x
:}
1
2
3
[1,2,3]

Nested concatentating for loops:

>>> :{
Stream.toList $
    Stream.concatForM (Stream.fromList [1,2,3]) $ \x -> do
      liftIO $ putStrLn (show x)
      pure $ Stream.concatForM (Stream.fromList [4,5,6]) $ \y -> do
        when (x == 1) $ liftIO $ putStrLn (show y)
        pure $ Stream.fromPure (x, y)
:}
1
4
5
6
2
3
[(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]

See fairConcatFor for documentation.

Apply a terminating Fold repeatedly on a stream and emit the results in the output stream. If the last fold is empty, it's result is not emitted. This means if the input stream is empty the result is also an empty stream. See foldManyPost for an alternate behavior which always results in a non-empty stream even if the input stream is empty.

Definition:

>>> foldMany f = Stream.parseMany (Parser.fromFold f)

Example, empty stream, omits the empty fold value:

>>> f = Fold.take 2 Fold.toList
>>> fmany = Stream.fold Fold.toList . Stream.foldMany f
>>> fmany $ Stream.fromList []
[]

Example, omits the last empty fold value:

>>> fmany $ Stream.fromList [1..4]
[[1,2],[3,4]]

Example, last fold non-empty:

>>> fmany $ Stream.fromList [1..5]
[[1,2],[3,4],[5]]

Note that using a closed fold e.g. Fold.take 0, would result in an infinite stream on a non-empty input stream.

Group the input stream into groups of n elements each and then fold each group using the provided fold function.

Definition:

>>> groupsOf n f = Stream.foldMany (Fold.take n f)

Usage:

>>> Stream.toList $ Stream.groupsOf 2 Fold.toList (Stream.enumerateFromTo 1 10)
[[1,2],[3,4],[5,6],[7,8],[9,10]]

This can be considered as an n-fold version of take where we apply take repeatedly on the leftover stream until the stream exhausts.

Apply a Parser repeatedly on a stream and emit the parsed values in the output stream.

Usage:

>>> s = Stream.fromList [1..10]
>>> parser = Parser.takeBetween 0 2 Fold.sum
>>> Stream.toList $ Stream.parseMany parser s
[Right 3,Right 7,Right 11,Right 15,Right 19]

This is the streaming equivalent of the many parse combinator.

Known Issues: When the parser fails there is no way to get the remaining stream.

Split on an infixed separator element, dropping the separator. The supplied Fold is applied on the split segments. Splits the stream on separator elements determined by the supplied predicate, separator is considered as infixed between two segments:

Definition:

Usage:

>>> splitOn p xs = Stream.fold Fold.toList $ Stream.splitSepBy_ p Fold.toList (Stream.fromList xs)
>>> splitOn (== '.') "a.b"
["a","b"]

Splitting an empty stream results in an empty stream i.e. zero splits:

>>> splitOn (== '.') ""
[]

If the stream does not contain the separator then it results in a single split:

>>> splitOn (== '.') "abc"
["abc"]

If one or both sides of the separator are missing then the empty segment on that side is folded to the default output of the fold:

>>> splitOn (== '.') "."
["",""]

>>> splitOn (== '.') ".a"
["","a"]

>>> splitOn (== '.') "a."
["a",""]

>>> splitOn (== '.') "a..b"
["a","","b"]

splitSepBy_ is an inverse of unfoldEachSepBy:

Stream.unfoldEachSepBy '.' Unfold.fromList . Stream.splitSepBy_ (== '.') Fold.toList === id

Assuming the input stream does not contain the separator:

Stream.splitSepBy_ (== '.') Fold.toList . Stream.unfoldEachSepBy '.' Unfold.fromList === id

Like splitSepBy_ but splits the stream on a sequence of elements rather than a single element. Parses a sequence of tokens separated by an infixed separator e.g. a;b;c is parsed as a, b, c. If the pattern is empty then each element is a match, thus the fold is finalized on each element.

>>> splitSepBy p xs = Stream.fold Fold.toList $ Stream.splitSepBySeq_ (Array.fromList p) Fold.toList (Stream.fromList xs)

>>> splitSepBy "" ""
[]

>>> splitSepBy "" "a...b"
["a",".",".",".","b"]

>>> splitSepBy ".." ""
[]

>>> splitSepBy ".." "a...b"
["a",".b"]

>>> splitSepBy ".." "abc"
["abc"]

>>> splitSepBy ".." ".."
["",""]

>>> splitSepBy "." ".a"
["","a"]

>>> splitSepBy "." "a."
["a",""]

Uses Rabin-Karp algorithm for substring search.

Parses a sequence of tokens suffixed by a separator e.g. a;b;c; is parsed as a;, b;, c;. If the pattern is empty the input stream is returned as it is.

Equivalent to the following:

>>> splitEndBySeq pat f = Stream.foldMany (Fold.takeEndBySeq pat f)

Usage:

>>> f p = Stream.splitEndBySeq (Array.fromList p) Fold.toList
>>> splitEndBy p xs = Stream.fold Fold.toList $ f p (Stream.fromList xs)

>>> splitEndBy "" ""
[]

>>> splitEndBy "" "a...b"
["a",".",".",".","b"]

>>> splitEndBy ".." ""
[]

>>> splitEndBy ".." "a...b"
["a..",".b"]

>>> splitEndBy ".." "abc"
["abc"]

>>> splitEndBy ".." ".."
[".."]

>>> splitEndBy "." ".a"
[".","a"]

>>> splitEndBy "." "a."
["a."]

Uses Rabin-Karp algorithm for substring search.

Like splitEndBySeq but drops the separators and returns only the tokens.

Equivalent to the following:

>>> splitEndBySeq_ pat f = Stream.foldMany (Fold.takeEndBySeq_ pat f)

Usage:

>>> f p = Stream.splitEndBySeq_ (Array.fromList p) Fold.toList
>>> splitEndBy_ p xs = Stream.fold Fold.toList $ f p (Stream.fromList xs)

>>> splitEndBy_ "" ""
[]

>>> splitEndBy_ "" "a...b"
["a",".",".",".","b"]

>>> splitEndBy_ ".." ""
[]

>>> splitEndBy_ ".." "a...b"
["a",".b"]

>>> splitEndBy_ ".." "abc"
["abc"]

>>> splitEndBy_ ".." ".."
[""]

>>> splitEndBy_ "." ".a"
["","a"]

>>> splitEndBy_ "." "a."
["a"]

Uses Rabin-Karp algorithm for substring search.

Split the stream after stripping leading, trailing, and repeated separators determined by the predicate supplied. The tokens after splitting are collected by the supplied fold. In other words, the tokens are parsed in the same way as words are parsed from whitespace separated text.

>>> f x = Stream.toList $ Stream.wordsBy (== '.') Fold.toList $ Stream.fromList x
>>> f "a.b"
["a","b"]
>>> f "a..b"
["a","b"]
>>> f ".a..b."
["a","b"]

Returns the elements of the stream in reverse order. The stream must be finite. Note that this necessarily buffers the entire stream in memory.

Definition:

>>> reverse m = Stream.concatEffect $ Stream.fold Fold.toListRev m >>= return . Stream.fromList

Returns the first stream appended with those unique elements from the second stream that are not already present in the first stream. Note that this is not a commutative operation unlike a set union, argument order matters. The behavior is similar to unionBy.

Equivalent to the following except that s2 is evaluated only once:

>>> unionBy eq s1 s2 = s1 `Stream.append` Stream.deleteFirstsBy eq s1 (Stream.ordNub s2)

Example:

>>> f s1 s2 = Stream.fold Fold.toList $ Stream.unionBy (==) (Stream.fromList s1) (Stream.fromList s2)
>>> f [1,2,2,4] [1,1,2,3,3]
[1,2,2,4,3]

First stream can be infinite, but second stream must be finite. Note that if the first stream is infinite the union means just the first stream. Thus union is useful only when both streams are finite. See sortedUnionBy where union can work on infinite streams if they are sorted.

Space: O(n)

Time: O(m x n)

Pre-release

Compare two streams for equality

Compare two streams lexicographically.

Returns True if the first stream is the same as or a prefix of the second. A stream is a prefix of itself.

>>> Stream.isPrefixOf (Stream.fromList "hello") (Stream.fromList "hello" :: Stream IO Char)
True

Returns True if the first stream is an infix of the second. A stream is considered an infix of itself.

>>> s = Stream.fromList "hello" :: Stream IO Char
>>> Stream.isInfixOf s s
True

Space: O(n) worst case where n is the length of the infix.

Pre-release

Requires Storable constraint

Returns True if all the elements of the first stream occur, in order, in the second stream. The elements do not have to occur consecutively. A stream is a subsequence of itself.

>>> Stream.isSubsequenceOf (Stream.fromList "hlo") (Stream.fromList "hello" :: Stream IO Char)
True

stripPrefix prefix input strips the prefix stream from the input stream if it is a prefix of input. Returns Nothing if the input does not start with the given prefix, stripped input otherwise. Returns Just nil when the prefix is the same as the input stream.

Space: O(1)

Run the action m b if the stream evaluation is aborted due to an exception. The exception is not caught, simply rethrown.

Observes exceptions only in the stream generation, and not in stream consumers.

Inhibits stream fusion

When evaluating a stream if an exception occurs, stream evaluation aborts and the specified exception handler is run with the exception as argument. The exception is caught and handled unless the handler decides to rethrow it. Note that exception handling is not applied to the stream returned by the exception handler.

Observes exceptions only in the stream generation, and not in stream consumers.

Inhibits stream fusion

Run the action m b before the stream yields its first element.

Same as the following but more efficient due to fusion:

>>> before action xs = Stream.concatMap (const xs) (Stream.fromEffect action)

Run the action IO b whenever the stream is evaluated to completion, or if it is garbage collected after a partial lazy evaluation.

The semantics of the action IO b are similar to the semantics of cleanup action in bracketIO.

See also afterUnsafe

Run the action IO b whenever the stream stream stops normally, aborts due to an exception or if it is garbage collected after a partial lazy evaluation.

The semantics of running the action IO b are similar to the cleanup action semantics described in bracketIO.

>>> finallyIO release stream = Stream.bracketIO (return ()) (const release) (const stream)

See also finallyIO' for stricter resource release guarantees.

See also finallyUnsafe

Inhibits stream fusion

Like finallyIO, based on bracketIO' semantics.

Like finallyIO, based on bracketIO'' semantics.

The alloc action IO b is executed with async exceptions disabled but keeping blocking operations interruptible (see mask). Uses the output b of the IO action as input to the function b -> Stream m a to generate an output stream.

b is usually a resource allocated under the IO monad, e.g. a file handle, that requires a cleanup after use. The cleanup is done using the b -> IO c action. bracketIO guarantees that the allocated resource is eventually (see details below) cleaned up even in the face of sync or async exceptions. If an exception occurs it is not caught, simply rethrown.

bracketIO only guarantees that the cleanup action runs, and it runs with async exceptions enabled. The action must ensure that it can successfully cleanup the resource in the face of sync or async exceptions.

Best case: Cleanup happens immediately in the following cases:

• the stream is consumed completely
• an exception occurs in the bracketed part of the pipeline

Worst case: In the following cases cleanup is deferred to GC.

• the bracketed stream is partially consumed and abandoned
• pipeline is aborted due to an exception outside the bracket

Use Streamly.Control.Exception.withAcquireIO for covering the entire pipeline with guaranteed cleanup at the end of bracket.

Observes exceptions only in the stream generation, and not in stream consumers.

See also: bracketUnsafe

Inhibits stream fusion

Like bracketIO but requires an AcquireIO reference in the underlying monad of the stream, and guarantees that all resources are freed before the scope of the monad level resource manager (Streamly.Control.Exception.withAcquireIO) ends. Where fusion matters, this combinator can be much faster than bracketIO as it allows stream fusion.

Best case: Cleanup happens immediately if the stream is consumed completely.

Worst case: In the following cases cleanup is guaranteed to occur at the end of the monad level bracket. However, if a GC occurs then cleanup will occur even earlier than that.

• the bracketed stream is partially consumed and abandoned
• pipeline is aborted due to an exception

This is the recommended default bracket operation.

Note: You can use acquire directly, instead of using this combinator, if you don’t need to release the resource when the stream ends. However, if you're using the stream inside another stream (like with concatMap), you usually do want to release it at the end of the stream.

Allows stream fusion

Like bracketIO, the only difference is that there is a guarantee that the resources will be freed at the end of the monad level bracket (AcquireIO).

Best case: Cleanup happens immediately in the following cases:

• the stream is consumed completely
• an exception occurs in the bracketed part of the pipeline

Worst case: In the following cases cleanup is guaranteed to occur at the end of the monad level bracket. However, if a GC occurs before that then cleanup will occur early.

• the bracketed stream is partially consumed and abandoned
• pipeline is aborted due to an exception outside the bracket

Note: Instead of using this combinator you can directly use acquire if you do not care about releasing the resource at the end of the stream and if you are not recovering from an exception using handle. You may want to care about releasing the resource at the end of a stream if you are using it in a nested manner (e.g. in concatMap).

Inhibits stream fusion

Like bracketIO but can use 3 separate cleanup actions depending on the mode of termination:

1. When the stream stops normally
2. When the stream is garbage collected
3. When the stream encounters an exception

bracketIO3 before onStop onGC onException action runs action using the result of before. If the stream stops, onStop action is executed, if the stream is abandoned onGC is executed, if the stream encounters an exception onException is executed.

The exception is not caught, it is rethrown.

Inhibits stream fusion

Pre-release

Transform the inner monad of a stream using a natural transformation.

Example, generalize the inner monad from Identity to any other:

>>> generalizeInner = Stream.morphInner (return . runIdentity)

Also known as hoist.

Lift the inner monad m of Stream m a to t m where t is a monad transformer.

Evaluate the inner monad of a stream as ReaderT.

Evaluate the inner monad of a stream as StateT and emit the resulting state and value pair after each step.

unfoldEach unfold stream uses unfold to map the input stream elements to streams and then flattens the generated streams into a single output stream.

Like concatMap but uses an Unfold for stream generation. Unlike concatMap this can fuse the Unfold code with the inner loop and therefore provide many times better performance.

>>> concatMap f = Stream.unfoldEach (Unfold.lmap f Unfold.fromStream)

Here is an example of a two level nested loop much faster than concatMap based nesting.

>>> :{
outerLoop =
  flip Stream.mapM (Stream.fromList [1,2,3]) $ \x -> do
      liftIO $ putStrLn (show x)
      return x
innerUnfold = Unfold.carry $ Unfold.lmap (const [4,5,6]) Unfold.fromList
innerLoop =
     flip Unfold.mapM innerUnfold $ \(x, y) -> do
         when (x == 1) $ liftIO $ putStrLn (show y)
         pure $ (x, y)
:}

>>> Stream.toList $ Stream.unfoldEach innerLoop outerLoop
1
4
5
6
2
3
[(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]