streamly-0.6.1: Beautiful Streaming, Concurrent and Reactive Composition

Streamly.Prelude

Description

This module is designed to be imported qualified:

import qualified Streamly.Prelude as S


Functions with the suffix M are general functions that work on monadic arguments. The corresponding functions without the suffix M work on pure arguments and can in general be derived from their monadic versions but are provided for convenience and for consistency with other pure APIs in the base package.

In many cases, short definitions of the combinators are provided in the documentation for illustration. The actual implementation may differ for performance reasons.

Functions having a MonadAsync constraint work concurrently when used with appropriate stream type combinator. Please be careful to not use parallely with infinite streams.

Deconstruction and folds accept a SerialT type instead of a polymorphic type to ensure that streams always have a concrete monomorphic type by default, reducing type errors. In case you want to use any other type of stream you can use one of the type combinators provided in the Streamly module to convert the stream type.

Synopsis

# Construction

## Primitives

Primitives to construct a stream from pure values or monadic actions. All other stream construction and generation combinators described later can be expressed in terms of these primitives. However, the special versions provided in this module can be much more efficient in most cases. Users can create custom combinators using these primitives.

nil :: IsStream t => t m a Source #

An empty stream.

> toList nil
[]


Since: 0.1.0

cons :: IsStream t => a -> t m a -> t m a infixr 5 Source #

Construct a stream by adding a pure value at the head of an existing stream. For serial streams this is the same as (return a) consM r but more efficient. For concurrent streams this is not concurrent whereas consM is concurrent. For example:

> toList $1 cons 2 cons 3 cons nil [1,2,3]  Since: 0.1.0 (.:) :: IsStream t => a -> t m a -> t m a infixr 5 Source # Operator equivalent of cons. > toList$ 1 .: 2 .: 3 .: nil
[1,2,3]


Since: 0.1.1

consM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a infixr 5 Source #

Constructs a stream by adding a monadic action at the head of an existing stream. For example:

> toList $getLine consM getLine consM nil hello world ["hello","world"]  Concurrent (do not use parallely to construct infinite streams) Since: 0.2.0 (|:) :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a infixr 5 Source # Operator equivalent of consM. We can read it as "parallel colon" to remember that | comes before :. > toList$ getLine |: getLine |: nil
hello
world
["hello","world"]

let delay = threadDelay 1000000 >> print 1
runStream $serially$ delay |: delay |: delay |: nil
runStream $parallely$ delay |: delay |: delay |: nil


Concurrent (do not use parallely to construct infinite streams)

Since: 0.2.0

## From Values

Generate a monadic stream from a seed value or values.

yield :: IsStream t => a -> t m a Source #

yield a = a cons nil


Create a singleton stream from a pure value.

The following holds in monadic streams, but not in Zip streams:

yield = pure
yield = yieldM . pure


In Zip applicative streams yield is not the same as pure because in that case pure is equivalent to repeat instead. yield and pure are equally efficient, in other cases yield may be slightly more efficient than the other equivalent definitions.

Since: 0.4.0

yieldM :: (Monad m, IsStream t) => m a -> t m a Source #

yieldM m = m consM nil


Create a singleton stream from a monadic action.

> toList $yieldM getLine hello ["hello"]  Since: 0.4.0 repeat :: IsStream t => a -> t m a Source # repeatM = fix . cons repeatM = cycle1 . yield  Generate an infinite stream by repeating a pure value. Since: 0.4.0 repeatM :: (IsStream t, MonadAsync m) => m a -> t m a Source # repeatM = fix . consM repeatM = cycle1 . yieldM  Generate a stream by repeatedly executing a monadic action forever. runStream$ serially $S.take 10$ S.repeatM $(threadDelay 1000000 >> print 1) runStream$ asyncly  $S.take 10$ S.repeatM $(threadDelay 1000000 >> print 1)  Concurrent, infinite (do not use with parallely) Since: 0.2.0 replicate :: (IsStream t, Monad m) => Int -> a -> t m a Source # replicate = take n . repeat  Generate a stream of length n by repeating a value n times. Since: 0.6.0 replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a Source # replicateM = take n . repeatM  Generate a stream by performing a monadic action n times. Same as: runStream$ serially $S.replicateM 10$ (threadDelay 1000000 >> print 1)
runStream $asyncly$ S.replicateM 10 $(threadDelay 1000000 >> print 1)  Concurrent Since: 0.1.1 ## Enumeration We can use the Enum type class to enumerate a type producing a list and then convert it to a stream: fromList$ enumFromThen from then


However, this is not particularly efficient. The Enumerable type class provides corresponding functions that generate a stream instead of a list, efficiently.

class Enum a => Enumerable a where Source #

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.Enumeration module to define new instances.

Since: 0.6.0

Methods

enumerateFrom :: (IsStream t, Monad m) => a -> t m a Source #

