Streamly.Internal.Data.Stream

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

A stream is a succession of Steps. A Yield produces a single value and the next state of the stream. Stop indicates there are no more values in the stream.

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

Flatten a stream of streams to a single stream.

>>> concat = Stream.concatMap id

Pre-release

Create a singleton stream from a pure value.

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

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.

Construct a stream from a list of pure values.

Convert a direct style step encoded StreamD to a CPS encoded StreamK

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.

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 Stream.nil if the stream finished before the fold.

Fold resulting in either breaking the stream or continuation of the fold. Instead of supplying the input stream in one go we can run the fold multiple times, each time supplying the next segment of the input stream. If the fold has not yet finished it returns a fold that can be run again otherwise it returns the fold result and the residual stream.

Internal

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

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

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

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.

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

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

Pre-release

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

Convert a CPS encoded StreamK to direct style step encoded StreamD

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.

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

Like foldMany but for the Refold type. The supplied action is used as the initial value for each refold.

Internal

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.

Definitions:

>>> drain = Stream.fold Fold.drain>>> drain = Stream.foldrM (\_ xs -> xs) (return ())

Run a stream, discarding the results.

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

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

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):}

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

Same as takeWhile but with a monadic predicate.

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.

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

Compare two streams for equality

Compare two streams lexicographically.

Like foldIterateM but using the Refold type instead. This could be much more efficient due to stream fusion.

Internal

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.

See cross for a much faster fused alternative.

Time: O(m x n)

Pre-release

A newtype wrapper for the Stream type with a cross product style monad instance.

A Monad bind behaves like a for loop:

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

Nested monad binds behave like nested for loops:

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

Append a stream to a fold lazily to build an accumulator incrementally.

Example, to continue folding a list of streams on the same sum fold:

>>> streams = [Stream.fromList [1..5], Stream.fromList [6..10]]>>> f = Prelude.foldl Stream.foldAddLazy Fold.sum streams>>> Stream.fold f Stream.nil55

Apply a stream of functions to a stream of values and flatten the results.

Note that the second stream is evaluated multiple times.

>>> crossApply = Stream.crossWith id

Same as concatIterateDfs but more efficient due to stream fusion.

Example, list a directory tree using DFS:

>>> f = Unfold.either Dir.eitherReaderPaths Unfold.nil>>> input = Stream.fromPure (Left ".")>>> ls = Stream.unfoldIterateDfs f input

Pre-release

Like unfoldIterateDfs but uses breadth first style traversal.

Pre-release

Like unfoldIterateBfs but processes the children in reverse order, therefore, may be slightly faster.

Pre-release

Generate a stream from an initial state, scan and concat the stream, generate a stream again from the final state of the previous scan and repeat the process.

Traverse the stream in depth first style (DFS). Map each element in the input stream to a stream and flatten, recursively map the resulting elements as well to a stream and flatten until no more streams are generated.

Example, list a directory tree using DFS:

>>> f = either (Just . Dir.readEitherPaths) (const Nothing)>>> input = Stream.fromPure (Left ".")>>> ls = Stream.concatIterateDfs f input

This is equivalent to using concatIterateWith StreamK.append.

Pre-release

Similar to concatIterateDfs except that it traverses the stream in breadth first style (BFS). First, all the elements in the input stream are emitted, and then their traversals are emitted.

Example, list a directory tree using BFS:

>>> f = either (Just . Dir.readEitherPaths) (const Nothing)>>> input = Stream.fromPure (Left ".")>>> ls = Stream.concatIterateBfs f input

Pre-release

Same as concatIterateBfs except that the traversal of the last element on a level is emitted first and then going backwards up to the first element (reversed ordering). This may be slightly faster than concatIterateBfs.

Like foldMany but evaluates the fold even if the fold did not receive any input, therefore, always results in a non-empty output even on an empty stream (default result of the fold).

Example, empty stream:

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

