Streaming.Pipes

Pipes.Group.Tutorial is the correct introduction to the use of this module, which is mostly just an optimized Pipes.Group, replacing FreeT with Stream. (See the introductory documentation for this package. The pipes-group tutorial is framed as a hunt for a genuinely streaming threeGroups. The formulation it opts for in the end would be expressed here thus:

import Pipes
import Streaming.Pipes 
import qualified Pipes.Prelude as P

threeGroups :: (Monad m, Eq a) => Producer a m () -> Producer a m ()
threeGroups = concats . takes 3 . groups

The only difference is that this simple module omits the detour via lenses. The program splits the initial producer into a connected stream of producers containing "equal" values; it takes three of those; and then erases the effects of splitting. So for example

>>> runEffect $ threeGroups (each "aabccoooooo") >-> P.print
'a'
'a'
'b'
'c'
'c'

For the rest, only part of the tutorial that would need revision is the bit at the end about writing explicit FreeT programs. Its examples use pattern matching, but the constructors of the Stream type are necessarily hidden, so one would have replaced by the various inspection combinators provided by the streaming library.

Streaming / Pipes interoperation

fromStream :: Monad m => Stream (Of a) m r -> Producer' a m r

Construct an ordinary pipes Producer from a Stream of elements

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

Construct a Stream of elements from a pipes Producer

toStreamingByteString :: Monad m => Producer ByteString m r -> ByteString m r

fromStreamingByteString :: Monad m => ByteString m r -> Producer' ByteString m r

Transforming a connected stream of Producers

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

takes' :: Monad m => Int -> Stream (Producer a m) m r -> Stream (Producer a m) m r

(takes' n) only keeps the first n Producers of a linked Stream of Producers

Unlike takes, takes' is not functor-general - it is aware that a Producer can be drained, as functors cannot generally be. Here, then, we drain the unused Producers in order to preserve the return value. This makes it a suitable argument for maps.

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. Compare hoist, which has a similar effect on the monadic parameter.

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

Streaming division of a Producer into two

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

span splits a Producer into two Producers; the outer Producer is the longest consecutive group of elements that satisfy the predicate. Its inverse is join

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

splitAt :: Monad m => Int -> Producer a m r -> Producer a m (Producer a m r)

splitAt divides a Producer into two Producers after a fixed number of elements. Its inverse is join

group :: (Monad m, Eq a) => Producer a m r -> Producer a m (Producer a m r)

Like groupBy, where the equality predicate is (==)

groupBy :: Monad m => (a -> a -> Bool) -> Producer a m r -> Producer a m (Producer a m r)

groupBy splits a Producer into two Producers; the second producer begins where we meet an element that is different according to the equality predicate. Its inverse is join

Splitting a Producer into a connected stream of Producers

groupsBy :: Monad m => (a -> a -> Bool) -> Producer a m r -> Stream (Producer a m) m r

groupsBy' :: Monad m => (a -> a -> Bool) -> Producer a m r -> Stream (Producer a m) m r

groupsBy' splits a Producer into a Stream of Producers grouped using the given equality predicate

This differs from groupsBy by comparing successive elements for equality instead of comparing each element to the first member of the group

>>> import Pipes (yield, each)
>>> import Pipes.Prelude (toList)
>>> let cmp c1 c2 = succ c1 == c2
>>> (toList . intercalates (yield '|') . groupsBy' cmp) (each "12233345")
"12|23|3|345"
>>> (toList . intercalates (yield '|') . groupsBy  cmp) (each "12233345")
"122|3|3|34|5"

groups :: (Monad m, Eq a) => Producer a m r -> Stream (Producer a m) m r

split :: (Eq a, Monad m) => a -> Producer a m r -> Stream (Producer a m) m r

breaks :: (Eq a, Monad m) => (a -> Bool) -> Producer a m r -> Stream (Producer a m) m r

Rejoining a connected stream of Producers

concats :: Monad m => Stream (Producer a m) m r -> Producer a m r

Join a stream of Producers into a single Producer

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

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

Folding over the separate layers of a connected stream of Producers

folds

Arguments

:: Monad m

=> (x -> a -> x)

Step function

-> x

Initial accumulator

-> (x -> b)

Extraction function

-> Stream (Producer a m) m r

-> Producer b m r

Fold each Producer in a producer Stream

purely folds
    :: Monad m => Fold a b -> Stream (Producer a m) m r -> Producer b m r

foldsM

Arguments

:: Monad m

=> (x -> a -> m x)

Step function

-> m x

Initial accumulator

-> (x -> m b)

Extraction function

-> Stream (Producer a m) m r

-> Producer b m r

Fold each Producer in a Producer stream, monadically

impurely foldsM
    :: Monad m => FoldM a b -> Stream (Producer a m) m r -> Producer b m r