enumerateFrom from generates a stream starting with the element from, enumerating up to maxBound when the type is Bounded or generating an infinite stream when the type is not Bounded.

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


For Fractional types, enumeration is numerically stable. However, no overflow or underflow checks are performed.

> S.toList $S.take 4$ S.enumerateFrom 1.1
[1.1,2.1,3.1,4.1]


Since: 0.6.0

enumerateFromTo :: (IsStream t, Monad m) => a -> a -> t m a Source #

Generate a finite stream starting with the element from, enumerating the type up to the value to. If to is smaller than from then an empty stream is returned.

> S.toList $S.enumerateFromTo 0 4 [0,1,2,3,4]  For Fractional types, the last element is equal to the specified to value after rounding to the nearest integral value. > S.toList$ S.enumerateFromTo 1.1 4
[1.1,2.1,3.1,4.1]
> S.toList $S.enumerateFromTo 1.1 4.6 [1.1,2.1,3.1,4.1,5.1]  Since: 0.6.0 enumerateFromThen :: (IsStream t, Monad m) => a -> a -> t m a Source # enumerateFromThen 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. Enumeration can occur downwards or upwards depending on whether then comes before or after from. For Bounded types the stream ends when maxBound is reached, for unbounded types it keeps enumerating infinitely. > S.toList$ S.take 4 $S.enumerateFromThen 0 2 [0,2,4,6] > S.toList$ S.take 4 $S.enumerateFromThen 0 (-2) [0,-2,-4,-6]  Since: 0.6.0 enumerateFromThenTo :: (IsStream t, Monad m) => a -> a -> a -> t m a Source # enumerateFromThenTo 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. Enumeration can occur downwards or upwards depending on whether then comes before or after from. > S.toList$ S.enumerateFromThenTo 0 2 6
[0,2,4,6]
> S.toList $S.enumerateFromThenTo 0 (-2) (-6) [0,-2,-4,-6]  Since: 0.6.0 Instances  Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Bool -> t m Bool Source #enumerateFromTo :: (IsStream t, Monad m) => Bool -> Bool -> t m Bool Source #enumerateFromThen :: (IsStream t, Monad m) => Bool -> Bool -> t m Bool Source #enumerateFromThenTo :: (IsStream t, Monad m) => Bool -> Bool -> Bool -> t m Bool Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Char -> t m Char Source #enumerateFromTo :: (IsStream t, Monad m) => Char -> Char -> t m Char Source #enumerateFromThen :: (IsStream t, Monad m) => Char -> Char -> t m Char Source #enumerateFromThenTo :: (IsStream t, Monad m) => Char -> Char -> Char -> t m Char Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Double -> t m Double Source #enumerateFromTo :: (IsStream t, Monad m) => Double -> Double -> t m Double Source #enumerateFromThen :: (IsStream t, Monad m) => Double -> Double -> t m Double Source #enumerateFromThenTo :: (IsStream t, Monad m) => Double -> Double -> Double -> t m Double Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Float -> t m Float Source #enumerateFromTo :: (IsStream t, Monad m) => Float -> Float -> t m Float Source #enumerateFromThen :: (IsStream t, Monad m) => Float -> Float -> t m Float Source #enumerateFromThenTo :: (IsStream t, Monad m) => Float -> Float -> Float -> t m Float Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Int -> t m Int Source #enumerateFromTo :: (IsStream t, Monad m) => Int -> Int -> t m Int Source #enumerateFromThen :: (IsStream t, Monad m) => Int -> Int -> t m Int Source #enumerateFromThenTo :: (IsStream t, Monad m) => Int -> Int -> Int -> t m Int Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Int8 -> t m Int8 Source #enumerateFromTo :: (IsStream t, Monad m) => Int8 -> Int8 -> t m Int8 Source #enumerateFromThen :: (IsStream t, Monad m) => Int8 -> Int8 -> t m Int8 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Int8 -> Int8 -> Int8 -> t m Int8 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Int16 -> t m Int16 Source #enumerateFromTo :: (IsStream t, Monad m) => Int16 -> Int16 -> t m Int16 Source #enumerateFromThen :: (IsStream t, Monad m) => Int16 -> Int16 -> t m Int16 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Int16 -> Int16 -> Int16 -> t m Int16 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Int32 -> t m Int32 Source #enumerateFromTo :: (IsStream t, Monad m) => Int32 -> Int32 -> t m Int32 Source #enumerateFromThen :: (IsStream t, Monad m) => Int32 -> Int32 -> t m Int32 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Int32 -> Int32 -> Int32 -> t m Int32 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Int64 -> t m Int64 Source #enumerateFromTo :: (IsStream t, Monad m) => Int64 -> Int64 -> t m Int64 Source #enumerateFromThen :: (IsStream t, Monad m) => Int64 -> Int64 -> t m Int64 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Int64 -> Int64 -> Int64 -> t m Int64 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Integer -> t m Integer Source #enumerateFromTo :: (IsStream t, Monad m) => Integer -> Integer -> t m Integer Source #enumerateFromThen :: (IsStream t, Monad m) => Integer -> Integer -> t m Integer Source #enumerateFromThenTo :: (IsStream t, Monad m) => Integer -> Integer -> Integer -> t m Integer Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Natural -> t m Natural Source #enumerateFromTo :: (IsStream t, Monad m) => Natural -> Natural -> t m Natural Source #enumerateFromThen :: (IsStream t, Monad m) => Natural -> Natural -> t m Natural Source #enumerateFromThenTo :: (IsStream t, Monad m) => Natural -> Natural -> Natural -> t m Natural Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Ordering -> t m Ordering Source #enumerateFromTo :: (IsStream t, Monad m) => Ordering -> Ordering -> t m Ordering Source #enumerateFromThen :: (IsStream t, Monad m) => Ordering -> Ordering -> t m Ordering Source #enumerateFromThenTo :: (IsStream t, Monad m) => Ordering -> Ordering -> Ordering -> t m Ordering Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Word -> t m Word Source #enumerateFromTo :: (IsStream t, Monad m) => Word -> Word -> t m Word Source #enumerateFromThen :: (IsStream t, Monad m) => Word -> Word -> t m Word Source #enumerateFromThenTo :: (IsStream t, Monad m) => Word -> Word -> Word -> t m Word Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Word8 -> t m Word8 Source #enumerateFromTo :: (IsStream t, Monad m) => Word8 -> Word8 -> t m Word8 Source #enumerateFromThen :: (IsStream t, Monad m) => Word8 -> Word8 -> t m Word8 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Word8 -> Word8 -> Word8 -> t m Word8 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Word16 -> t m Word16 Source #enumerateFromTo :: (IsStream t, Monad m) => Word16 -> Word16 -> t m Word16 Source #enumerateFromThen :: (IsStream t, Monad m) => Word16 -> Word16 -> t m Word16 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Word16 -> Word16 -> Word16 -> t m Word16 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Word32 -> t m Word32 Source #enumerateFromTo :: (IsStream t, Monad m) => Word32 -> Word32 -> t m Word32 Source #enumerateFromThen :: (IsStream t, Monad m) => Word32 -> Word32 -> t m Word32 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Word32 -> Word32 -> Word32 -> t m Word32 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Word64 -> t m Word64 Source #enumerateFromTo :: (IsStream t, Monad m) => Word64 -> Word64 -> t m Word64 Source #enumerateFromThen :: (IsStream t, Monad m) => Word64 -> Word64 -> t m Word64 Source #enumerateFromThenTo :: (IsStream t, Monad m) => Word64 -> Word64 -> Word64 -> t m Word64 Source # Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => () -> t m () Source #enumerateFromTo :: (IsStream t, Monad m) => () -> () -> t m () Source #enumerateFromThen :: (IsStream t, Monad m) => () -> () -> t m () Source #enumerateFromThenTo :: (IsStream t, Monad m) => () -> () -> () -> t m () Source # Integral a => Enumerable (Ratio a) Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Ratio a -> t m (Ratio a) Source #enumerateFromTo :: (IsStream t, Monad m) => Ratio a -> Ratio a -> t m (Ratio a) Source #enumerateFromThen :: (IsStream t, Monad m) => Ratio a -> Ratio a -> t m (Ratio a) Source #enumerateFromThenTo :: (IsStream t, Monad m) => Ratio a -> Ratio a -> Ratio a -> t m (Ratio a) Source # HasResolution a => Enumerable (Fixed a) Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Fixed a -> t m (Fixed a) Source #enumerateFromTo :: (IsStream t, Monad m) => Fixed a -> Fixed a -> t m (Fixed a) Source #enumerateFromThen :: (IsStream t, Monad m) => Fixed a -> Fixed a -> t m (Fixed a) Source #enumerateFromThenTo :: (IsStream t, Monad m) => Fixed a -> Fixed a -> Fixed a -> t m (Fixed a) Source # Enumerable a => Enumerable (Identity a) Source # Instance detailsDefined in Streamly.Enumeration MethodsenumerateFrom :: (IsStream t, Monad m) => Identity a -> t m (Identity a) Source #enumerateFromTo :: (IsStream t, Monad m) => Identity a -> Identity a -> t m (Identity a) Source #enumerateFromThen :: (IsStream t, Monad m) => Identity a -> Identity a -> t m (Identity a) Source #enumerateFromThenTo :: (IsStream t, Monad m) => Identity a -> Identity a -> Identity a -> t m (Identity a) Source # enumerate :: (IsStream t, Monad m, Bounded a, Enumerable a) => t m a Source # enumerate = enumerateFrom minBound Enumerate a Bounded type from its minBound to maxBound Since: 0.6.0 enumerateTo :: (IsStream t, Monad m, Bounded a, Enumerable a) => a -> t m a Source # enumerateTo = enumerateFromTo minBound Enumerate a Bounded type from its minBound to specified value. Since: 0.6.0 ## From Generators Generate a monadic stream from a seed value and a generator function. unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a Source # unfoldr step s = case step s of Nothing -> nil Just (a, b) -> a cons unfoldr step b  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 > 3 then Nothing else Just (b, b + 1) in toList$ unfoldr f 0

