Streamly.Data.Stream

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

>>> 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 sabc

>>> 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 sabc

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 s12

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) s12

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 input123

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"

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 xinnerUnfold = Unfold.carry $ Unfold.lmap (const [4,5,6]) Unfold.fromListinnerLoop =     flip Unfold.mapM innerUnfold $ \(x, y) -> do         when (x == 1) $ liftIO $ putStrLn (show y)         pure $ (x, y):}

>>> Stream.toList $ Stream.unfoldEach innerLoop outerLoop145623[(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:}123[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):}145623[(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 Streamly.Data.Parser.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 sTrue

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.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 Streamly.Control.Exception.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.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 Streamly.Control.Exception.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 (Streamly.Control.Exception.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 Streamly.Control.Exception.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 xinnerUnfold = Unfold.carry $ Unfold.lmap (const [4,5,6]) Unfold.fromListinnerLoop =     flip Unfold.mapM innerUnfold $ \(x, y) -> do         when (x == 1) $ liftIO $ putStrLn (show y)         pure $ (x, y):}

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