Safe Haskell | Trustworthy |
---|---|

Language | Haskell2010 |

- data Stream f m r
- unfold :: (Monad m, Functor f) => (s -> m (Either r (f s))) -> s -> Stream f m r
- replicates :: (Monad m, Functor f) => Int -> f () -> Stream f m ()
- repeats :: (Monad m, Functor f) => f () -> Stream f m r
- repeatsM :: (Monad m, Functor f) => m (f ()) -> Stream f m r
- effect :: (Monad m, Functor f) => m (Stream f m r) -> Stream f m r
- wrap :: (Monad m, Functor f) => f (Stream f m r) -> Stream f m r
- yields :: (Monad m, Functor f) => f r -> Stream f m r
- streamBuild :: (forall b. (r -> b) -> (m b -> b) -> (f b -> b) -> b) -> Stream f m r
- cycles :: (Monad m, Functor f) => Stream f m () -> Stream f m r
- delays :: (MonadIO m, Applicative f) => Double -> Stream f m r
- never :: (Monad m, Applicative f) => Stream f m r
- untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r
- intercalates :: (Monad m, Monad (t m), MonadTrans t) => t m x -> Stream (t m) m r -> t m r
- concats :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r
- iterT :: (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a
- iterTM :: (Functor f, Monad m, MonadTrans t, Monad (t m)) => (f (t m a) -> t m a) -> Stream f m a -> t m a
- destroy :: (Functor f, Monad m) => Stream f m r -> (f b -> b) -> (m b -> b) -> (r -> b) -> b
- streamFold :: (Functor f, Monad m) => (r -> b) -> (m b -> b) -> (f b -> b) -> Stream f m r -> b
- inspect :: (Functor f, Monad m) => Stream f m r -> m (Either r (f (Stream f m r)))
- maps :: (Monad m, Functor f) => (forall x. f x -> g x) -> Stream f m r -> Stream g m r
- mapsM :: (Monad m, Functor f) => (forall x. f x -> m (g x)) -> Stream f m r -> Stream g m r
- decompose :: (Monad m, Functor f) => Stream (Compose m f) m r -> Stream f m r
- mapsM_ :: (Functor f, Monad m) => (forall x. f x -> m x) -> Stream f m r -> m r
- run :: Monad m => Stream m m r -> m r
- distribute :: (Monad m, Functor f, MonadTrans t, MFunctor t, Monad (t (Stream f m))) => Stream f (t m) r -> t (Stream f m) r
- groups :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r -> Stream (Sum (Stream f m) (Stream g m)) m r
- chunksOf :: (Monad m, Functor f) => Int -> Stream f m r -> Stream (Stream f m) m r
- splitsAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r)
- takes :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m ()
- cutoff :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Maybe 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
- zips :: (Monad m, Functor f, Functor g) => Stream f m r -> Stream g m r -> Stream (Compose f g) m r
- unzips :: (Monad m, Functor f, Functor g) => Stream (Compose f g) m r -> Stream f (Stream g m) r
- interleaves :: (Monad m, Applicative h) => Stream h m r -> Stream h m r -> Stream h m r
- separate :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r -> Stream f (Stream g m) r
- unseparate :: (Monad m, Functor f, Functor g) => Stream f (Stream g m) r -> Stream (Sum f g) m r
- switch :: Sum f g r -> Sum g f r
- bracketStream :: (Functor f, MonadResource m) => IO a -> (a -> IO ()) -> (a -> Stream f m b) -> Stream f m b
- unexposed :: (Functor f, Monad m) => Stream f m r -> Stream f m r
- hoistExposed :: (Monad m1, Functor f) => (m1 (Stream f m r) -> m (Stream f m r)) -> Stream f m1 r -> Stream f m r
- mapsExposed :: (Monad m, Functor f) => (forall x. f x -> g x) -> Stream f m r -> Stream g m r
- mapsMExposed :: (Monad m, Functor f1) => (f1 (Stream f m r) -> m (f (Stream f m r))) -> Stream f1 m r -> Stream f m r
- destroyExposed :: (Monad m, Functor f) => Stream f m t -> (f b -> b) -> (m b -> b) -> (t -> b) -> b

# The 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.

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

The *producer* concept uses the simple functor ` (a,_) `

- or the stricter
` Of a _ `

. Then the news at each step or layer is just: an individual item of type `a`

.
Since `Stream (Of a) m r`

is equivalent to `Pipe.Producer a m r`

, much of
the `pipes`

`Prelude`

can easily be mirrored in a `streaming`

`Prelude`

. Similarly,
a simple `Consumer a m r`

or `Parser a m r`

concept arises when the base functor is
` (a -> _) `

. `Stream ((->) input) m result`

consumes `input`

until it returns a
`result`

