| Copyright | (c) 2017 Harendra Kumar |
|---|---|
| License | BSD3 |
| Maintainer | harendra.kumar@gmail.com |
| Stability | experimental |
| Portability | GHC |
| Safe Haskell | None |
| Language | Haskell2010 |
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
- nil :: IsStream t => t m a
- cons :: IsStream t => a -> t m a -> t m a
- (.:) :: IsStream t => a -> t m a -> t m a
- consM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
- (|:) :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
- yield :: IsStream t => a -> t m a
- yieldM :: (Monad m, IsStream t) => m a -> t m a
- repeat :: IsStream t => a -> t m a
- repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
- replicate :: (IsStream t, Monad m) => Int -> a -> t m a
- replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
- class Enum a => Enumerable a where
- enumerateFrom :: (IsStream t, Monad m) => a -> t m a
- enumerateFromTo :: (IsStream t, Monad m) => a -> a -> t m a
- enumerateFromThen :: (IsStream t, Monad m) => a -> a -> t m a
- enumerateFromThenTo :: (IsStream t, Monad m) => a -> a -> a -> t m a
- enumerate :: (IsStream t, Monad m, Bounded a, Enumerable a) => t m a
- enumerateTo :: (IsStream t, Monad m, Bounded a, Enumerable a) => a -> t m a
- unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
- unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
- iterate :: IsStream t => (a -> a) -> a -> t m a
- iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
- fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
- fromIndicesM :: (IsStream t, Monad m) => (Int -> m a) -> t m a
- fromList :: (Monad m, IsStream t) => [a] -> t m a
- fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
- fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
- fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
- fromHandle :: (IsStream t, MonadIO m) => Handle -> t m String
- uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
- foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
- foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
- foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b
- foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
- foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
- foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
- foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
- foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
- runStream :: Monad m => SerialT m a -> m ()
- runN :: Monad m => Int -> SerialT m a -> m ()
- runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
- (!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
- head :: Monad m => SerialT m a -> m (Maybe a)
- last :: Monad m => SerialT m a -> m (Maybe a)
- findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
- find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
- lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
- findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
- elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
- tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
- init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
- null :: Monad m => SerialT m a -> m Bool
- elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
- notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
- all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
- any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
- and :: Monad m => SerialT m Bool -> m Bool
- or :: Monad m => SerialT m Bool -> m Bool
- length :: Monad m => SerialT m a -> m Int
- sum :: (Monad m, Num a) => SerialT m a -> m a
- product :: (Monad m, Num a) => SerialT m a -> m a
- maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
- maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
- minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
- minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
- the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
- toList :: Monad m => SerialT m a -> m [a]
- toHandle :: MonadIO m => Handle -> SerialT m String -> m ()
- scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
- scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
- scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
- scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
- scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
- map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
- sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
- mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
- filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
- filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
- take :: (IsStream t, Monad m) => Int -> t m a -> t m a
- takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
- takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
- drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
- dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
- dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
- deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
- uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
- insertBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
- intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
- reverse :: IsStream t => t m a -> t m a
- mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
- mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
- mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m)) => (a -> m (Maybe b)) -> t m a -> t m b
- findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
- elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
- mergeBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
- mergeByM :: (IsStream t, Monad m) => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
- mergeAsyncBy :: (IsStream t, MonadAsync m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
- mergeAsyncByM :: (IsStream t, MonadAsync m) => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
- zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
- zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
- zipAsyncWith :: (IsStream t, MonadAsync m) => (a -> b -> c) -> t m a -> t m b -> t m c
- zipAsyncWithM :: (IsStream t, MonadAsync m) => (a -> b -> m c) -> t m a -> t m b -> t m c
- indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
- indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
- concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
- concatMap :: (IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
- eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
- cmpBy :: (IsStream t, Monad m) => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
- isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
- isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
- stripPrefix :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m (Maybe (t m a))
- once :: (Monad m, IsStream t) => m a -> t m a
- each :: (IsStream t, Foldable f) => f a -> t m a
- scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
- foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
- foldlM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
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.
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
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$enumFromThenfrom 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
enumerateTo :: (IsStream t, Monad m, Bounded a, Enumerable a) => a -> t m a Source #
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 =foldrconsnil
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 =foldrconsMnil
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 #
fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a Source #
fromFoldableM =foldrconsMnil
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
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.
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.
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)
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
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
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.
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.
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
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
r <- readIORef ref
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 #
Deprecated: Please use yieldM instead.
Same as yieldM
Since: 0.2.0
each :: (IsStream t, Foldable f) => f a -> t m a Source #
Deprecated: Please use fromFoldable instead.
Same as fromFoldable.
Since: 0.1.0
scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b Source #
Deprecated: Please use scanx instead.
Since: 0.1.1