Example, last fold empty:

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

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 without consuming the input.

Pre-release

Binary BFS style reduce, folds a level entirely using the supplied fold function, collecting the outputs as next level of the tree, then repeats the same process on the next level. The last elements of a previously folded level are folded first.

N-Ary BFS style iterative fold, if the input stream finished before the fold then it returns Left otherwise Right. If the fold returns Left we terminate.

Unimplemented

Like splitOnSuffix but generates a stream of (index, len) tuples marking the places where the predicate matches in the stream.

Pre-release

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]

Create a singleton stream from a pure value.

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

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

>>> Stream.toList Stream.nil[]

Construct a stream from a list of pure values.

Convert a direct style step encoded StreamD to a CPS encoded StreamK

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

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:}01[0,1,2]

Generate an infinite stream by repeating a pure value.

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

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

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

Pre-release

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

Convert a list of monadic actions to a Stream

Keep reading Storable elements from an immutable Ptr onwards.

Unsafe: The caller is responsible for safe addressing.

Pre-release

Convert a CPS encoded StreamK to direct style step encoded StreamD

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

Read bytes from an immutable Addr# until a 0 byte is encountered, the 0 byte is not included in the stream.

>>> :set -XMagicHash>>> fromByteStr# addr = Stream.takeWhile (/= 0) $ Stream.fromPtr $ Ptr addr

Unsafe: The caller is responsible for safe addressing.

Note that this is completely safe when reading from Haskell string literals because they are guaranteed to be NULL terminated:

>>> Stream.toList $ Stream.fromByteStr# "\1\2\3\0"#[1,2,3]

WARNING! O(n^2) time complexity wrt number of elements. Use the O(n) complexity StreamK.Streamly.Data.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

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]

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.

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

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

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

Construct a stream from a Foldable containing pure values:

/WARNING: O(n^2), suitable only for a small number of elements in the stream/

For floating point numbers if the increment is less than the precision then it just gets lost. Therefore we cannot always increment it correctly by just repeated addition. 9007199254740992 + 1 + 1 :: Double => 9.007199254740992e15 9007199254740992 + 2 :: Double => 9.007199254740994e15

Instead we accumulate the increment counter and compute the increment every time before adding it to the starting number.

This works for Integrals as well as floating point numbers, but enumerateFromStepIntegral is faster for integrals.

>>> enumerateFromBounded from = Stream.enumerateFromTo from maxBound

enumerateFrom for Bounded Enum types.

enumerateFromTo for Enum types not larger than Int.

enumerateFromThenTo for Enum types not larger than Int.

enumerateFromThen for Enum types not larger than Int.

Note: We convert the Enum to Int and enumerate the Int. If a type is bounded but does not have a Bounded instance then we can go on enumerating it beyond the legal values of the type, resulting in the failure of toEnum when converting back to Enum. Therefore we require a Bounded instance for this function to be safely used.

Enumerate an Integral type. enumerateFromIntegral from generates a stream whose first element is from and the successive elements are in increments of 1. The stream is bounded by the size of the Integral type.

>>> Stream.toList $ Stream.take 4 $ Stream.enumerateFromIntegral (0 :: Int)[0,1,2,3]

Enumerate an Integral type in steps. enumerateFromThenIntegral from then generates a stream whose first element is from, the second element is then and the successive elements are in increments of then - from. The stream is bounded by the size of the Integral type.

>>> Stream.toList $ Stream.take 4 $ Stream.enumerateFromThenIntegral (0 :: Int) 2[0,2,4,6]

>>> Stream.toList $ Stream.take 4 $ Stream.enumerateFromThenIntegral (0 :: Int) (-2)[0,-2,-4,-6]

Enumerate an Integral type up to a given limit. enumerateFromToIntegral from to generates a finite stream whose first element is from and successive elements are in increments of 1 up to to.

>>> Stream.toList $ Stream.enumerateFromToIntegral 0 4[0,1,2,3,4]

