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.

>>> 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 by repeating a pure value.

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

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

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]

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.

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

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

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 monadic fold.

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

See also: usingStateT

Postscan a stream using the given monadic fold.

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 = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
>>> s = Stream.enumerateFromTo 1.0 100.0
>>> :{
 Stream.fold Fold.toList
  $ fmap (fromJust . fst)
  $ Stream.takeWhile (\(_,x) -> x <= 10)
  $ Stream.postscan (Fold.tee Fold.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 = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
>>> indexed = Stream.postscan 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]

Insert an effect and its output before consuming an element of a stream except the first one.

>>> input = Stream.fromList "hello"
>>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') input
h.,e.,l.,l.,o"h,e,l,l,o"

Be careful about the order of effects. In the above example we used trace after the intersperse, if we use it before the intersperse the output would be he.l.l.o."h,e,l,l,o".

>>> Stream.fold Fold.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar input
he.l.l.o."h,e,l,l,o"

Insert a side effect before consuming an element of a stream except the first one.

>>> input = Stream.fromList "hello"
>>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
h.e.l.l.o

Pre-release

Insert a pure value between successive elements of a stream.

>>> input = Stream.fromList "hello"
>>> Stream.fold Fold.toList $ Stream.intersperse ',' input
"h,e,l,l,o"

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.scanMaybe (Fold.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

Use a filtering fold on a stream.

>>> scanMaybe f = Stream.catMaybes . Stream.postscan f

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. When one stream stops the rest of the other stream is used in the output stream.

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:

>>> :{
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.

unfoldMany 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.

intersperse followed by unfold and concat.

>>> intercalate u a = Stream.unfoldMany u . Stream.intersperse a
>>> intersperse = Stream.intercalate Unfold.identity
>>> unwords = Stream.intercalate Unfold.fromList " "

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

intersperseMSuffix followed by unfold and concat.

>>> intercalateSuffix u a = Stream.unfoldMany u . Stream.intersperseMSuffix a
>>> intersperseMSuffix = Stream.intercalateSuffix Unfold.identity
>>> unlines = Stream.intercalateSuffix Unfold.fromList "\n"

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

Given a stream value in the underlying monad, lift and join the underlying 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.concatMapM (return . f)
>>> concatMap f = Stream.concat . fmap f
>>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)

See unfoldMany for a fusible alternative.

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 unfoldMany for a fusible alternative.

Apply a Fold repeatedly on a stream and emit the results in the output stream.

Definition:

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

Example, empty stream:

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

Example, last fold empty:

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

Example, last fold non-empty:

>>> fmany $ Stream.fromList [1..5]
[3,7,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.

groupsOf n f = foldMany (FL.take n f)

>>> Stream.toList $ Stream.groupsOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
[3,7,11,15,19]

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.

Example:

>>> s = Stream.fromList [1..10]
>>> parser = Parser.takeBetween 0 2 Fold.sum
>>> Stream.fold Fold.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:

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

An empty stream is folded to the default value of the fold:

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

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"]

splitOn is an inverse of intercalating single element:

Stream.intercalate (Stream.fromPure '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id

Assuming the input stream does not contain the separator:

Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.fromPure '.') Unfold.fromList === id

Split the stream after stripping leading, trailing, and repeated separators as per the fold supplied. Therefore, ".a..b." with . as the separator would be parsed as ["a","b"]. In other words, its like parsing words from whitespace separated text.

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

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 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.nilM action <> xs
>>> 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.bracketIO (return ()) (const release)

See also finallyUnsafe

Inhibits stream fusion

Run the alloc action IO b with async exceptions disabled but keeping blocking operations interruptible (see mask). Use 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 under the IO monad, e.g. a file handle, that requires a cleanup after use. The cleanup action b -> IO c, runs whenever (1) the stream ends normally, (2) due to a sync or async exception or, (3) if it gets garbage collected after a partial lazy evaluation. The exception is not caught, it is 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.

When the stream ends normally or on a sync exception, cleanup action runs immediately in the current thread context, whereas in other cases it runs in the GC context, therefore, cleanup may be delayed until the GC gets to run. An example where GC based cleanup happens is when a stream is being folded but the fold terminates without draining the entire stream or if the consumer of the stream encounters an exception.

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

See also: bracketUnsafe

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.

chunksOf n stream groups the elements in the input stream into arrays of n elements each.

Same as the following but may be more efficient:

>>> chunksOf n = Stream.foldMany (Array.writeN n)

Pre-release