Folding a subset of a stream using 4.0 protocols

Question

I am trying to understand pipe 4.0 and want to convert some conduit code. Suppose I have an Int s stream, and I would like to skip the first five, and then get the sum from the next 5. Using simple lists, this would be:

 sum . take 5 . drop 5 

In the channel it will be:

 drop 5 isolate 5 =$ fold (+) 0 

Or as a complete program:

 import Data.Conduit import Data.Conduit.List (drop, isolate, fold) import Prelude hiding (drop) main :: IO () main = do res <- mapM_ yield [1..20] $$ do drop 5 isolate 5 =$ fold (+) 0 print res 

However, I'm not quite sure how to do this with pipes.

Answer 1

I have not used Pipes before, but after going through the tutorial, I found it very simple:

 import Pipes import qualified Pipes.Prelude as P nums :: Producer Int IO () nums = each [1..20] process :: Producer Int IO () process = nums >-> (P.drop 5) >-> (P.take 5) result :: IO Int result = P.fold (+) 0 id process main = result >>= print 

UPDATE:

Since there is no “efficient” processing in this example, we can use the Identity monad as the base monad for the channel:

 import Pipes import qualified Pipes.Prelude as P import Control.Monad.Identity nums :: Producer Int Identity () nums = each [1..20] process :: Producer Int Identity () process = nums >-> (P.drop 5) >-> (P.take 5) result :: Identity Int result = P.fold (+) 0 id process main = print $ runIdentity result 

UPDATE 1 :

Below is the solution I came up with (for the gist link comment), but I feel it can be made more elegant.

 fun :: Pipe Int (Int, Int) Identity () fun = do replicateM_ 5 await a <- replicateM 5 await replicateM_ 5 await b <- replicateM 5 await yield (sum a, sum b) main = f $ runIdentity $ P.head $ nums >-> fun where f (Just (a,b)) = print (a,b) f Nothing = print "Not enough data" 

Answer 2

To answer your comment, this still works in the general case. I also posted the same answer on reddit, where you also asked a similar question, but I duplicate the answer here:

 import Pipes import Pipes.Parse import qualified Pipes.Prelude as P main :: IO () main = do res <- (`evalStateT` (each [1..20])) $ do runEffect $ for (input >-> P.take 5) discard P.sum (input >-> P.take 5) print res 

This will generalize to more complex cases that you had in mind.