{-| "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@. The module also includes optimized functions for interoperation: > fromStream :: Monad m => Stream (Of a) m r -> Producer' a m r > toStream :: Monad m => Producer a m r -> Stream (Of a) m r . It is not a drop in replacement for @Pipes.Group@. The only systematic difference is that this simple module omits lenses. It is hoped that this will may make elementary usage easier to grasp. The lenses exported the pipes packages only come into their own with the simple @StateT@ parsing procedure pipes promotes. We are not attempting here to replicate this advanced procedure, but only to make elementary forms of breaking and splitting possible in the simplest possible way. . The @pipes-group@ tutorial is framed as a hunt for a genuinely streaming @threeGroups@, which would collect the first three groups of matching items while never holding more than the present item in memory. 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 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' The new user might look at the examples of splitting, breaking and joining in @Streaming.Prelude@ keeping in mind that @Producer a m r@ is equivalent to @Stream (Of a) m r@. . For the rest, only part of the tutorial that would need revision is the bit at the end about writing explicit @FreeT@ programs. Here one does not proceed by pattern matching, but uses `inspect` in place of `runFreeT` > inspect :: (Monad m, Functor f) => Stream f m r -> m (Either r (f (Stream f m r))) and for construction of a @Stream (Producer a m) m r@, the usual battery of combinators: > wrap :: (Monad m, Functor f) => f (Stream f m r) -> Stream f m r > effect :: (Monad m, Functor f) => m (Stream f m r) -> Stream f m r > yields :: (Monad m, Functor f) => f r -> Stream f m r > lift :: (Monad m, Functor f) => m r -> Stream f m r and so on. -} {-#LANGUAGE RankNTypes, BangPatterns #-} module Streaming.Pipes ( -- * @Streaming@ \/ @Pipes@ interoperation fromStream, toStream, toStreamingByteString, fromStreamingByteString, -- * Splitting a 'Producer' into a connected stream of 'Producer's chunksOf, groups, groupsBy, groupsBy', split, breaks, -- * Rejoining a connected stream of 'Producer's concats, intercalates, -- * Folding over the separate layers of a connected stream of 'Producer's folds, foldsM, -- * Transforming a connected stream of 'Producer's takes, takes', maps, -- * Streaming division of a 'Producer' into two span, break, splitAt, group, groupBy, ) where import Pipes import Streaming hiding (concats, groups, chunksOf) import qualified Streaming.Internal as SI import qualified Pipes.Internal as PI import qualified Pipes.Prelude as P import qualified Pipes as P import qualified Streaming.Prelude as S import Control.Monad (liftM) import Prelude hiding (span, splitAt, break) import qualified Data.ByteString as B import qualified Data.ByteString.Streaming as Q import qualified Data.ByteString.Streaming.Internal as Q {- | Construct an ordinary pipes 'Producer' from a 'Stream' of elements >>> runEffect $ fromStream (S.each [1..3]) >-> P.print 1 2 3 -} fromStream :: Monad m => Stream (Of a) m r -> Producer' a m r fromStream = loop where loop stream = case stream of -- this should be rewritten without constructors SI.Return r -> PI.Pure r SI.Effect m -> PI.M (liftM loop m) SI.Step (a:>rest) -> PI.Respond a (\_ -> loop rest) {-# INLINABLE fromStream #-} {- | Construct a 'Stream' of elements from a @pipes@ 'Producer' >>> S.print $ toStream $ P.each [1..3] 1 2 3 -} toStream :: Monad m => Producer a m r -> Stream (Of a) m r toStream = loop where loop stream = case stream of PI.Pure r -> SI.Return r PI.M m -> SI.Effect (liftM loop m) PI.Respond a f -> SI.Step (a :> loop (f ())) PI.Request x g -> PI.closed x {-# INLINABLE toStream #-} {-| Link the chunks of a producer of bytestrings into a single byte stream -} toStreamingByteString :: Monad m => Producer B.ByteString m r -> Q.ByteString m r toStreamingByteString = loop where loop stream = case stream of PI.Pure r -> Q.Empty r PI.M m -> Q.Go (liftM loop m) PI.Respond a f -> Q.Chunk a (loop (f ())) PI.Request x g -> PI.closed x {-# INLINABLE toStreamingByteString #-} {-| Successively yield the chunks hidden in a byte stream. -} fromStreamingByteString :: Monad m => Q.ByteString m r -> Producer' B.ByteString m r fromStreamingByteString = loop where loop stream = case stream of -- this should be rewritten without constructors Q.Empty r -> PI.Pure r Q.Go m -> PI.M (liftM loop m) Q.Chunk a rest -> PI.Respond a (\_ -> loop rest) {-# INLINABLE fromStreamingByteString #-} {-| 'span' splits a 'Producer' into two 'Producer's; the outer 'Producer' is the longest consecutive group of elements that satisfy the predicate. Its inverse is 'Control.Monad.join' -} span :: Monad m => (a -> Bool) -> Producer a m r -> Producer a m (Producer a m r) span predicate = loop where loop p = do e <- lift (next p) case e of Left r -> return (return r) Right (a, p') -> if predicate a then yield a >> loop p' else return (yield a >> p') {-# INLINABLE span #-} {-| 'break' splits a 'Producer' into two 'Producer's; the outer 'Producer' is the longest consecutive group of elements that fail the predicate. Its inverse is 'Control.Monad.join' -} break :: Monad m => (a -> Bool) -> Producer a m r -> Producer a m (Producer a m r) break predicate = span (not . predicate) {-#INLINE break #-} split :: (Eq a, Monad m) => a -> Producer a m r -> Stream (Producer a m) m r split t = loop where loop stream = do e <- lift $ next stream case e of Left r -> SI.Return r Right (a, p') -> if a /= t then SI.Step $ fmap loop (yield a >> span (/= t) p') else loop p' {-#INLINABLE split #-} breaks :: (Eq a, Monad m) => (a -> Bool) -> Producer a m r -> Stream (Producer a m) m r breaks predicate = loop where loop stream = do e <- lift $ next stream case e of Left r -> SI.Return r Right (a, p') -> if not (predicate a) then SI.Step $ fmap loop (yield a >> break (predicate) p') else loop p' {-#INLINABLE breaks #-} {-| 'splitAt' divides a 'Producer' into two 'Producer's after a fixed number of elements. Its inverse is 'Control.Monad.join' -} splitAt :: Monad m => Int -> Producer a m r -> Producer a m (Producer a m r) splitAt = loop where loop n p | n <= 0 = return p loop n p = do e <- lift (next p) case e of Left r -> return (return r) Right (a, p') -> yield a >> loop (n - 1) p' {-# INLINABLE splitAt #-} {-| 'groupBy' splits a 'Producer' into two 'Producer's; the second producer begins where we meet an element that is different according to the equality predicate. Its inverse is 'Control.Monad.join' -} groupBy :: Monad m => (a -> a -> Bool) -> Producer a m r -> Producer a m (Producer a m r) groupBy equals = loop where loop p = do x <- lift (next p) case x of Left r -> return (return r) Right (a, p') -> span (equals a) (yield a >> p') {-# INLINABLE groupBy #-} -- | Like 'groupBy', where the equality predicate is ('==') group :: (Monad m, Eq a) => Producer a m r -> Producer a m (Producer a m r) group = groupBy (==) {-# INLINABLE group #-} {-| 'groupsBy' splits a 'Producer' into a 'Stream' of 'Producer's of equal items. Its inverse is 'concats' -} groupsBy :: Monad m => (a -> a -> Bool) -> Producer a m r -> Stream (Producer a m) m r groupsBy equals = loop where loop p = SI.Effect $ do e <- next p return $ case e of Left r -> SI.Return r Right (a, p') -> SI.Step (fmap loop (yield a >> span (equals a) p')) {-# INLINABLE groupsBy #-} {-| @groupsBy'@ splits a 'Producer' into a 'Stream' of 'Producer's grouped using the given relation. Its inverse is 'concats' This differs from 'groupsBy' by comparing successive elements instead of comparing each element to the first member of the group >>> import Pipes (yield, each) >>> import Pipes.Prelude (toList) >>> let rel c1 c2 = succ c1 == c2 >>> (toList . intercalates (yield '|') . groupsBy' rel) (each "12233345") "12|23|3|345" >>> (toList . intercalates (yield '|') . groupsBy rel) (each "12233345") "122|3|3|34|5" -} groupsBy' :: Monad m => (a -> a -> Bool) -> Producer a m r -> Stream (Producer a m) m r groupsBy' equals = loop where loop p = SI.Effect $ do e <- next p return $ case e of Left r -> SI.Return r Right (a, p') -> SI.Step (fmap loop (loop0 (yield a >> p'))) loop0 p1 = do e <- lift (next p1) case e of Left r -> return (return r) Right (a2, p2) -> do yield a2 let loop1 a p = do e' <- lift (next p) case e' of Left r -> return (return r) Right (a', p') -> if equals a a' then do yield a' loop1 a' p' else return (yield a' >> p') loop1 a2 p2 {-# INLINABLE groupsBy' #-} groups:: (Monad m, Eq a) => Producer a m r -> Stream (Producer a m) m r groups = groupsBy (==) {-| 'chunksOf' splits a 'Producer' into a 'Stream' of 'Producer's of a given length. Its inverse is 'concats'. >>> let listN n = L.purely P.folds L.list . P.chunksOf n >>> runEffect $ listN 3 P.stdinLn >-> P.take 2 >-> P.map unwords >-> P.print 1<Enter> 2<Enter> 3<Enter> "1 2 3" 4<Enter> 5<Enter> 6<Enter> "4 5 6" >>> let stylish = P.concats . P.maps (<* P.yield "-*-") . P.chunksOf 2 >>> runEffect $ stylish (P.each $ words "one two three four five six") >-> P.stdoutLn one two -*- three four -*- five six -*- -} chunksOf :: Monad m => Int -> Producer a m r -> Stream (Producer a m) m r chunksOf n = loop where loop p = SI.Effect $ do e <- next p return $ case e of Left r -> SI.Return r Right (a, p') -> SI.Step (fmap loop (splitAt n (yield a >> p'))) {-# INLINABLE chunksOf #-} -- | Join a stream of 'Producer's into a single 'Producer' concats :: Monad m => Stream (Producer a m) m r -> Producer a m r concats = loop where loop stream = case stream of SI.Return r -> return r SI.Effect m -> PI.M $ liftM loop m SI.Step p -> do rest <- p loop rest {-# INLINABLE concats #-} -- {-| Join a 'FreeT'-delimited stream of 'Producer's into a single 'Producer' by -- intercalating a 'Producer' in between them -- -} -- intercalates -- :: Monad m => Producer a m () -> Stream (Producer a m) m r -> Producer a m r -- intercalates sep = loop where -- loop stream = case stream of -- -- x <- lift (runFreeT f) -- case x of -- Pure r -> return r -- Free p -> do -- f' <- p -- go1 f' -- go1 f = do -- x <- lift (runFreeT f) -- case x of -- Pure r -> return r -- Free p -> do -- sep -- f' <- p -- go1 f' -- {-# INLINABLE intercalates #-} {- $folds These folds are designed to be compatible with the @foldl@ library. See the 'Control.Foldl.purely' and 'Control.Foldl.impurely' functions from that library for more details. For example, to count the number of 'Producer' layers in a 'Stream', you can write: > import Control.Applicative (pure) > import qualified Control.Foldl as L > import Pipes.Group > import qualified Pipes.Prelude as P > > count :: Monad m => Stream (Producer a m) m () -> m Int > count = P.sum . L.purely folds (pure 1) -} {-| Fold each 'Producer' in a producer 'Stream' > purely folds > :: Monad m => Fold a b -> Stream (Producer a m) m r -> Producer b m r -} folds :: Monad m => (x -> a -> x) -- ^ Step function -> x -- ^ Initial accumulator -> (x -> b) -- ^ Extraction function -> Stream (Producer a m) m r -- ^ -> Producer b m r folds step begin done = loop where loop stream = case stream of SI.Return r -> return r SI.Effect m -> PI.M $ liftM loop m SI.Step p -> do (stream', b) <- lift (fold p begin) yield b loop stream' fold p x = do y <- next p case y of Left f -> return (f, done x) Right (a, p') -> fold p' $! step x a {-# INLINABLE folds #-} {-| 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 -} foldsM :: 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 foldsM step begin done = loop where loop stream = case stream of SI.Return r -> return r SI.Effect m -> PI.M (liftM loop m) SI.Step p -> do (f', b) <- lift $ begin >>= foldM p yield b loop f' foldM p x = do y <- next p case y of Left f -> do b <- done x return (f, b) Right (a, p') -> do x' <- step x a foldM p' $! x' {-# INLINABLE foldsM #-} {-| @(takes' n)@ only keeps the first @n@ 'Producer's 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 'Producer's in order to preserve the return value. This makes it a suitable argument for 'maps'. -} takes' :: Monad m => Int -> Stream (Producer a m) m r -> Stream (Producer a m) m r takes' = loop where loop !n stream | n <= 0 = drain_loop stream loop n stream = case stream of SI.Return r -> SI.Return r SI.Effect m -> SI.Effect (liftM (loop n) m) SI.Step p -> SI.Step (fmap (loop (n - 1)) p) drain_loop stream = case stream of SI.Return r -> SI.Return r SI.Effect m -> SI.Effect (liftM drain_loop m) SI.Step p -> SI.Effect $ do stream' <- runEffect (P.for p P.discard) return $ drain_loop stream' {-# INLINABLE takes' #-}