[0,1,2,3]


Since: 0.1.0

unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a Source #

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 > 3
then return Nothing
else print b >> return (Just (b, b + 1))
in runStream $unfoldrM f 0   0 1 2 3  When run concurrently, the next unfold step can run concurrently with the processing of the output of the previous step. Note that more than one step cannot run concurrently as the next step depends on the output of the previous step. (asyncly$ S.unfoldrM (\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
& S.foldlM' (\_ a -> threadDelay 1000000 >> print a) ()


Concurrent

Since: 0.1.0

iterate :: IsStream t => (a -> a) -> a -> t m a Source #

iterate f x = x cons iterate f x


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

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


Since: 0.1.2

iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a Source #

iterateM f m = m consM iterateM f m


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.

When run concurrently, the next iteration can run concurrently with the processing of the previous iteration. Note that more than one iteration cannot run concurrently as the next iteration depends on the output of the previous iteration.

runStream $serially$ S.take 10 $S.iterateM (\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0 runStream$ asyncly  $S.take 10$ S.iterateM
(\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0


Concurrent

Since: 0.1.2

fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a Source #

fromIndices f = let g i = f i cons g (i + 1) in g 0


Generate an infinite stream, whose values are the output of a function f applied on the corresponding index. Index starts at 0.

> S.toList $S.take 5$ S.fromIndices id
[0,1,2,3,4]


Since: 0.6.0

fromIndicesM :: (IsStream t, Monad m) => (Int -> m a) -> t m a Source #

fromIndicesM f = let g i = f i consM g (i + 1) in g 0


Generate an infinite stream, whose values are the output of a monadic function f applied on the corresponding index. Index starts at 0.

Since: 0.6.0

## From Containers

Convert an input structure, container or source into a stream. All of these can be expressed in terms of primitives.

fromList :: (Monad m, IsStream t) => [a] -> t m a Source #

fromList = foldr cons nil


Construct a stream from a list of pure values. This is more efficient than fromFoldable for serial streams.

Since: 0.4.0

fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a Source #

fromListM = foldr consM nil


Construct a stream from a list of monadic actions. This is more efficient than fromFoldableM for serial streams.

Since: 0.4.0

fromFoldable :: (IsStream t, Foldable f) => f a -> t m a Source #

fromFoldable = foldr cons nil


Construct a stream from a Foldable containing pure values:

Since: 0.2.0

fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a Source #

fromFoldableM = foldr consM nil


Construct a stream from a Foldable containing monadic actions.

runStream $serially$ S.fromFoldableM $replicateM 10 (threadDelay 1000000 >> print 1) runStream$ asyncly  $S.fromFoldableM$ replicateM 10 (threadDelay 1000000 >> print 1)


Concurrent (do not use with parallely on infinite containers)

Since: 0.3.0

## From External Containers

fromHandle :: (IsStream t, MonadIO m) => Handle -> t m String Source #

Read lines from an IO Handle into a stream of Strings.

Since: 0.1.0

# Elimination

## Primitives

It is easy to express all the folds in terms of the uncons primitive, however the specific implementations provided later are generally more efficient. Folds are inherently serial as each step needs to use the result of the previous step.

uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a)) Source #

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.

Since: 0.1.0

## General Folds

Right and left folds. As a simple rule, always use lazy right fold for construction and strict left fold for reduction. By construction we mean using a constructor as the outermost operation in the fold function, by reduction we mean using a function as the outermost operation in the fold function.

Right Fold Left Fold
Construction consumes input lazily and streams it in FIFO order Construction consumes all input, and constructs in reverse (LIFO) order
Reduction ends up buffering all input before it can be reduced Strict reduction works incrementally, without buffering.

Almost always, we need lazy construction and strict reduction, therefore, strict foldr and lazy foldl are rarely useful. If needed, strict foldr and lazy foldl can be expressed in terms of the available versions. For example, a lazy foldl can be replaced by a strict foldl to reverse the structure followed by a foldr.

The following equations may help understand the relation between the two folds for lists:

foldr f z xs = foldl (flip f) z (reverse xs)
foldl f z xs = foldr (flip f) z (reverse xs)


More generally:

foldl f z xs = foldr g id xs z where g x k = k . flip f x
foldr f z xs = foldl g id xs z where g k x = k . f x


foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b Source #

Lazy right associative fold.

For lists a foldr looks like:

foldr f z []     = z
foldr f z (x:xs) = x f foldr f z xs


The recursive expression is the second argument of the fold step f. Therefore, the evaluation of the recursive call depends on f. It can terminate recursion by not inspecting the second argument based on a condition. When expanded fully, it results in the following right associated expression:

foldr f z xs == x1 f (x2 f ...(xn f z))


When f is a constructor, we can see that the first deconstruction of this expression would be x1 on the left and the recursive expression on the right. Therefore, we can deconstruct it to access the input elements in the first-in-first-out (FIFO) order and consume the reconstructed structure lazily. The recursive expression on the right gets evaluated incrementall as demanded by the consumer. For example:

> S.foldr (:) [] $S.fromList [1,2,3,4] [1,2,3,4]  When f is a function strict in its second argument, the right side of the expression gets evaluated as follows: foldr f z xs == x1 f tail1 tail1 == x2 f tail2 tail2 == x3 f tail3 ... tailn == xn f z  In foldl' we have both the arguments of f available at each step, therefore, each step can be reduced immediately. However, in foldr the second argument to f is a recursive call, therefore, it ends up building the whole expression in memory before it can be reduced, consuming the whole input. This makes foldr much less efficient for reduction compared to foldl'. For example: > S.foldr (+) 0$ S.fromList [1,2,3,4]
10


When the underlying monad m is strict (e.g. IO), then foldr ends up evaluating all of its input because of strict evaluation of the recursive call:

> S.foldr (\_ _ -> []) [] $S.fromList (1:undefined) *** Exception: Prelude.undefined In a lazy monad, we can consume the input lazily, and terminate the fold by conditionally not inspecting the recursive expression. > runIdentity$ S.foldr (\x rest -> if x == 3 then [] else x : rest) [] $S.fromList (4:1:3:undefined) [4,1] The arguments to the folding function (a -> b -> b) are in the head and tail order of the output, a is the head and b is the tail. Remember, in a right fold the zero is on the right, it is the tail end. Since: 0.1.0 foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a) Source # Lazy right fold for non-empty streams, using first element as the starting value. Returns Nothing if the stream is empty. Since: 0.5.0 foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b Source # Lazy right fold with a monadic step function. For example, to fold a stream into a list: >> S.foldrM (\x xs -> return (x : xs)) []$ fromList [1,2,3]
[1,2,3]


Since: 0.2.0

foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b Source #

Strict left associative fold.

For lists a foldl looks like:

foldl f z []     = z
foldl f z (x:xs) = foldl f (z f x) xs


The recursive call at the head of the output expression is bound to be evaluated until recursion terminates, deconstructing the whole input container and building the following left associated expression:

foldl f z xs == (((z f x1) f x2) ...) f xn


When f is a constructor, we can see that the first deconstruction of this expression would be the recursive expression on the left and xn on the right. Therefore, it can access the input elements only in the reverse (LIFO) order. For example:

> S.foldl' (flip (:)) [] $S.fromList [1,2,3,4] [4,3,2,1]  The strict left fold foldl' forces the reduction of its argument z f x before using it, therefore it never builds the whole expression in memory. Thus, z f x1 would get reduced to z1 and then z1 f x2 would get reduced to z2 and so on, incrementally reducing the expression as it recurses. However, it evaluates the accumulator only to WHNF, it may further help to use a strict data structure as accumulator. For example: > S.foldl' (+) 0$ S.fromList [1,2,3,4]
10

0 + 1
(0 + 1) + 2
((0 + 1) + 2) + 3
(((0 + 1) + 2) + 3) + 4


foldl strictly deconstructs the whole input container irrespective of whether it needs it or not:

> S.foldl' (\acc x -> if x == 3 then acc else x : acc) [] $S.fromList (4:1:3:undefined) *** Exception: Prelude.undefined However, evaluation of the items contained in the input container is lazy as demanded by the fold step function: > S.foldl' (\acc x -> if x == 3 then acc else x : acc) []$ S.fromList [4,1,3,undefined]
[4,1]

To perform a left fold without consuming all the input one can use scanl to stream the intermediate results of the fold and use them lazily.

In stateful or event-driven programming, we can consider z as the initial state and the stream being folded as a stream of events, thus foldl' processes all the events in the stream updating the state on each event and then ultimately returning the final state.

The arguments to the folding function (b -> a -> b) are in the head and tail order of the output expression, b is the head and a is the tail. Remember, in a left fold the zero is on the left, at the head of the expression.

Since: 0.2.0

foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a) Source #

