Tubes

Write effect-ful stream processing functions and compose them into a series of tubes.

This exists primarily for my own education. It is updated often as I try things and is probably, at this moment, wrong.

If you want to know more about efficient stream processing:

http://okmij.org/ftp/Streams.html

My goals were to

• learn more about iteratees and
• explore the relationships between functions, pairs, sum types, and products.

Tubes

type Tube a b = FreeT (TubeF a b)

A Tube is a computation which can

• yield an intermediate value downstream and suspend execution; and
• await a value from upstream, deferring execution until it is received.

Moreover, individual Tubes may be freely composed into larger ones, so long as their types match. Thus, one may write small, reusable building blocks and construct efficient stream process pipelines.

Since a much better engineered, more popular, and decidedly more mature library already uses the term "pipes" I have opted instead to think of my work as a series of tubes.

newtype TubeF a b k

TubeF is the union of unary functions and binary products into a single type, here defined with a Boehm-Berarducci encoding.

This type is equivalent to the following:

       data TubeF a b k
           = Await (a -> k) -- :: (a -> k) -> TubeF a b k
           | Yield (b  , k) -- :: (b  , k) -> TubeF a b k
   

The type signatures for the two value constructors should bear a strong resemblance to the actual type signature of runT. Instead of encoding tubes as structures which build up when composed, a TubeF is a control flow mechanism which picks one of two provided continuations.

People using this library should never have to contend with these details but it is worth mentioning.

Constructors

TubeF

Fields

runT :: forall r. ((a -> k) -> r) -> ((b, k) -> r) -> r

Instances

Functor (TubeF a b)

type Source b m r = forall x. Tube x b m r

A computation which only yields and never awaits

type Sink a m r = forall x. Tube a x m r

A computation which only awaits and never yields.

type Action m r = forall x. Tube x x m r

A computation which neither yields nor awaits

Core infrastructure

run :: FreeT f m a -> m (FreeF f a (FreeT f m a))

run is shorter than runFreeT and who knows, maybe it'll change some day

await :: Monad m => Tube a b m a

Command to wait for a new value upstream

yield :: Monad m => b -> Tube a b m ()

Command to send a value downstream

each :: (Monad m, Foldable t) => t b -> Tube a b m ()

Convert a list to a Source

for :: Monad m => Tube a b m r -> (b -> Tube a c m s) -> Tube a c m r

Enumerate yielded values into a continuation, creating a new Source

(~>) :: Monad m => Tube a b m r -> (b -> Tube a c m s) -> Tube a c m r

Infix version of for

(>-) :: Monad m => Tube a b m r -> (b -> Tube b c m r) -> Tube a c m r

Connect a task to a continuation yielding another task; see ><

(><) :: Monad m => Tube a b m r -> Tube b c m r -> Tube a c m r infixl 3

Compose two tubes into a new tube.

liftT :: (MonadTrans t, Monad m) => FreeT f m a -> t m (FreeF f a (FreeT f m a))

This performs a neat trick: a Tube with a return type a will be turned into a new Tube containing the underlying TubeF value.

In this way the >< and >- functions can replace the () return value with a continuation and recursively traverse the computation until a final result is reached.

Utilities

cat :: Monad m => Tube a a m r

Continuously relays any values it receives. Iteratee identity.

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

Transforms all incoming values according to some function.

drop :: Monad m => Int -> Tube a a m r

Refuses to yield the first n values it receives.

take :: Monad m => Int -> Tube a a m ()

Relay only the first n elements of a stream.

takeWhile :: Monad m => (a -> Bool) -> Tube a a m ()

Terminates the stream upon receiving a value violating the predicate

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

Yields only values satisfying some predicate.

reduce

Arguments

:: Monad m

=> (x -> a -> x)

step function

-> x

initial value

-> (x -> b)

final transformation

-> Source a m ()

stream source

-> m b

Strict left-fold of a stream. Note that the actual return type of the source is not relevant, only the intermediate yield type.

every :: (Foldable t, Monad m) => t b -> Tube a (Maybe b) m ()

Similar to each except it explicitly marks the stream as exhausted

unyield :: Monad m => FreeT (TubeF x b) m () -> m (Maybe (b, FreeT (TubeF x b) m ()))

Taps the next value from a source, maybe.

prompt :: Source String IO ()

Source of Strings from stdin. This is mostly for debugging / ghci example purposes.

display :: Sink String IO ()

Sink for Strings to stdout. This is mostly for debugging / ghci example purposes.

Pump

type Pump a b = CofreeT (PumpF a b)

A Pump is the dual to a Tube: where a Tube is a computation manipulating a stream of values, a Pump can be situated on either end of a tube to both insert values when requested and handle any yielded results.

This module is subject to change before I upload `0.2.0.0` to Hackage.

data PumpF a b k

Constructors

PumpF

Fields

recvF :: (a, k)

sendF :: b -> k

Instances

Functor (PumpF a b)

pump :: Comonad w => w a -> (w a -> (b, w a)) -> (w a -> c -> w a) -> Pump b c w a

Creates a Pump for a Tube using a comonadic seed value, a function to give it more data upon request, and a function to handle any yielded results. . Values received from the Tube may be altered and sent back into the tube, hence this mechanism does act like something of a pump.

send :: Comonad w => Pump a b w r -> b -> Pump a b w r

Send a value into a Pump, effectively re-seeding the stream.

recv :: Comonad w => Pump a b w r -> (a, Pump a b w r)

Pull a value from a Pump, along with the rest of the Pump.

runPump :: (Comonad w, Monad m) => (x -> y -> r) -> Pump a b w x -> Tube a b m y -> m r

Given a suitably matching Tube and Pump, you can use the latter to execute the former.

Re-exports

lift :: MonadTrans t => forall m a. Monad m => m a -> t m a

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

runFreeT :: FreeT f m a -> m (FreeF f a (FreeT f m a))