Enumerate an Integral type in steps up to a given limit. enumerateFromThenToIntegral from then to generates a finite stream whose first element is from, the second element is then and the successive elements are in increments of then - from up to to.

>>> Stream.toList $ Stream.enumerateFromThenToIntegral 0 2 6[0,2,4,6]

>>> Stream.toList $ Stream.enumerateFromThenToIntegral 0 (-2) (-6)[0,-2,-4,-6]

enumerateFromStepIntegral from step generates an infinite stream whose first element is from and the successive elements are in increments of step.

CAUTION: This function is not safe for finite integral types. It does not check for overflow, underflow or bounds.

>>> Stream.toList $ Stream.take 4 $ Stream.enumerateFromStepIntegral 0 2[0,2,4,6]

>>> Stream.toList $ Stream.take 3 $ Stream.enumerateFromStepIntegral 0 (-2)[0,-2,-4]

Numerically stable enumeration from a Fractional number in steps of size 1. enumerateFromFractional from generates a stream whose first element is from and the successive elements are in increments of 1. No overflow or underflow checks are performed.

This is the equivalent to enumFrom for Fractional types. For example:

>>> Stream.toList $ Stream.take 4 $ Stream.enumerateFromFractional 1.1[1.1,2.1,3.1,4.1]

Numerically stable enumeration from a Fractional number to a given limit. enumerateFromToFractional from to generates a finite stream whose first element is from and successive elements are in increments of 1 up to to.

This is the equivalent of enumFromTo for Fractional types. For example:

>>> Stream.toList $ Stream.enumerateFromToFractional 1.1 4[1.1,2.1,3.1,4.1]

>>> Stream.toList $ Stream.enumerateFromToFractional 1.1 4.6[1.1,2.1,3.1,4.1,5.1]

Notice that the last element is equal to the specified to value after rounding to the nearest integer.

Numerically stable enumeration from a Fractional number in steps. enumerateFromThenFractional from then generates a stream whose first element is from, the second element is then and the successive elements are in increments of then - from. No overflow or underflow checks are performed.

This is the equivalent of enumFromThen for Fractional types. For example:

>>> Stream.toList $ Stream.take 4 $ Stream.enumerateFromThenFractional 1.1 2.1[1.1,2.1,3.1,4.1]

>>> Stream.toList $ Stream.take 4 $ Stream.enumerateFromThenFractional 1.1 (-2.1)[1.1,-2.1,-5.300000000000001,-8.500000000000002]

Numerically stable enumeration from a Fractional number in steps up to a given limit. enumerateFromThenToFractional from then to generates a finite stream whose first element is from, the second element is then and the successive elements are in increments of then - from up to to.

This is the equivalent of enumFromThenTo for Fractional types. For example:

>>> Stream.toList $ Stream.enumerateFromThenToFractional 0.1 2 6[0.1,2.0,3.9,5.799999999999999]

>>> Stream.toList $ Stream.enumerateFromThenToFractional 0.1 (-2) (-6)[0.1,-2.0,-4.1000000000000005,-6.200000000000001]

times returns a stream of time value tuples with clock of 10 ms granularity. The first component of the tuple is an absolute time reference (epoch) denoting the start of the stream and the second component is a time relative to the reference.

>>> f = Fold.drainMapM (\x -> print x >> threadDelay 1000000)>>> Stream.fold f $ Stream.take 3 $ Stream.times(AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))(AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))(AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))

Note: This API is not safe on 32-bit machines.

Pre-release

timesWith g returns a stream of time value tuples. The first component of the tuple is an absolute time reference (epoch) denoting the start of the stream and the second component is a time relative to the reference.

The argument g specifies the granularity of the relative time in seconds. A lower granularity clock gives higher precision but is more expensive in terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.