Strict left fold, for non-empty streams, using first element as the starting value. Returns Nothing if the stream is empty.

Since: 0.5.0

foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b Source #

Like foldl' but with a monadic step function.

Since: 0.2.0

foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b Source #

Strict left fold with an extraction function. Like the standard strict left fold, but applies a user supplied extraction function (the third argument) to the folded value at the end. This is designed to work with the foldl library. The suffix x is a mnemonic for extraction.

Since: 0.2.0

foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b Source #

Like foldx, but with a monadic step function.

Since: 0.2.0

## Run Effects

runStream :: Monad m => SerialT m a -> m () Source #

Run a stream, discarding the results. By default it interprets the stream as SerialT, to run other types of streams use the type adapting combinators for example runStream . asyncly.

Since: 0.2.0

runN :: Monad m => Int -> SerialT m a -> m () Source #

runN n = runStream . take n

Run maximum up to n iterations of a stream.

Since: 0.6.0

runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m () Source #

runWhile p = runStream . takeWhile p

Run a stream as long as the predicate holds true.

Since: 0.6.0

## To Elements

Folds that extract selected elements of a stream or their properties.

(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a) Source #

Lookup the element at the given index.

Since: 0.6.0

head :: Monad m => SerialT m a -> m (Maybe a) Source #

