Streaming

Free monad transformer

The Stream data type is equivalent to FreeT and can represent any effectful succession of steps, where the form of the steps or commands is specified by the first (functor) parameter. The (hidden) implementation is

data Stream f m r = Step !(f (Stream f m r)) | Delay (m (Stream f m r)) | Return r

In the simplest case, the base functor is (,) a . Here the news or command at each step is an individual element of type a , i.e. the command is a yield statement. The associated Streaming Prelude uses the left-strict pair Of a b in place of the Haskell pair (a,b) In it, various operations are defined for fundamental streaming types like

Stream (Of a) m r                   -- a generator or producer (in the pipes sense) 
                                       -- compare [a], or rather ([a],r) 
Stream (Of a) m (Stream (Of a) m r) -- the effectful splitting of a producer
                                       -- compare ([a],[a]) or rather ([a],([a],r))
Stream (Stream (Of a) m) m r        -- segmentation of a producer
                                       -- cp. [[a]], or rather ([a],([a],([a],(...,r))))

and so on. But of course any functor can be used, and this is part of the point of this prelude - as we already see from the type of the segmented stream, Stream (Stream (Of a) m) m r

and operations like e.g.

chunksOf :: Monad m => Int -> Stream f m r -> Stream (Stream f m) m r
mapsM Streaming.Prelude.length' :: Stream (Stream (Of a) m) r -> Stream (Of Int) m r

To avoid breaking reasoning principles, the constructors should not be used directly. A pattern-match should go by way of inspect - or, in the producer case, next. These mirror the type of runFreeT. The constructors are exported by the Internal module.

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)

Constructing a Stream on a base functor

unfold :: (Monad m, Functor f) => (s -> m (Either r (f s))) -> s -> Stream f m r

Build a Stream by unfolding steps starting from a seed. See also the specialized unfoldr in the prelude.

unfold inspect = id -- modulo the quotient we work with
unfold Pipes.next :: Monad m => Producer a m r -> Stream ((,) a) m r
unfold (curry (:>) . Pipes.next) :: Monad m => Producer a m r -> Stream (Of a) m r

construct :: (forall b. (f b -> b) -> (m b -> b) -> (r -> b) -> b) -> Stream f m r

Reflect a church-encoded stream; cp. GHC.Exts.build

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.

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