>>> import Control.Concurrent (threadDelay)>>> f = Fold.drainMapM (\x -> print x >> threadDelay 1000000)>>> Stream.fold f $ Stream.take 3 $ Stream.timesWith 0.01(AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))(AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))(AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))

Note: This API is not safe on 32-bit machines.

Pre-release

absTimes returns a stream of absolute timestamps using a clock of 10 ms granularity.

>>> f = Fold.drainMapM print>>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimesAbsTime (TimeSpec {sec = ..., nsec = ...})AbsTime (TimeSpec {sec = ..., nsec = ...})AbsTime (TimeSpec {sec = ..., nsec = ...})

Note: This API is not safe on 32-bit machines.

Pre-release

absTimesWith g returns a stream of absolute timestamps using a clock of granularity g specified in seconds. A low granularity clock is more expensive in terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.

>>> f = Fold.drainMapM print>>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimesWith 0.01AbsTime (TimeSpec {sec = ..., nsec = ...})AbsTime (TimeSpec {sec = ..., nsec = ...})AbsTime (TimeSpec {sec = ..., nsec = ...})

Note: This API is not safe on 32-bit machines.

Pre-release

relTimes returns a stream of relative time values starting from 0, using a clock of granularity 10 ms.

>>> f = Fold.drainMapM print>>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesRelTime64 (NanoSecond64 ...)RelTime64 (NanoSecond64 ...)RelTime64 (NanoSecond64 ...)

Note: This API is not safe on 32-bit machines.

Pre-release

relTimesWith g returns a stream of relative time values starting from 0, using a clock of granularity g specified in seconds. A low granularity clock is more expensive in terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.

>>> f = Fold.drainMapM print>>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01RelTime64 (NanoSecond64 ...)RelTime64 (NanoSecond64 ...)RelTime64 (NanoSecond64 ...)

Note: This API is not safe on 32-bit machines.

Pre-release

durations g returns a stream of relative time values measuring the time elapsed since the immediate predecessor element of the stream was generated. The first element of the stream is always 0. durations uses a clock of granularity g specified in seconds. A low granularity clock is more expensive in terms of CPU usage. The minimum granularity is 1 millisecond. Durations lower than 1 ms will be 0.

Note: This API is not safe on 32-bit machines.

Unimplemented

Generate a singleton event at or after the specified absolute time. Note that this is different from a threadDelay, a threadDelay starts from the time when the action is evaluated, whereas if we use AbsTime based timeout it will immediately expire if the action is evaluated too late.

Unimplemented

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

Construct a stream from a Foldable containing pure values:

/WARNING: O(n^2), suitable only for a small number of elements in the stream/

Take n Storable elements starting from an immutable Ptr onwards.

>>> fromPtrN n = Stream.take n . Stream.fromPtr

Unsafe: The caller is responsible for safe addressing.

Pre-release

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.

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

Run a Parse over a stream.

Parse a stream using the supplied Parser.

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

Execute a monadic action for each element of the Stream

Definitions:

>>> drain = Stream.fold Fold.drain>>> drain = Stream.foldrM (\_ xs -> xs) (return ())

Run a stream, discarding the results.

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):}

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

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 a Parse over a stream and return rest of the Stream.

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 the first stream is a suffix of the second. A stream is considered a suffix of itself.

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

Space: O(n), buffers entire input stream and the suffix.

Pre-release

Suboptimal - Help wanted.

Much faster than isSuffixOf.

Drops the given suffix from a stream. Returns Nothing if the stream does not end with the given suffix. Returns Just nil when the suffix is the same as the stream.

It may be more efficient to convert the stream to an Array and use stripSuffix on that especially if the elements have a Storable or Prim instance.

Space: O(n), buffers the entire input stream as well as the suffix

Pre-release

Much faster than stripSuffix.

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 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 gbracket but with following differences:

• alloc action m c runs with async exceptions enabled

• cleanup action c -> m d won't run if the stream is garbage collected after partial evaluation.