.

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`

The constructors are exported by the `Internal`

module.

# Introducing a stream

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

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

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

Repeat a functorial layer, command or instruction a fixed number of times.

replicates n = takes n . repeats

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

Repeat a functorial layer (a "command" or "instruction") forever.

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

Repeat an effect containing a functorial layer, command or instruction forever.

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

Wrap an effect that returns a stream

effect = join . lift

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

`yields`

is like `lift`

for items in the streamed functor.
It makes a singleton or one-layer succession.

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

Viewed in another light, it is like a functor-general version of `yield`

:

S.yield a = yields (a :> ())

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

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

streamFold return_ effect_ step_ (streamBuild psi) = psi return_ effect_ step_

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

Construct an infinite stream by cycling a finite one

cycles = forever

`>>>`

never :: (Monad m, Applicative f) => Stream f m r Source

`never`

interleaves the pure applicative action with the return of the monad forever.
It is the `empty`

of the `Alternative`

instance, thus

never <|> a = a a <|> never = a

and so on. If w is a monoid then `never :: Stream (Of w) m r`

is
the infinite sequence of `mempty`

, and
`str1 <|> str2`

appends the elements monoidally until one of streams ends.
Thus we have, e.g.

`>>>`

1<Enter> 2<Enter> 3<Enter> 1 2 3 4<Enter> 5<Enter> 6<Enter> 4 5 6`S.stdoutLn $ S.take 2 $ S.stdinLn <|> S.repeat " " <|> S.stdinLn <|> S.repeat " " <|> S.stdinLn`

This is equivalent to

`>>>`

`S.stdoutLn $ S.take 2 $ foldr (<|>) never [S.stdinLn, S.repeat " ", S.stdinLn, S.repeat " ", S.stdinLn ]`

Where `f`

is a monad, `(<|>)`

sequences the conjoined streams stepwise. See the
definition of `paste`

here,
where the separate steps are bytestreams corresponding to the lines of a file.

Given, say,

data Branch r = Branch r r deriving Functor -- add obvious applicative instance

then `never :: Stream Branch Identity r`

is the pure infinite binary tree with
(inaccessible) `r`

s in its leaves. Given two binary trees, `tree1 <|> tree2`

intersects them, preserving the leaves that came first,
so `tree1 <|> never = tree1`

`Stream Identity m r`

is an action in `m`

that is indefinitely delayed. Such an
action can be constructed with e.g. `untilJust`

.

untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r

Given two such items, `<|>`

instance races them.
It is thus the iterative monad transformer specially defined in
Control.Monad.Trans.Iter

So, for example, we might write

`>>>`

`let justFour str = if length str == 4 then Just str else Nothing`

`>>>`

`let four = untilJust (liftM justFour getLine)`

`>>>`

one<Enter> two<Enter> three<Enter> four<Enter> "four"`run four`

The `Alternative`

instance in
Control.Monad.Trans.Free
is avowedly wrong, though no explanation is given for this.

# Eliminating a stream

intercalates :: (Monad m, Monad (t m), MonadTrans t) => t m x -> Stream (t m) m r -> t m r Source

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 Source

Dissolves the segmentation into layers of `Stream f m`

layers.

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

Specialized fold following the usage of `Control.Monad.Trans.Free`

iterT alg = streamFold return join alg

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

Specialized fold following the usage of `Control.Monad.Trans.Free`

iterTM alg = streamFold return (join . lift)

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

Map a stream directly to its church encoding; compare `Data.List.foldr`

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

`streamFold`

reorders the arguments of `destroy`

to be more akin
to `foldr`

It is more convenient to query in ghci to figure out
what kind of 'algebra' you need to write.

`>>>`

(Monad m, Functor f) => (f (m a) -> m a) -> Stream f m a -> m a -- iterT`:t streamFold return join`

`>>>`

(Monad m, Monad (t m), Functor f, MonadTrans t) => (f (t m a) -> t m a) -> Stream f m a -> t m a -- iterTM`:t streamFold return (join . lift)`

`>>>`

(Monad m, Functor f, Functor g) => (f (Stream g m r) -> Stream g m r) -> Stream f m r -> Stream g m r`:t streamFold return effect`

`>>>`

(Monad m, Functor f, Functor g) => (f (Stream g m a) -> g (Stream g m a)) -> Stream f m a -> Stream g m a -- maps`:t \f -> streamFold return effect (wrap . f)`

`>>>`

(Monad m, Functor f, Functor g) => (f (Stream g m a) -> m (g (Stream g m a))) -> Stream f m a -> Stream g m a -- mapped`:t \f -> streamFold return effect (effect . liftM wrap . f)`

# Inspecting a stream wrap by wrap

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

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

# Transforming streams

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

Map layers of one functor to another with a transformation. Compare
hoist, which has a similar effect on the `monadic`

parameter.

maps id = id maps f . maps g = maps (f . g)

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

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

is more fundamental than `mapsM`

, which is best understood as a convenience
for effecting this frequent composition:

mapsM phi = decompose . maps (Compose . phi)

The streaming prelude exports the same function under the better name `mapped`

,
which overlaps with the lens libraries.

decompose :: (Monad m, Functor f) => Stream (Compose m f) m r -> Stream f m r Source

Rearrange a succession of layers of the form `Compose m (f x)`

.

we could as well define `decompose`

by `mapsM`

:

decompose = mapped getCompose

but `mapped`

is best understood as:

mapped phi = decompose . maps (Compose . phi)

since `maps`

and `hoist`

are the really fundamental operations that preserve the
shape of the stream:

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

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

Map each layer to an effect, and run them all.

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

Make it possible to 'run' the underlying transformed monad.

groups :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r -> Stream (Sum (Stream f m) (Stream g m)) m r Source

Group layers in an alternating stream into adjoining sub-streams of one type or another.

# Splitting streams

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

Break a stream into substreams each with n functorial layers.

`>>>`

2 2 1`S.print $ mapped S.sum $ chunksOf 2 $ each [1,1,1,1,1]`

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

Split a succession of layers after some number, returning a streaming or effectful pair.

`>>>`

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

`>>>`

2 3`S.print rest`

splitAt 0 = return splitAt n >=> splitAt m = splitAt (m+n)

Thus, e.g.

`>>>`

1 2 3 4`rest <- S.print $ splitsAt 2 >=> splitsAt 2 $ each [1..5]`

`>>>`

5`S.print rest`

# Zipping and unzipping streams

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 Source

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

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

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

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))`