Extract the first element of the stream, if any.

head = (!! 0)

Since: 0.1.0

last :: Monad m => SerialT m a -> m (Maybe a) Source #

Extract the last element of the stream, if any.

last xs = xs !! (length xs - 1)

Since: 0.1.1

findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a) Source #

Returns the first element that satisfies the given predicate.

Since: 0.6.0

find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a) Source #

Like findM but with a non-monadic predicate.

find p = findM (return . p)

Since: 0.5.0

lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b) Source #

In a stream of (key-value) pairs (a, b), return the value b of the first pair where the key equals the given value a.

lookup = snd <$> find ((==) . fst) Since: 0.5.0 findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int) Source # Returns the first index that satisfies the given predicate. Since: 0.5.0 elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int) Source # Returns the first index where a given value is found in the stream. elemIndex a = findIndex (== a) Since: 0.5.0 ## To Parts Folds that extract selected parts of a stream. tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a)) Source # Extract all but the first element of the stream, if any. Since: 0.1.1 init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a)) Source # Extract all but the last element of the stream, if any. Since: 0.5.0 ## To Boolean Folds that summarize the stream to a boolean value. null :: Monad m => SerialT m a -> m Bool Source # Determine whether the stream is empty. Since: 0.1.1 elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool Source # Determine whether an element is present in the stream. Since: 0.1.0 notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool Source # Determine whether an element is not present in the stream. Since: 0.1.0 all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool Source # Determine whether all elements of a stream satisfy a predicate. Since: 0.1.0 any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool Source # Determine whether any of the elements of a stream satisfy a predicate. Since: 0.1.0 and :: Monad m => SerialT m Bool -> m Bool Source # Determines if all elements of a boolean stream are True. Since: 0.5.0 or :: Monad m => SerialT m Bool -> m Bool Source # Determines whether at least one element of a boolean stream is True. Since: 0.5.0 ## To Summary Folds that summarize the stream to a single value. length :: Monad m => SerialT m a -> m Int Source # Determine the length of the stream. Since: 0.1.0 sum :: (Monad m, Num a) => SerialT m a -> m a Source # Determine the sum of all elements of a stream of numbers. Returns 0 when the stream is empty. Note that this is not numerically stable for floating point numbers. Since: 0.1.0 product :: (Monad m, Num a) => SerialT m a -> m a Source # Determine the product of all elements of a stream of numbers. Returns 1 when the stream is empty. Since: 0.1.1 ## To Summary (Maybe) Folds that summarize a non-empty stream to a Just value and return Nothing for an empty stream. maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a) Source # Determine the maximum element in a stream using the supplied comparison function. Since: 0.6.0 maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a) Source # maximum = maximumBy compare  Determine the maximum element in a stream. Since: 0.1.0 minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a) Source # Determine the minimum element in a stream using the supplied comparison function. Since: 0.6.0 minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a) Source # minimum = minimumBy compare  Determine the minimum element in a stream. Since: 0.1.0 the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a) Source # Ensures that all the elements of the stream are identical and then returns that unique element. Since: 0.6.0 ## To Containers Convert or divert a stream into an output structure, container or sink. toList :: Monad m => SerialT m a -> m [a] Source # toList = S.foldr (:) []  Convert a stream into a list in the underlying monad. Same as: Since: 0.1.0 toHandle :: MonadIO m => Handle -> SerialT m String -> m () Source # toHandle h = S.mapM_$ hPutStrLn h