Inhibits stream fusion

Pre-release

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

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

Run the alloc action m c with async exceptions disabled but keeping blocking operations interruptible (see mask). Use the output c as input to c -> Stream m b to generate an output stream. When generating the stream use the supplied try operation forall s. m s -> m (Either e s) to catch synchronous exceptions. If an exception occurs run the exception handler c -> e -> Stream m b -> m (Stream m b). Note that gbracket does not rethrow the exception, it has to be done by the exception handler if desired.

The cleanup action c -> m d, runs whenever the stream ends normally, due to a sync or async exception or if it gets garbage collected after a partial lazy evaluation. See bracket for the semantics of the cleanup action.

gbracket can express all other exception handling combinators.

Inhibits stream fusion

Pre-release

Like after, with following differences:

• action m b won't run if the stream is garbage collected after partial evaluation.

• Monad m does not require any other constraints.

• has slightly better performance than after.

Same as the following, but with stream fusion:

>>> afterUnsafe action xs = xs <> Stream.nilM action

Pre-release

Like bracket but with following differences:

• alloc action m b runs with async exceptions enabled

• cleanup action b -> m c won't run if the stream is garbage collected after partial evaluation.

• has slightly better performance than bracketIO.

Inhibits stream fusion

Pre-release

Like finally with following differences:

• action m b won't run if the stream is garbage collected after partial evaluation.

• has slightly better performance than finallyIO.

Inhibits stream fusion

Pre-release

Like handle but the exception handler is also provided with the stream that generated the exception as input. The exception handler can thus re-evaluate the stream to retry the action that failed. The exception handler can again call ghandle on it to retry the action multiple times.

This is highly experimental. In a stream of actions we can map the stream with a retry combinator to retry each action on failure.

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.

Generalize the inner monad of the stream from Identity to any monad.

Definition:

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

Lift the inner monad m of a stream Stream m a to t m using the supplied lift function.

Evaluate the inner monad of a stream using the supplied runner function.

Evaluate the inner monad of a stream using the supplied stateful runner function and the initial state. The state returned by an invocation of the runner is supplied as input state to the next invocation.

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.

Lazy left fold to a transformer monad.

Right fold to a transformer monad. This is the most general right fold function. foldrS is a special case of foldrT, however foldrS implementation can be more efficient:

>>> foldrS = Stream.foldrT

>>> step f x xs = lift $ f x (runIdentityT xs)>>> foldrM f z s = runIdentityT $ Stream.foldrT (step f) (lift z) s

foldrT can be used to translate streamly streams to other transformer monads e.g. to a different streaming type.

Pre-release

Run a stream transformation using a given environment.

Evaluate the inner monad of a stream as StateT.

>>> evalStateT s = fmap snd . Stream.runStateT s

Run a stateful (StateT) stream transformation using a given state.

>>> usingStateT s f = Stream.evalStateT s . f . Stream.liftInner

See also: scan

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

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.Streamly.Data.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]

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.

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.

Schedule the execution of two streams in a fair round-robin manner, executing each stream once, alternately. Execution of a stream may not necessarily result in an output, a stream may choose to Skip producing an element until later giving the other stream a chance to run. Therefore, this combinator fairly interleaves the execution of two streams rather than fairly interleaving the output of the two streams. This can be useful in co-operative multitasking without using explicit threads. This can be used as an alternative to async.

Do not use dynamically.

Pre-release

Unfold the elements of a stream, intersperse the given element between the unfolded streams and then concat them into a single stream.

>>> unwords = Stream.interpose ' '

Pre-release

Unfold the elements of a stream, append the given element after each unfolded stream and then concat them into a single stream.

>>> unlines = Stream.interposeSuffix '\n'

Pre-release

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"

Like foldMany but for the Refold type. The supplied action is used as the initial value for each refold.

Internal

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.

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

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"

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 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 Streamly.Data.Parser.many parse combinator.

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

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.