`>>>`

hello goodbye world world`Q.stdout $ Q.unlines $ paste "hello\nworld\n" "goodbye\nworld\n"`

separate :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r -> Stream f (Stream g m) r Source

Given a stream on a sum of functors, make it a stream on the left functor,
with the streaming on the other functor as the governing monad. This is
useful for acting on one or the other functor with a fold. It generalizes
`partitionEithers`

massively, but actually streams properly.

`>>>`

`let odd_even = S.maps (S.distinguish even) $ S.each [1..10::Int]`

`>>>`

separate odd_even :: Monad m => Stream (Of Int) (Stream (Of Int) m) ()`:t separate odd_even`

Now, for example, it is convenient to fold on the left and right values separately:

`>>>`

[2,4,6,8,10] :> ([1,3,5,7,9] :> ())`S.toList $ S.toList $ separate odd_even`

Or we can write them to separate files or whatever:

`>>>`

`runResourceT $ S.writeFile "even.txt" . S.show $ S.writeFile "odd.txt" . S.show $ S.separate odd_even`

`>>>`

2 4 6 8 10`:! cat even.txt`

`>>>`

1 3 5 7 9`:! cat odd.txt`

Of course, in the special case of `Stream (Of a) m r`

, we can achieve the above
effects more simply by using `copy`

`>>>`

[2,4,6,8,10] :> ([1,3,5,7,9] :> ())`S.toList . S.filter even $ S.toList . S.filter odd $ S.copy $ each [1..10::Int]`

But `separate`

and `unseparate`

are functor-general.

unseparate :: (Monad m, Functor f, Functor g) => Stream f (Stream g m) r -> Stream (Sum f g) m r Source

# Assorted Data.Functor.x help

switch :: Sum f g r -> Sum g f r Source

Swap the order of functors in a sum of functors.

`>>>`

'a' 'a' 'a' "bnn" :> ()`S.toListM' $ S.print $ separate $ maps S.switch $ maps (S.distinguish (=='a')) $ S.each "banana"`

`>>>`

'b' 'n' 'n' "aaa" :> ()`S.toListM' $ S.print $ separate $ maps (S.distinguish (=='a')) $ S.each "banana"`

# ResourceT help

bracketStream :: (Functor f, MonadResource m) => IO a -> (a -> IO ()) -> (a -> Stream f m b) -> Stream f m b Source

# For use in implementation

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

This is akin to the `observe`

of `Pipes.Internal`

. It reeffects the layering
in instances of `Stream f m r`

so that it replicates that of
`FreeT`

.

hoistExposed :: (Monad m1, Functor f) => (m1 (Stream f m r) -> m (Stream f m r)) -> Stream f m1 r -> Stream f m r Source

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

mapsMExposed :: (Monad m, Functor f1) => (f1 (Stream f m r) -> m (f (Stream f m r))) -> Stream f1 m r -> Stream f m r Source

destroyExposed :: (Monad m, Functor f) => Stream f m t -> (f b -> b) -> (m b -> b) -> (t -> b) -> b Source