Write a stream of Strings to an IO Handle.

Since: 0.1.0

# Transformation

## Scanning

Scans stream all the intermediate reduction steps of the corresponding folds. The following equations hold for lists:

scanl f z xs == map (foldl f z) $inits xs scanr f z xs == map (foldr f z)$ tails

We do not provide a right associative scan, it can be recovered from a scanl' as follows:

scanr f z xs ==  reverse $scanl' (flip f) z (reverse xs) Scan is like a stateful map. If we discard the state, we get the map: S.drop 1$ S.scanl' (\_ x -> f x) z xs == map f xs

scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b Source #

Strict left scan.

> S.toList $S.scanl' (+) 0$ fromList [1,2,3,4]
[0,1,3,6,10]

> S.toList $S.scanl' (flip (:)) []$ S.fromList [1,2,3,4]
[[],[1],[2,1],[3,2,1],[4,3,2,1]]


The output of scanl' is the initial value of the accumulator followed by all the intermediate steps and the final result of foldl'.

By streaming the accumulated state after each fold step, we can share the state across multiple stages of stream composition. Each stage can modify or extend the state, do some processing with it and emit it for the next stage, thus modularizing the stream processing. This can be useful in stateful or event-driven programming.

Consider the following example, computing the sum and the product of the elements in a stream in one go using a foldl':

