packages feed

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