Streaming.Prelude

This module is very closely modeled on Pipes.Prelude.

Import qualified thus:

import Streaming
import qualified Streaming as S

The Streaming module exports types, functor-general operations and some other kit; it may clash with free and pipes-group, but not with standard base modules.

Interoperation with pipes is accomplished with this isomorphism, which uses Pipes.Prelude.unfoldr from HEAD:

Pipes.unfoldr Streaming.next        :: Stream (Of a) m r   -> Producer a m r
Streaming.unfoldr Pipes.next        :: Producer a m r      -> Stream (Of a) m r                     

Interoperation with iostreams is thus:

Streaming.reread IOStreams.read     :: InputStream a       -> Stream (Of a) IO ()
IOStreams.unfoldM Streaming.uncons  :: Stream (Of a) IO () -> IO (InputStream a)

A simple exit to conduit would be, for example:

Conduit.unfoldM Streaming.uncons    :: Stream (Of a) m ()  -> Source m a

Types

data Stream f m r

Instances

Functor f => MFunctor (Stream f)

Functor f => MMonad (Stream f)

Functor f => MonadTrans (Stream f)

(Functor f, Monad m) => Monad (Stream f m)

(Functor f, Monad m) => Functor (Stream f m)

(Functor f, Monad m) => Applicative (Stream f m)

(MonadIO m, Functor f) => MonadIO (Stream f m)

(Eq r, Eq (m (Stream f m r)), Eq (f (Stream f m r))) => Eq (Stream f m r)

(Typeable (* -> *) f, Typeable (* -> *) m, Data r, Data (m (Stream f m r)), Data (f (Stream f m r))) => Data (Stream f m r)

(Show r, Show (m (Stream f m r)), Show (f (Stream f m r))) => Show (Stream f m r)

data Of a b

A left-strict pair; the base functor for streams of individual elements.

Constructors

!a :> b infixr 4

Instances

Functor (Of a)

Foldable (Of a)

Traversable (Of a)

(Eq a, Eq b) => Eq (Of a b)

(Data a, Data b) => Data (Of a b)

(Ord a, Ord b) => Ord (Of a b)

(Read a, Read b) => Read (Of a b)

(Show a, Show b) => Show (Of a b)

lazily :: Of a b -> (a, b)

strictly :: (a, b) -> Of a b

Introducing streams of elements

yield :: Monad m => a -> Stream (Of a) m ()

A singleton stream

>>> S.sum $ do {S.yield 1; lift $ putStrLn "hello"; S.yield 2; lift $ putStrLn "goodbye"; S.yield 3}
hello
goodbye
6

>>> S.sum $ S.take 3 $ forever $ do {lift $ putStrLn "enter a number" ; n <- lift $ readLn; S.yield n }
enter a number
100
enter a number
200
enter a number
300
600

enter a number 1 enter a number 1000 1001

each :: (Monad m, Foldable f) => f a -> Stream (Of a) m ()

Stream the elements of a foldable container.

>>> S.print $ S.each [1..3]
1
2
3

unfoldr :: Monad m => (s -> m (Either r (a, s))) -> s -> Stream (Of a) m r

Build a Stream by unfolding steps starting from a seed. This is one natural way to consume a Producer. It is worth adding it to the functor-general unfold to avoid dealing with the left-strict pairing we are using in place of Haskell pairing.

unfoldr Pipes.next :: Monad m => Producer a m r -> Stream (Of a) m r
unfold (curry (:>) . Pipes.next) :: Monad m => Producer a m r -> Stream (Of a) m r

stdinLn :: MonadIO m => Stream (Of String) m ()

repeatedly stream lines as String from stdin

>>> S.stdoutLn $ S.show (S.each [1..3])
1
2
3

readLn :: (MonadIO m, Read a) => Stream (Of a) m ()

Read values from stdin, ignoring failed parses

>>> S.sum $ S.take 2 $ forever S.readLn :: IO Int
3
#$%^&\^?
1000
1003

fromHandle :: MonadIO m => Handle -> Stream (Of String) m ()

Read Strings from a Handle using hGetLine

Terminates on end of input

repeat :: a -> Stream (Of a) m r

Repeat an element ad inf. .

>>> S.print $ S.take 3 $ S.repeat 1
1
1
1

repeatM :: Monad m => m a -> Stream (Of a) m r

Repeat a monadic action ad inf., streaming its results.

>>> L.purely fold L.list $ S.take 2 $ repeatM getLine
hello
world
["hello","world"]

replicateM :: Monad m => Int -> m a -> Stream (Of a) m ()

Repeat an action several times, streaming the results.

Consuming streams of elements

stdoutLn :: MonadIO m => Stream (Of String) m () -> m ()

Write Strings to stdout using putStrLn; terminates on a broken output pipe

>>> S.stdoutLn $ S.show (S.each [1..3])
1
2
3

stdoutLn' :: MonadIO m => Stream (Of String) m r -> m r

Write Strings to stdout using putStrLn

This does not handle a broken output pipe, but has a polymorphic return value

mapM_ :: Monad m => (a -> m b) -> Stream (Of a) m r -> m r