> S.foldl' (\(s, p) x -> (s + x, p * x)) (0,1) $S.fromList [1,2,3,4] (10,24)  Using scanl' we can compute the sum in the first stage and pass it down to the next stage for computing the product: > S.foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)$ S.scanl' (\(s, _) x -> (s + x, x)) (0,1)
$S.fromList [1,2,3,4] (10,24)  IMPORTANT: scanl' evaluates the accumulator to WHNF. To avoid building lazy expressions inside the accumulator, it is recommended that a strict data structure is used for accumulator. Since: 0.2.0 scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b Source # Like scanl' but with a monadic fold function. Since: 0.4.0 scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a Source # Like scanl' but for a non-empty stream. The first element of the stream is used as the initial value of the accumulator. Does nothing if the stream is empty. > S.toList$ S.scanl1 (+) $fromList [1,2,3,4] [1,3,6,10]  Since: 0.6.0 scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a Source # Like scanl1' but with a monadic step function. Since: 0.6.0 scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b Source # Strict left scan with an extraction function. Like scanl', but applies a user supplied extraction function (the third argument) at each step. This is designed to work with the foldl library. The suffix x is a mnemonic for extraction. Since: 0.2.0 ## Mapping Map is a strictly one-to-one transformation of stream elements. It cannot add or remove elements from the stream, just transforms them. map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b Source # map = fmap  Same as fmap. > S.toList$ S.map (+1) $S.fromList [1,2,3] [2,3,4]  Since: 0.4.0 ## Flattening sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a Source # sequence = mapM id  Replace the elements of a stream of monadic actions with the outputs of those actions. > runStream$ S.sequence $S.fromList [putStr "a", putStr "b", putStrLn "c"] abc runStream$ S.replicateM 10 (return $threadDelay 1000000 >> print 1) & (serially . S.sequence) runStream$ S.replicateM 10 (return $threadDelay 1000000 >> print 1) & (asyncly . S.sequence)  Concurrent (do not use with parallely on infinite streams) Since: 0.1.0 mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b Source # mapM f = sequence . map f  Apply a monadic function to each element of the stream and replace it with the output of the resulting action. > runStream$ S.mapM putStr $S.fromList ["a", "b", "c"] abc runStream$ S.replicateM 10 (return 1)
& (serially . S.mapM (\x -> threadDelay 1000000 >> print x))

runStream $S.replicateM 10 (return 1) & (asyncly . S.mapM (\x -> threadDelay 1000000 >> print x))  Concurrent (do not use with parallely on infinite streams) Since: 0.1.0 ## Filtering Filtering may remove some elements from the stream. filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a Source # Include only those elements that pass a predicate. Since: 0.1.0 filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a Source # Same as filter but with a monadic predicate. Since: 0.4.0 take :: (IsStream t, Monad m) => Int -> t m a -> t m a Source # Take first n elements from the stream and discard the rest. Since: 0.1.0 takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a Source # End the stream as soon as the predicate fails on an element. Since: 0.1.0 takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a Source # Same as takeWhile but with a monadic predicate. Since: 0.4.0 drop :: (IsStream t, Monad m) => Int -> t m a -> t m a Source # Discard first n elements from the stream and take the rest. Since: 0.1.0 dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a Source # Drop elements in the stream as long as the predicate succeeds and then take the rest of the stream. Since: 0.1.0 dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a Source # Same as dropWhile but with a monadic predicate. Since: 0.4.0 deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a Source # Deletes the first occurence of the element in the stream that satisfies the given equality predicate. > S.toList$ S.deleteBy (==) 3 $S.fromList [1,3,3,5] [1,3,5]  Since: 0.6.0 uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a Source # Drop repeated elements that are adjacent to each other. Since: 0.6.0 ## Insertion Insertion adds more elements to the stream. insertBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a Source # insertBy cmp elem stream inserts elem before the first element in stream that is less than elem when compared using cmp. insertBy cmp x = mergeBy cmp (yield x)  > S.toList$ S.insertBy compare 2 $S.fromList [1,3,5] [1,2,3,5]  Since: 0.6.0 intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a Source # Generate a stream by performing a monadic action between consecutive elements of the given stream. Concurrent (do not use with parallely on infinite streams) > S.toList$ S.intersperseM (putChar 'a' >> return ',') $S.fromList "hello" aaaa"h,e,l,l,o"  Since: 0.5.0 ## Reordering reverse :: IsStream t => t m a -> t m a Source # Returns the elements of the stream in reverse order. The stream must be finite. Since: 0.1.1 # Hybrid Operations ## Map and Fold mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m () Source # Apply a monadic action to each element of the stream and discard the output of the action. Since: 0.1.0 ## Map and Filter mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b Source # Map a Maybe returning function to a stream, filter out the Nothing elements, and return a stream of values extracted from Just. Since: 0.3.0 mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m)) => (a -> m (Maybe b)) -> t m a -> t m b Source # Like mapMaybe but maps a monadic function. Concurrent (do not use with parallely on infinite streams) Since: 0.3.0 ## Scan and filter findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int Source # Find all the indices where the element in the stream satisfies the given predicate. Since: 0.5.0 elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int Source # Find all the indices where the value of the element in the stream is equal to the given value. Since: 0.5.0 # Multi-Stream Operations New streams can be constructed by appending, merging or zipping existing streams. ## Appending Streams form a Semigroup and a Monoid under the append operation. >> S.toList$ S.fromList [1,2] <> S.fromList [3,4]
[1,2,3,4]
>> S.toList $fold$ [S.fromList [1,2], S.fromList [3,4]]
[1,2,3,4]