Lift for items in the base functor. Makes a singleton or one-layer succession.`

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

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

Repeat a functorial layer, command or instruct several times.

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

Repeat a functorial layer, command or instruction forever.

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

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

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

Transforming streams

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 transformation

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

Map layers of one functor to another with a transformation involving the base monad

distribute :: (Monad m, Functor f, MonadTrans t, MFunctor t, Monad (t (Stream f m))) => Stream f (t m) r -> t (Stream f m) r

Make it possible to 'run' the underlying transformed monad. A simple minded example might be:

debugFibs = flip runStateT 1 $ distribute $ loop 1 where
  loop n = do
    S.yield n
    s <- lift get 
    liftIO $ putStr "Current state is:  " >> print s
    lift $ put (s + n :: Int)
    loop s

>>> S.print $  S.take 4 $ S.drop 4 $ debugFibs
Current state is:  1
Current state is:  2
Current state is:  3
Current state is:  5
5
Current state is:  8
8
Current state is:  13
13
Current state is:  21
21

Inspecting a stream

inspect :: (Functor f, Monad m) => Stream f m r -> m (Either r (f (Stream f m r)))

Inspect the first stage of a freely layered sequence. Compare Pipes.next and the replica Streaming.Prelude.next. This is the uncons for the general unfold.

unfold inspect = id
Streaming.Prelude.unfoldr StreamingPrelude.next = id

Zipping streams

zips :: (Monad m, Functor f, Functor g) => Stream f m r -> Stream g m r -> Stream (Compose f g) m r

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

interleaves :: (Monad m, Applicative h) => Stream h m r -> Stream h m r -> Stream h m r

Interleave functor layers, with the effects of the first preceding the effects of the second.

interleaves = zipsWith (liftA2 (,))

>>> let paste = \a b -> interleaves (Q.lines a) (maps (Q.cons' '\t') (Q.lines b))
>>> Q.stdout $ Q.unlines $ paste "hello\nworld\n" "goodbye\nworld\n"
hello	goodbye
world	world

Eliminating a Stream

intercalates :: (Monad m, Monad (t m), MonadTrans t) => t m a -> Stream (t m) m b -> t m b

Interpolate a layer at each segment. This specializes to e.g.

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

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

Dissolves the segmentation into layers of Stream f m layers.

concats stream = destroy stream join (join . lift) return

>>> S.print $ concats $ maps (cons 1776) $ chunksOf 2 (each [1..5])
1776
1
2
1776
3
4
1776
5

iterTM :: (Functor f, Monad m, MonadTrans t, Monad (t m)) => (f (t m a) -> t m a) -> Stream f m a -> t m a

Specialized fold

iterTM alg stream = destroy stream alg (join . lift) return

iterT :: (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a

Specialized fold

iterT alg stream = destroy stream alg join return

destroy :: (Functor f, Monad m) => Stream f m r -> (f b -> b) -> (m b -> b) -> (r -> b) -> b

Map a stream directly to its church encoding; compare Data.List.foldr It permits distinctions that should be hidden, as can be seen from e.g.

isPure stream = destroy_ (const True) (const False) (const True)

and similar nonsense. The crucial constraint is that the m x -> x argument is an Eilenberg-Moore algebra. See Atkey "Reasoning about Stream Processing with Effects"

The destroy exported by the safe modules is

destroy str = destroy (observe str)

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

Map each layer to an effect in the base monad, and run them all.

runEffect :: Monad m => Stream m m r -> m r

Run the effects in a stream that merely layers effects.

Splitting and joining Streams

splitsAt :: (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.

>>> rest <- S.print $ S.splitAt 1 $ each [1..3]
1
>>> S.print rest
2
3

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

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

Break a stream into substreams each with n functorial layers.

>>> S.print $ maps' sum' $ chunksOf 2 $ each [1,1,1,1,1,1,1]
2
2
2
1

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

Dissolves the segmentation into layers of Stream f m layers.

concats stream = destroy stream join (join . lift) return

>>> S.print $ concats $ maps (cons 1776) $ chunksOf 2 (each [1..5])
1776
1
2
1776
3
4
1776
5

Base functor for streams of individual items

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

re-exports

class MFunctor t where

A functor in the category of monads, using hoist as the analog of fmap:

hoist (f . g) = hoist f . hoist g

hoist id = id

Methods

hoist :: Monad m => (forall a. m a -> n a) -> t m b -> t n b

Lift a monad morphism from m to n into a monad morphism from (t m) to (t n)

Instances

MFunctor ListT

MFunctor Backwards

MFunctor MaybeT

MFunctor IdentityT

MFunctor Lift

MFunctor (ReaderT r)

MFunctor (StateT s)

MFunctor (StateT s)

MFunctor (ExceptT e)

MFunctor (ErrorT e)

MFunctor (WriterT w)

MFunctor (WriterT w)

MFunctor (Product f)

Functor f => MFunctor (Compose f)

Functor f => MFunctor (Stream f)

MFunctor (RWST r w s)

MFunctor (RWST r w s)

class (MFunctor t, MonadTrans t) => MMonad t where

A monad in the category of monads, using lift from MonadTrans as the analog of return and embed as the analog of (=<<):

embed lift = id

embed f (lift m) = f m

embed g (embed f t) = embed (\m -> embed g (f m)) t

Methods

embed :: Monad n => (forall a. m a -> t n a) -> t m b -> t n b

Embed a newly created MMonad layer within an existing layer

embed is analogous to (=<<)

Instances

MMonad ListT

MMonad MaybeT

MMonad IdentityT

MMonad (ReaderT r)

MMonad (ExceptT e)

Error e => MMonad (ErrorT e)

Monoid w => MMonad (WriterT w)

Monoid w => MMonad (WriterT w)

Functor f => MMonad (Stream f)

class MonadTrans t where

The class of monad transformers. Instances should satisfy the following laws, which state that lift is a monad transformation:

• lift . return = return

• lift (m >>= f) = lift m >>= (lift . f)

Methods

lift :: Monad m => m a -> t m a

Lift a computation from the argument monad to the constructed monad.

Instances

MonadTrans ListT

MonadTrans MaybeT

MonadTrans IdentityT

MonadTrans (ReaderT r)

MonadTrans (StateT s)

MonadTrans (StateT s)

MonadTrans (ExceptT e)

Error e => MonadTrans (ErrorT e)

Monoid w => MonadTrans (WriterT w)

Monoid w => MonadTrans (WriterT w)

Functor f => MonadTrans (Stream f)

Monoid w => MonadTrans (RWST r w s)

Monoid w => MonadTrans (RWST r w s)

class Monad m => MonadIO m where

Monads in which IO computations may be embedded. Any monad built by applying a sequence of monad transformers to the IO monad will be an instance of this class.

Instances should satisfy the following laws, which state that liftIO is a transformer of monads:

• liftIO . return = return

• liftIO (m >>= f) = liftIO m >>= (liftIO . f)

Methods

liftIO :: IO a -> m a

Lift a computation from the IO monad.

Instances

MonadIO IO

MonadIO m => MonadIO (ListT m)

MonadIO m => MonadIO (MaybeT m)

MonadIO m => MonadIO (IdentityT m)

MonadIO m => MonadIO (ReaderT r m)

MonadIO m => MonadIO (StateT s m)

MonadIO m => MonadIO (StateT s m)

MonadIO m => MonadIO (ExceptT e m)

(Error e, MonadIO m) => MonadIO (ErrorT e m)

(Monoid w, MonadIO m) => MonadIO (WriterT w m)

(Monoid w, MonadIO m) => MonadIO (WriterT w m)

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

(Monoid w, MonadIO m) => MonadIO (RWST r w s m)

(Monoid w, MonadIO m) => MonadIO (RWST r w s m)

newtype Compose f g a :: (* -> *) -> (* -> *) -> * -> * infixr 9

Right-to-left composition of functors. The composition of applicative functors is always applicative, but the composition of monads is not always a monad.

Constructors

Compose infixr 9

Fields

getCompose :: f (g a)

Instances

Functor f => MFunctor (Compose f)

(Functor f, Functor g) => Functor (Compose f g)

(Applicative f, Applicative g) => Applicative (Compose f g)

(Foldable f, Foldable g) => Foldable (Compose f g)

(Traversable f, Traversable g) => Traversable (Compose f g)

(Alternative f, Applicative g) => Alternative (Compose f g)

(Functor f, Eq1 f, Eq1 g) => Eq1 (Compose f g)

(Functor f, Ord1 f, Ord1 g) => Ord1 (Compose f g)

(Functor f, Read1 f, Read1 g) => Read1 (Compose f g)

(Functor f, Show1 f, Show1 g) => Show1 (Compose f g)

(Functor f, Eq1 f, Eq1 g, Eq a) => Eq (Compose f g a)

(Functor f, Ord1 f, Ord1 g, Ord a) => Ord (Compose f g a)

(Functor f, Read1 f, Read1 g, Read a) => Read (Compose f g a)

(Functor f, Show1 f, Show1 g, Show a) => Show (Compose f g a)

join :: Monad m => m (m a) -> m a

The join function is the conventional monad join operator. It is used to remove one level of monadic structure, projecting its bound argument into the outer level.

liftA2 :: Applicative f => (a -> b -> c) -> f a -> f b -> f c

Lift a binary function to actions.

liftA3 :: Applicative f => (a -> b -> c -> d) -> f a -> f b -> f c -> f d

Lift a ternary function to actions.

void :: Functor f => f a -> f ()

void value discards or ignores the result of evaluation, such as the return value of an IO action.

Examples

Replace the contents of a Maybe Int with unit:

>>> void Nothing
Nothing
>>> void (Just 3)
Just ()

Replace the contents of an Either Int Int with unit, resulting in an Either Int '()':

>>> void (Left 8675309)
Left 8675309
>>> void (Right 8675309)
Right ()

Replace every element of a list with unit:

>>> void [1,2,3]
[(),(),()]

Replace the second element of a pair with unit:

>>> void (1,2)
(1,())

Discard the result of an IO action:

>>> mapM print [1,2]
1
2
[(),()]
>>> void $ mapM print [1,2]
1
2