Reduce a stream to its return value with a monadic action.

>>> mapM_ Prelude.print $ each [1..3] >> return True
1
2
3
True

print :: (MonadIO m, Show a) => Stream (Of a) m r -> m r

toHandle :: MonadIO m => Handle -> Stream (Of String) m r -> m r

drain :: Monad m => Stream (Of a) m r -> m r

Reduce a stream, performing its actions but ignoring its elements.

Stream transformers

map :: Monad m => (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r

Standard map on the elements of a stream.

mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r

Replace each element of a stream with the result of a monadic action

maps' :: (Monad m, Functor f) => (forall x. f x -> m (a, x)) -> Stream f m r -> Stream (Of a) m r

Map free layers of a functor to a corresponding stream of individual elements. This simplifies the use of folds marked with a ''' in Streaming.Prelude

maps' sum' :: (Monad m, Num a) => Stream (Stream (Of a) m) m r -> Stream (Of a) m r
maps' (Pipes.fold' (+) (0::Int) id) :: Monad m => Stream (Producer Int m) m r -> Stream (Of Int) m r

maps :: (Monad m, Functor f) => (forall x. f x -> g x) -> Stream f m r -> Stream g m r

Map layers of one functor to another with a natural transformation

sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r

Like the sequence but streaming. The result type is a stream of a's, but is not accumulated; the effects of the elements of the original stream are interleaved in the resulting stream. Compare:

sequence :: Monad m =>       [m a]           -> m [a]
sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r

mapFoldable :: (Monad m, Foldable t) => (a -> t b) -> Stream (Of a) m r -> Stream (Of b) m r

For each element of a stream, stream a foldable container of elements instead

>>> D.print $ D.mapFoldable show $ D.yield 12
'1'
'2'

filter :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r

Skip elements of a stream that fail a predicate

filterM :: Monad m => (a -> m Bool) -> Stream (Of a) m r -> Stream (Of a) m r

Skip elements of a stream that fail a monadic test

for :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> Stream f m x) -> Stream f m r

for replaces each element of a stream with an associated stream. Note that the associated stream may layer any functor.

take :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m ()

End stream after n elements; the original return value is lost. splitAt preserves this information. Note the function is functor-general.

takeWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m ()

End stream when an element fails a condition; the original return value is lost span preserves this information.

drop :: Monad m => Int -> Stream (Of a) m r -> Stream (Of a) m r

Ignore the first n elements of a stream, but carry out the actions

dropWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r

Ignore elements of a stream until a test succeeds.

concat :: (Monad m, Foldable f) => Stream (Of (f a)) m r -> Stream (Of a) m r

Make a stream of traversable containers into a stream of their separate elements

>>> Streaming.print $ concat (each ["hi","ho"])
'h'
'i'
'h'
'o'

>>> S.print $  S.concat (S.each [Just 1, Nothing, Just 2, Nothing])
1
2

>>> S.print $  S.concat (S.each [Right 1, Left "error!", Right 2])
1
2

scan :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of b) m r

Strict left scan, streaming, e.g. successive partial results.

Control.Foldl.purely scan :: Monad m => Fold a b -> Stream (Of a) m r -> Stream (Of b) m r

>>> Streaming.print $ Foldl.purely Streaming.scan Foldl.list $ each [3..5]
[]
[3]
[3,4]
[3,4,5]

scanM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> Stream (Of b) m r

Strict, monadic left scan

Control.Foldl.impurely scanM :: Monad m => FoldM a m b -> Stream (Of a) m r -> Stream (Of b) m r

>>> let v =  L.impurely scanM L.vector $ each [1..4::Int] :: Stream (Of (U.Vector Int)) IO ()
>>> S.print v
fromList []
fromList [1]
fromList [1,2]
fromList [1,2,3]
fromList [1,2,3,4]

chain :: Monad m => (a -> m ()) -> Stream (Of a) m r -> Stream (Of a) m r

Apply an action to all values flowing downstream

>>> let debug str = chain print str
>>> S.product (debug (S.each [2..4])) >>= print
2
3
4
24

read :: (Monad m, Read a) => Stream (Of String) m r -> Stream (Of a) m r

Make a stream of strings into a stream of parsed values, skipping bad cases

show :: (Monad m, Show a) => Stream (Of a) m r -> Stream (Of String) m r

cons :: Monad m => a -> Stream (Of a) m r -> Stream (Of a) m r

The natural cons for a Stream (Of a).

cons a stream = yield a >> stream

Useful for interoperation

Splitting and inspecting streams of elements

next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r))

The standard way of inspecting the first item in a stream of elements, if the stream is still 'running'. The Right case contains a Haskell pair, where the more general inspect would return a left-strict pair. There is no reason to prefer inspect since, if the Right case is exposed, the first element in the pair will have been evaluated to whnf.

next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r))
inspect :: Monad m => Stream (Of a) m r -> m (Either r (Of a (Stream (Of a) m r)))

Interoperate with pipes producers thus:

Pipes.unfoldr Stream.next :: Stream (Of a) m r -> Producer a m r
Stream.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r 

Similarly:

IOStreams.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) IO b -> IO (InputStream a)
Conduit.unfoldM (liftM (either (const Nothing) Just) . next)   :: Stream (Of a) m r -> Source a m r

But see uncons

uncons :: Monad m => Stream (Of a) m () -> m (Maybe (a, Stream (Of a) m ()))

Inspect the first item in a stream of elements, without a return value. uncons provides convenient exit into another streaming type:

IOStreams.unfoldM uncons :: Stream (Of a) IO b -> IO (InputStream a)
Conduit.unfoldM uncons   :: Stream (Of a) m r -> Conduit.Source m a

splitAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r)

Split a succession of layers after some number, returning a streaming or -- effectful pair. This function is the same as the splitsAt exported by the -- Streaming module, but since this module is imported qualified, it can -- usurp a Prelude name. It specializes to:

 splitAt :: (Monad m, Functor f) => Int -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)

break :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)

Break a sequence when a element falls under a predicate, keeping the rest of the stream as the return value.

>>> rest <- S.print $ S.break even $ each [1,1,2,3]
1
1
>>> S.print rest
2
3

span :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)

Stream elements until one fails the condition, return the rest.

Folds

Use these to fold the elements of a Stream. The general folds fold, fold'', foldM and 'foldM\'' are arranged for use with Foldl All functions marked with a final '\'' (e.g. 'fold\'', 'sum\') carry the stream's return value -- or, in the case of 'maps\'' are tailored to take such an operation as argument.

 maps' sum' :: (Monad m, Num n) => Stream (Stream (Of n)) m r -> Stream (Of n) m r
 maps' (fold' mappend mempty id) :: :: (Monad m, Num n) => Stream (Stream (Of n)) m r -> Stream (Of n) m r

fold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m () -> m b

Strict fold of a Stream of elements

Control.Foldl.purely fold :: Monad m => Fold a b -> Stream (Of a) m () -> m b

fold' :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m (b, r)

Strict fold of a Stream of elements that preserves the return value

Control.Foldl.purely fold' :: Monad m => Fold a b -> Stream (Of a) m r -> m (b, r)

foldM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m () -> m b

Strict, monadic fold of the elements of a 'Stream (Of a)'

Control.Foldl.impurely foldM :: Monad m => FoldM a b -> Stream (Of a) m () -> m b

foldM' :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> m (b, r)

Strict, monadic fold of the elements of a 'Stream (Of a)'

Control.Foldl.impurely foldM' :: Monad m => FoldM a b -> Stream (Of a) m r -> m (b, r)

sum :: (Monad m, Num a) => Stream (Of a) m () -> m a

Fold a Stream of numbers into their sum

sum' :: (Monad m, Num a) => Stream (Of a) m r -> m (a, r)

Fold a Stream of numbers into their sum with the return value

 maps' sum' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r

product :: (Monad m, Num a) => Stream (Of a) m () -> m a

Fold a Stream of numbers into their product

product' :: (Monad m, Num a) => Stream (Of a) m r -> m (a, r)

Fold a Stream of numbers into their product with the return value

 maps' product' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r

toList :: Stream (Of a) Identity () -> [a]

Convert a pure Stream (Of a) into a list of as

toListM :: Monad m => Stream (Of a) m () -> m [a]

Convert an effectful 'Stream (Of a)' into a list of as

Note: Needless to say this function does not stream properly. It is basically the same as mapM which, like replicateM, sequence and similar operations on traversable containers is a leading cause of space leaks.

toListM' :: Monad m => Stream (Of a) m r -> m ([a], r)

Convert an effectful Stream into a list alongside the return value

 maps' toListM' :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m 

foldrM :: Monad m => (a -> m r -> m r) -> Stream (Of a) m r -> m r

A natural right fold for consuming a stream of elements. See also the more general iterT in the Streaming module and the still more general destroy

foldrT :: (Monad m, MonadTrans t, Monad (t m)) => (a -> t m r -> t m r) -> Stream (Of a) m r -> t m r

A natural right fold for consuming a stream of elements. See also the more general iterTM in the Streaming module and the still more general destroy

foldrT (\a p -> Pipes.yield a >> p) :: Monad m => Stream (Of a) m r -> Producer a m r
foldrT (\a p -> Conduit.yield a >> p) :: Monad m => Stream (Of a) m r -> Conduit a m r

Short circuiting folds

Zips

zip :: Monad m => Stream (Of a) m r -> Stream (Of b) m r -> Stream (Of (a, b)) m r

Zip two Streamss

zipWith :: Monad m => (a -> b -> c) -> Stream (Of a) m r -> Stream (Of b) m r -> Stream (Of c) m r

Zip two Streamss using the provided combining function

Interoperation

reread :: Monad m => (s -> m (Maybe a)) -> s -> Stream (Of a) m ()

Read an IORef (Maybe a) or a similar device until it reads Nothing. reread provides convenient exit from the io-streams library

reread readIORef    :: IORef (Maybe a) -> Stream (Of a) IO ()
reread Streams.read :: System.IO.Streams.InputStream a -> Stream (Of a) IO ()