## Merging

Streams form a commutative semigroup under the merge operation.

mergeBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a Source #

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.

> S.toList $S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8]) [1,2,3,4,5,6,8]  Since: 0.6.0 mergeByM :: (IsStream t, Monad m) => (a -> a -> m Ordering) -> t m a -> t m a -> t m a Source # Like mergeBy but with a monadic comparison function. Merge two streams randomly: > randomly _ _ = randomIO >>= x -> return$ if x then LT else GT
> S.toList $S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2]) [2,1,2,2,2,1,1,1]  Merge two streams in a proportion of 2:1: proportionately m n = do ref <- newIORef$ cycle $concat [replicate m LT, replicate n GT] return$ \_ _ -> do
writeIORef ref $tail r return$ head r

main = do
f <- proportionately 2 1
xs <- S.toList $S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2]) print xs  [1,1,2,1,1,2,1,1,2]  Since: 0.6.0 mergeAsyncBy :: (IsStream t, MonadAsync m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a Source # Like mergeBy but merges concurrently (i.e. both the elements being merged are generated concurrently). Since: 0.6.0 mergeAsyncByM :: (IsStream t, MonadAsync m) => (a -> a -> m Ordering) -> t m a -> t m a -> t m a Source # Like mergeByM but merges concurrently (i.e. both the elements being merged are generated concurrently). Since: 0.6.0 ## Zipping zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c Source # Zip two streams serially using a pure zipping function. > S.toList$ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
[5,7,9]


Since: 0.1.0

zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c Source #

Like zipWith but using a monadic zipping function.

Since: 0.4.0

zipAsyncWith :: (IsStream t, MonadAsync m) => (a -> b -> c) -> t m a -> t m b -> t m c Source #

Like zipWith but zips concurrently i.e. both the streams being zipped are generated concurrently.

Since: 0.1.0

zipAsyncWithM :: (IsStream t, MonadAsync m) => (a -> b -> m c) -> t m a -> t m b -> t m c Source #

Like zipWithM but zips concurrently i.e. both the streams being zipped are generated concurrently.

Since: 0.4.0

indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a) Source #

indexed = S.zipWith (,) (S.intFrom 0)

Pair each element in a stream with its index.

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


Since: 0.6.0

indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a) Source #

indexedR n = S.zipWith (,) (S.intFromThen n (n - 1))

Pair each element in a stream with its index, starting from the given index n and counting down.

> S.toList $S.indexedR 10$ S.fromList "hello"
[(9,h),(8,e),(7,l),(6,l),(5,o)]


Since: 0.6.0

## Flattening

concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b Source #

Map each element to a stream using a monadic function and then flatten the results into a single stream.

Since: 0.6.0

concatMap :: (IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b Source #

Map each element to a stream and then flatten the results into a single stream.

concatMap f = concatMapM (return . f)

Since: 0.6.0

## Folds

eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool Source #

Compare two streams for equality using an equality function.

Since: 0.6.0

cmpBy :: (IsStream t, Monad m) => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering Source #

Compare two streams lexicographically using a comparison function.

Since: 0.6.0

isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool Source #

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

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


Since: 0.6.0

isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool Source #

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 treated as a subsequence of itself.

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


Since: 0.6.0

stripPrefix :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m (Maybe (t m a)) Source #

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

Since: 0.6.0

# Deprecated

once :: (Monad m, IsStream t) => m a -> t m a Source #

Same as yieldM

Since: 0.2.0

each :: (IsStream t, Foldable f) => f a -> t m a Source #

Same as fromFoldable.

Since: 0.1.0

scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b Source #

Since: 0.1.1

foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b Source #