pipes-extra (empty) → 0.0.1
raw patch · 8 files changed
+384/−0 lines, 8 filesdep +basedep +bytestringdep +mtlsetup-changed
Dependencies added: base, bytestring, mtl, pipes-core
Files
- Control/Pipe/Binary.hs +119/−0
- Control/Pipe/ChunkPipe.hs +29/−0
- Control/Pipe/Coroutine.hs +46/−0
- Control/Pipe/Tee.hs +48/−0
- Control/Pipe/Zip.hs +80/−0
- LICENSE +30/−0
- Setup.hs +2/−0
- pipes-extra.cabal +30/−0
+ Control/Pipe/Binary.hs view
@@ -0,0 +1,119 @@+{-# LANGUAGE ScopedTypeVariables #-}+module Control.Pipe.Binary (+ -- ** Handle and File IO+ fileReader,+ handleReader,+ fileWriter,+ handleWriter,++ -- ** Chunked Byte Stream Manipulation+ take,+ takeWhile,+ dropWhile,+ lines,+ bytes,+ ) where++import Control.Monad+import Control.Monad.Trans (MonadIO, liftIO, lift)+import Control.Pipe+import Control.Pipe.Exception+import Control.Pipe.Combinators (tryAwait, feed)+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as BC+import Data.Monoid+import Data.Word+import System.IO+import Prelude hiding (take, takeWhile, dropWhile, lines, catch)++-- | Read data from a file.+fileReader :: MonadIO m => FilePath -> Pipe () B.ByteString m ()+fileReader path = bracket+ (liftIO $ openFile path ReadMode)+ (liftIO . hClose)+ handleReader++-- | Read data from an open handle.+handleReader :: MonadIO m => Handle -> Pipe () B.ByteString m ()+handleReader h = go+ where+ go = do+ eof <- lift . liftIO $ hIsEOF h+ unless eof $ do+ chunk <- lift . liftIO $ B.hGetSome h 4096+ -- lift . liftIO . putStrLn $ "chunk size " ++ show (B.length chunk)+ yield chunk+ go++-- | Write data to a file.+--+-- The file is only opened if some data arrives into the pipe.+fileWriter :: MonadIO m => FilePath -> Pipe B.ByteString Void m ()+fileWriter path = do+ -- receive some data before opening the handle+ input <- await+ -- feed it to the actual worker pipe+ feed input go+ where+ go = bracket+ (liftIO $ openFile path WriteMode)+ (liftIO . hClose)+ handleWriter++-- | Write data to a handle.+handleWriter:: MonadIO m => Handle -> Pipe B.ByteString Void m ()+handleWriter h = forever $ do+ chunk <- await+ -- lift . liftIO . putStrLn $ "writing chunk " ++ show (B.length chunk)+ lift . liftIO . B.hPut h $ chunk++-- | Act as an identity for the first 'n' bytes, then terminate returning the+-- unconsumed portion of the last chunk.+take :: Monad m => Int -> Pipe B.ByteString B.ByteString m B.ByteString+take size = do+ chunk <- await+ let (chunk', leftover) = B.splitAt size chunk+ yield chunk'+ if B.null leftover+ then take $ size - B.length chunk'+ else return leftover++-- | Act as an identity as long as the given predicate holds, then terminate+-- returning the unconsumed portion of the last chunk.+takeWhile :: Monad m => (Word8 -> Bool) -> Pipe B.ByteString B.ByteString m B.ByteString+takeWhile p = go+ where+ go = do+ chunk <- await+ let (chunk', leftover) = B.span p chunk+ unless (B.null chunk) $ yield chunk'+ if B.null leftover+ then go+ else return leftover++-- | Drop bytes as long as the given predicate holds, then act as an identity.+dropWhile :: Monad m => (Word8 -> Bool) -> Pipe B.ByteString B.ByteString m r+dropWhile p = do+ leftover <- takeWhile (not . p) >+> discard+ yield leftover+ idP++-- | Split the chunked input stream into lines, and yield them individually.+lines :: Monad m => Pipe B.ByteString B.ByteString m r+lines = go B.empty+ where+ go leftover = do+ mchunk <- tryAwait+ case mchunk of+ Nothing -> yield leftover >> idP+ Just chunk -> split chunk leftover+ split chunk leftover+ | B.null chunk = go leftover+ | B.null rest = go (mappend leftover chunk)+ | otherwise = yield (mappend leftover line) >>+ split (B.drop 1 rest) mempty+ where (line, rest) = B.breakByte 10 chunk++-- | Yield individual bytes of the chunked input stream.+bytes :: Monad m => Pipe B.ByteString Word8 m r+bytes = forever $ await >>= B.foldl (\p c -> p >> yield c) (return ())
+ Control/Pipe/ChunkPipe.hs view
@@ -0,0 +1,29 @@+-- | This module contains utilities to create and combine pipes that accept+-- "chunked" input and return unconsumed portions of their internal buffer.+--+-- The main interface is an alternative monad instance for Pipe, which passes+-- leftover data along automatically.+module Control.Pipe.ChunkPipe (+ ChunkPipe(..),+ nonchunked,+ ) where++import Control.Pipe+import Data.Monoid++-- | Newtype wrapper for Pipe proving a monad instance that takes care of+-- passing leftover data automatically.+--+-- An individual 'ChunkPipe' is just a regular pipe, but returns unconsumed+-- input in a pair alongside the actual return value.+newtype ChunkPipe a b m r = ChunkPipe { unChunkPipe :: Pipe a b m (a, r) }++instance (Monoid a, Monad m) => Monad (ChunkPipe a b m) where+ return = nonchunked . return+ (ChunkPipe p) >>= f = ChunkPipe $ p >>= \(leftover, result) ->+ (yield leftover >> idP) >+> unChunkPipe (f result)++-- | Create a 'ChunkPipe' out of a regular pipe that is able to consume all its+-- input.+nonchunked :: (Monoid a, Monad m) => Pipe a b m r -> ChunkPipe a b m r+nonchunked p = ChunkPipe $ p >>= \r -> return (mempty, r)
+ Control/Pipe/Coroutine.hs view
@@ -0,0 +1,46 @@+module Control.Pipe.Coroutine (+ Coroutine,+ coroutine,+ suspend,+ suspendE,+ resume,+ step,+ terminate+ ) where++import Control.Monad+import Control.Pipe+import qualified Control.Exception as E++data Coroutine a b m r = Coroutine+ { suspend :: Pipe a b m r+ , suspendE :: E.SomeException -> Pipe a b m r }++resume :: Monad m+ => Pipe a b m r+ -> Pipe a x m (Either r (b, Coroutine a b m r))+resume (Pure r) = return $ Left r+resume (Throw e) = throwP e+resume (Free c h) = go c >>= \x -> case x of+ Left p -> resume p+ Right (b, p) -> return $ Right (b, Coroutine p h)+ where+ go (Await k) = liftM (Left . k) await+ go (Yield b p) = return $ Right (b, p)+ go (M m s) = liftM Left $ liftP s m++coroutine :: Monad m+ => Pipe a b m r+ -> Coroutine a b m r+coroutine p = Coroutine p throwP++step :: Monad m+ => Coroutine a b m r+ -> Pipe a x m (Either r (b, Coroutine a b m r))+step = resume . suspend++terminate :: Monad m+ => Coroutine a b m r+ -> Pipe a x m ()+terminate (Coroutine _ h) =+ void (catchP discard h) >+> return ()
+ Control/Pipe/Tee.hs view
@@ -0,0 +1,48 @@+-- |+-- Module : Control.Pipe.Tee+-- Copyright : (c) 2012 Jeremy Shaw+-- (c) 2012 Paolo Capriotti+-- License : BSD3+--+-- Maintainer : jeremy@n-heptane.com+-- Stability : experimental+-- Portability : portable+--+-- The 'tee' combinators act like 'idP', but also send a copy of the input to the+-- supplied 'Consumer' (e.g. a file). This is typically done for the purpose of+-- logging a pipeline, showing progress, etc.+module Control.Pipe.Tee (+ tee,+ teeFile,+ teeFileBS+ ) where++import Control.Monad.Trans (MonadIO(..))+import Control.Pipe+import Control.Pipe.Binary (fileWriter)+import Data.ByteString (ByteString)++-- | Acts like 'idP', but also passes a copy to the supplied consumer.+tee :: (Monad m)+ => Pipe a Void m r -- ^ 'Consumer' that will receive a copy of all the input+ -> Pipe a a m r+tee consumer =+ splitP >+> firstP consumer >+> discardL++-- | Acts like 'idP', but also writes a copy to a file.+teeFile :: (MonadIO m)+ => (a -> ByteString) -- ^ function to convert the value to a+ -- 'ByteString' which can be written to+ -- the log+ -> FilePath -- ^ file to log to+ -> Pipe a a m ()+teeFile showBS logFile =+ tee (pipe showBS >+> fileWriter logFile)++-- | Acts like, 'idP', but also writes a copy to the specified log file.+--+-- This function is equivalent to @teeFile id@.+teeFileBS :: (MonadIO m)+ => FilePath -- ^ file to log to+ -> Pipe ByteString ByteString m ()+teeFileBS = teeFile id
+ Control/Pipe/Zip.hs view
@@ -0,0 +1,80 @@+module Control.Pipe.Zip (+ controllable,+ controllable_,+ zip,+ zip_,+ ProducerControl(..),+ ZipControl(..),+ ) where++import qualified Control.Exception as E+import Control.Monad+import Control.Pipe+import Control.Pipe.Coroutine+import Control.Pipe.Exception+import Prelude hiding (zip)++data ProducerControl r+ = Done r+ | Error E.SomeException++controllable :: Monad m+ => Producer a m r+ -> Pipe (Either () (ProducerControl r)) a m r+controllable p = do+ x <- pipe (const ()) >+> resume p+ case x of+ Left r -> return r+ Right (b, p') ->+ join $ onException+ (await >>= \c -> case c of+ Left () -> yield b >> return (controllable (suspend p'))+ Right (Done r) -> return $ (pipe (const ()) >+> terminate p') >> return r+ Right (Error e) -> return $ controllable (suspendE p' e))+ (pipe (const ()) >+> terminate p')++controllable_ :: Monad m+ => Producer a m r+ -> Producer a m r+controllable_ p = pipe Left >+> controllable p++data ZipControl r+ = LeftZ (ProducerControl r)+ | RightZ (ProducerControl r)++zip :: Monad m+ => Producer a m r+ -> Producer b m r+ -> Pipe (Either () (ZipControl r)) (Either a b) m r+zip p1 p2 = translate >+> (controllable p1 *+* controllable p2)+ where+ translate = forever $ await >>= \c -> case c of+ Left () -> (yield . Left . Left $ ()) >> (yield . Right . Left $ ())+ Right (LeftZ c) -> (yield . Left . Right $ c) >> (yield . Right . Left $ ())+ Right (RightZ c) -> (yield . Left . Left $ ()) >> (yield . Right . Right $ c)++zip_ :: Monad m+ => Producer a m r+ -> Producer b m r+ -> Producer (Either a b) m r+zip_ p1 p2 = pipe Left >+> zip p1 p2++(*+*) :: Monad m+ => Pipe a b m r+ -> Pipe a' b' m r+ -> Pipe (Either a a') (Either b b') m r+p1 *+* p2 = (continue p1 *** continue p2) >+> both+ where+ continue p = do+ r <- p >+> pipe Right+ yield $ Left r+ discard+ both = await >>= \x -> case x of+ Left c -> either (const right) (\a -> yield (Left a) >> both) c+ Right c -> either (const left) (\b -> yield (Right b) >> both) c+ left = await >>= \x -> case x of+ Left c -> either return (\a -> yield (Left a) >> left) c+ Right _ -> left+ right = await >>= \x -> case x of+ Left _ -> right+ Right c -> either return (\b -> yield (Right b) >> right) c
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c)2012, Paolo Capriotti++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Paolo Capriotti nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ pipes-extra.cabal view
@@ -0,0 +1,30 @@+Name: pipes-extra+Version: 0.0.1+License: BSD3+License-file: LICENSE+Author: Paolo Capriotti+Maintainer: p.capriotti@gmail.com+Stability: Experimental+Homepage: https://github.com/pcapriotti/pipes-extra+Category: Control, Enumerator+Build-type: Simple+Synopsis: Various basic utilities for Pipes.+Description: This module contains basic utilities for Pipes to deal with files and chunked binary data, as well as extra combinators like 'zip' and 'tee'.+Cabal-version: >=1.8++Source-Repository head+ Type: git+ Location: https://github.com/pcapriotti/pipes-extra++Library+ Build-Depends:+ base (== 4.*),+ mtl,+ pipes-core (== 0.0.*),+ bytestring (== 0.9.*)+ Exposed-Modules:+ Control.Pipe.Binary,+ Control.Pipe.Coroutine,+ Control.Pipe.ChunkPipe,+ Control.Pipe.Tee,+ Control.Pipe.Zip