process-streaming 0.3.0.0 → 0.5.0.0
raw patch · 7 files changed
+1025/−649 lines, 7 filesdep +attoparsecdep +bytestringdep +containersdep −profunctorsdep ~basedep ~mtldep ~pipes
Dependencies added: attoparsec, bytestring, containers, directory, lens, pipes-attoparsec, pipes-group, pipes-parse, process-streaming, semigroups, tasty, tasty-hunit, void
Dependencies removed: profunctors
Dependency ranges changed: base, mtl, pipes, pipes-bytestring, pipes-text, transformers
Files
- CHANGELOG +18/−0
- README.md +5/−7
- process-streaming.cabal +50/−11
- src/System/Process/Lens.hs +76/−6
- src/System/Process/Streaming.hs +621/−359
- src/System/Process/Streaming/Tutorial.hs +0/−266
- tests/test.hs +255/−0
CHANGELOG view
@@ -1,3 +1,21 @@+0.5.0.0 +------- + +- Now the constructors for "PipingPolicy" take "Siphons" directly, instead of +continuation functions. + +- Removed "separated" and "combined" functions, added new "PipingPolicy" +constructors in their place. + +- Removed "LeftoverPolicy", its function is now performed by Siphons. + +- Removed "surely", "safely", "monoidally" functions with confusing +signatures. + +- Removed all occurrences of unbounded buffers in the code. + +- Implemented support for branching pipelines of processes. + 0.3.0.0 -------
README.md view
@@ -6,15 +6,13 @@ The basic goals: -- Concurrent, buffered, streaming access to stdin, stdout and stderr. - -- Easy integration with parsers pipes-parse package and folds from - Pipes.Prelude. +- Concurrent, streaming access to stdin, stdout and stderr. -- Avoid launching exceptions: use Either or similar solution to signal error - conditions. +- Easy integration with regular consumers, parsers from pipes-parse and various + folds. -- Avoid deadlock scenarios caused by full output buffers. +- Avoid launching exceptions: use Either or similar solution to signal non-IO + related error conditions. Relevant thread in the Haskell Pipes Google Group:
process-streaming.cabal view
@@ -1,5 +1,5 @@ name: process-streaming -version: 0.3.0.0 +version: 0.5.0.0 license: BSD3 license-file: LICENSE data-files: @@ -16,30 +16,69 @@ CHANGELOG Library + default-language: Haskell2010 hs-source-dirs: src exposed-modules: System.Process.Streaming - System.Process.Streaming.Tutorial System.Process.Lens other-modules: build-depends: - base >= 4.4 && < 4.8, - transformers >= 0.2.0.0 && < 0.5, - mtl >= 2.0.1 && < 2.3, + base >= 4.4 && < 5, + transformers >= 0.4 && < 0.5, + mtl >= 2.2 && < 2.3, free >= 4.2 && < 5, bifunctors >= 4.1 && < 5, - profunctors >= 3.1.1 && < 4.1, async >= 2.0.1 && < 2.1, process >= 1.2.0 && < 1.3, - pipes >= 4.0 && < 4.2, - pipes-bytestring >= 2.0.0 && < 2.1, - pipes-text >= 0.0.0.9 && < 0.0.2, + pipes >= 4.1.2 && < 4.2, + pipes-bytestring >= 2.1.0 && < 2.2, + pipes-text >= 0.0.0.10 && < 0.0.2, text >= 0.11.2 && < 1.2, pipes-concurrency >= 2.0.2 && < 3, pipes-safe >= 2.2.0 && < 3, - exceptions >= 0.6.0 && < 0.7 + pipes-parse >=3.0.1 && <3.1, + exceptions >= 0.6.0 && < 0.7, + void >= 0.6 && < 0.7, + containers >= 0.4, + semigroups >= 0.15 && < 0.16 - default-language: Haskell2010 +Test-suite test + default-language: + Haskell2010 + type: + exitcode-stdio-1.0 + hs-source-dirs: + tests + main-is: + test.hs + build-depends: + base >= 4.4 && < 5 + , transformers >= 0.4 && < 0.5 + , mtl >= 2.2 && < 2.3 + , free >= 4.2 && < 5 + , bifunctors >= 4.1 && < 5 + , async >= 2.0.1 && < 2.1 + , process >= 1.2.0 && < 1.3 + , pipes >= 4.1.2 && < 4.2 + , pipes-bytestring >= 2.1.0 && < 2.2 + , pipes-text >= 0.0.0.10 && < 0.0.2 + , text >= 0.11.2 && < 1.2 + , pipes-concurrency >= 2.0.2 && < 3 + , pipes-safe >= 2.2.0 && < 3 + , pipes-parse >=3.0.1 && <3.1 + , exceptions >= 0.6.0 && < 0.7 + , void >= 0.6 && < 0.7 + , semigroups >= 0.15 && < 0.16 + , containers >= 0.4 + , process-streaming + , tasty >= 0.9 + , tasty-hunit >= 0.9 + , attoparsec >= 0.11 + , pipes-attoparsec >= 0.5 + , pipes-group >= 1.0.1 + , bytestring >= 0.10 + , lens >= 4 + , directory >= 1.2 Source-repository head type: git
src/System/Process/Lens.hs view
@@ -19,8 +19,13 @@ , _delegate_ctlc , handles , nohandles - , handlesioe + , handleso + , handlese , handlesoe + , handlesi + , handlesio + , handlesie + , handlesioe ) where import Data.Maybe @@ -111,7 +116,7 @@ {-| A 'Lens' for the return value of 'createProcess' that focuses on the handles. - > handlesioe :: Lens' (Maybe Handle, Maybe Handle, Maybe Handle,ProcessHandle) (Maybe Handle, Maybe Handle, Maybe Handle) + > handles :: Lens' (Maybe Handle, Maybe Handle, Maybe Handle,ProcessHandle) (Maybe Handle, Maybe Handle, Maybe Handle) -} handles :: forall m. Functor m => ((Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) -> m (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) handles f quad = setHandles quad <$> f (getHandles quad) @@ -120,6 +125,11 @@ getHandles (c1'',c2'',c3'',c4'') = (c1'',c2'',c3'') +{-| + A 'Prism' that matches when none of the standard streams have been piped. + + > nohandles :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) () + -} nohandles :: forall m. Applicative m => (() -> m ()) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) nohandles f quad = case impure quad of Left l -> pure l @@ -129,11 +139,44 @@ impure x = Left x justify () = (Nothing, Nothing, Nothing) + {-| - A 'Prism' that removes the 'Maybe's from @stdin@, @stdout@ and @stderr@ or fails to match if any of them is 'Nothing'. + A 'Prism' that matches when only @stdin@ has been piped. - > handlesioe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) -> (Handle, Handle, Handle) + > handlesi :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle) -} +handlesi :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) +handlesi f quad = case impure quad of + Left l -> pure l + Right r -> fmap justify (f r) + where + impure (Just h1, Nothing, Nothing) = Right h1 + impure x = Left x + justify h1 = (Just h1, Nothing, Nothing) + +handlesio :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) +handlesio f quad = case impure quad of + Left l -> pure l + Right r -> fmap justify (f r) + where + impure (Just h1, Just h2, Nothing) = Right (h1,h2) + impure x = Left x + justify (h1,h2) = (Just h1, Just h2, Nothing) + +handlesie :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) +handlesie f quad = case impure quad of + Left l -> pure l + Right r -> fmap justify (f r) + where + impure (Just h1, Nothing, Just h2) = Right (h1,h2) + impure x = Left x + justify (h1,h2) = (Just h1, Nothing, Just h2) + +{-| + A 'Prism' that matches when all three @stdin@, @stdout@ and @stderr@ have been piped. + + > handlesioe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle, Handle) + -} handlesioe :: forall m. Applicative m => ((Handle, Handle, Handle) -> m (Handle, Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) handlesioe f quad = case impure quad of Left l -> pure l @@ -144,9 +187,9 @@ justify (h1, h2, h3) = (Just h1, Just h2, Just h3) {-| - A 'Prism' that removes the 'Maybe's from @stdout@ and @stderr@ or fails to match if any of them is 'Nothing'. + A 'Prism' that matches when only @stdout@ and @stderr@ have been piped. - > handlesoe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) -> (Handle, Handle) + > handlesoe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle) -} handlesoe :: forall m. Applicative m => ((Handle, Handle) -> m (Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) handlesoe f quad = case impure quad of @@ -157,3 +200,30 @@ impure x = Left x justify (h2, h3) = (Nothing, Just h2, Just h3) +{-| + A 'Prism' that matches when only @stdout@ has been piped. + + > handleso :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle) + -} +handleso :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) +handleso f quad = case impure quad of + Left l -> pure l + Right r -> fmap justify (f r) + where + impure (Nothing, Just h2, Nothing) = Right h2 + impure x = Left x + justify h2 = (Nothing, Just h2, Nothing) + +{-| + A 'Prism' that matches when only @stderr@ has been piped. + + > handlese :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle) + -} +handlese :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) +handlese f quad = case impure quad of + Left l -> pure l + Right r -> fmap justify (f r) + where + impure (Nothing, Nothing, Just h2) = Right h2 + impure x = Left x + justify h2 = (Nothing, Nothing, Just h2)
src/System/Process/Streaming.hs view
@@ -3,67 +3,71 @@ -- This module contains helper functions and types built on top of -- "System.Process" and "Pipes". -- --- They provide concurrent, buffered (to avoid deadlocks) streaming access to --- the inputs and outputs of system processes. +-- They provide concurrent, streaming access to the inputs and outputs of +-- system processes. -- --- There's also an emphasis in having error conditions explicit in the types, --- instead of throwing exceptions. +-- Error conditions that are not directly related to IO are made explicit +-- in the types. -- -- Regular 'Consumer's, 'Parser's from @pipes-parse@ and folds from --- "Pipes.Prelude" (also folds from @pipes-bytestring@ and @pipes-text@) can be --- used to consume the output streams of the external processes. +-- "Pipes.Prelude" (also folds from @pipes-bytestring@ and @pipes-text@) +-- can be used to consume the output streams of the external processes. -- ----------------------------------------------------------------------------- {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE ViewPatterns #-} module System.Process.Streaming ( -- * Execution execute - , exitCode - , safeExecute - , simpleSafeExecute - -- * Piping standard streams + , executeFallibly + -- * Piping Policies , PipingPolicy , nopiping + , pipeo + , pipee , pipeoe + , pipeoec + , pipei + , pipeio + , pipeie , pipeioe - - -- * Separated stdout/stderr - , separated + , pipeioec - -- * Stdout/stderr combined as text - , combined + -- * Pumping bytes into stdin + , Pump (..) + , fromProducer + , fromSafeProducer + , fromFallibleProducer + -- * Siphoning bytes out of stdout/stderr + , Siphon + , siphon + , siphon' + , fromFold + , fromFold' + , fromFold'_ + , fromConsumer + , fromSafeConsumer + , fromFallibleConsumer + , fromParser + , unwanted + , DecodingFunction + , encoded + -- * Line handling , LinePolicy , linePolicy - - -- * Decoding and leftovers - , encoding - , LeftoverPolicy(..) - , ignoreLeftovers - , failOnLeftovers - - -- * Construction of feeding/consuming functions - , useConsumer - , useProducer - , surely - , safely - , fallibly - , monoidally - , nop - - -- * Concurrency helpers - , Conceit (..) - , conceit - , mapConceit - - , Siphon (..) - , forkSiphon - , SiphonL (..) - , SiphonR (..) - + -- * Pipelines + , executePipeline + , executePipelineFallibly + , CreatePipeline (..) + , simplePipeline + , Stage (..) + , SubsequentStage (..) -- * Re-exports -- $reexports , module System.Process @@ -71,17 +75,22 @@ import Data.Maybe import Data.Bifunctor -import Data.Profunctor import Data.Functor.Identity import Data.Either import Data.Monoid +import Data.Foldable import Data.Traversable import Data.Typeable +import Data.Tree import Data.Text +import Data.Text.Encoding +import Data.Void +import Data.List.NonEmpty +import qualified Data.List.NonEmpty as N import Control.Applicative import Control.Monad import Control.Monad.Trans.Free -import Control.Monad.Error +import Control.Monad.Except import Control.Monad.State import Control.Monad.Writer.Strict import qualified Control.Monad.Catch as C @@ -93,6 +102,7 @@ import qualified Pipes.Prelude as P import Pipes.Lift import Pipes.ByteString +import Pipes.Parse import qualified Pipes.Text as T import Pipes.Concurrent import Pipes.Safe (SafeT, runSafeT) @@ -102,6 +112,9 @@ import System.Process.Lens import System.Exit +execute :: PipingPolicy Void a -> CreateProcess -> IO (ExitCode,a) +execute pp cprocess = either absurd id <$> executeFallibly pp cprocess + {-| Executes an external process. The standard streams are piped and consumed in a way defined by the 'PipingPolicy' argument. @@ -115,44 +128,44 @@ If an error @e@ or an exception happens, the external process is terminated. -} -execute :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a)) -execute (PipingPolicy tr somePrism action) procSpec = mask $ \restore -> do - (min,mout,merr,phandle) <- createProcess (tr procSpec) +executeFallibly :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a)) +executeFallibly pp record = case pp of + PPNone a -> executeInternal record nohandles $ + \() -> (return . Right $ a,return ()) + PPOutput action -> executeInternal (record{std_out = CreatePipe}) handleso $ + \h->(action (fromHandle h),hClose h) + PPError action -> executeInternal (record{std_err = CreatePipe}) handlese $ + \h->(action (fromHandle h),hClose h) + PPOutputError action -> executeInternal (record{std_out = CreatePipe, std_err = CreatePipe}) handlesoe $ + \(hout,herr)->(action (fromHandle hout,fromHandle herr),hClose hout `finally` hClose herr) + PPInput action -> executeInternal (record{std_in = CreatePipe}) handlesi $ + \h -> (action (toHandle h, hClose h), return ()) + PPInputOutput action -> executeInternal (record{std_in = CreatePipe,std_out = CreatePipe}) handlesio $ + \(hin,hout) -> (action (toHandle hin,hClose hin,fromHandle hout), hClose hout) + PPInputError action -> executeInternal (record{std_in = CreatePipe,std_err = CreatePipe}) handlesie $ + \(hin,herr) -> (action (toHandle hin,hClose hin,fromHandle herr), hClose herr) + PPInputOutputError action -> executeInternal (record{std_in = CreatePipe, std_out = CreatePipe, std_err = CreatePipe}) handlesioe $ + \(hin,hout,herr) -> (action (toHandle hin,hClose hin,fromHandle hout,fromHandle herr), hClose hout `finally` hClose herr) + +executeInternal :: CreateProcess -> (forall m. Applicative m => (t -> m t) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (t ->(IO (Either e a),IO ())) -> IO (Either e (ExitCode,a)) +executeInternal record somePrism allocator = mask $ \restore -> do + (min,mout,merr,phandle) <- createProcess record case getFirst . getConst . somePrism (Const . First . Just) $ (min,mout,merr) of Nothing -> - throwIO (userError "stdin/stdout/stderr handle unexpectedly null") + throwIO (userError "stdin/stdout/stderr handle unwantedly null") `finally` terminateCarefully phandle Just t -> - let (a, cleanup) = action t in + let (action,cleanup) = allocator t in -- Handles must be closed *after* terminating the process, because a close -- operation may block if the external process has unflushed bytes in the stream. - (restore (terminateOnError phandle a) `onException` terminateCarefully phandle) - `finally` - cleanup + (restore (terminateOnError phandle action) `onException` terminateCarefully phandle) `finally` cleanup exitCode :: (ExitCode,a) -> Either Int a exitCode (ec,a) = case ec of ExitSuccess -> Right a ExitFailure i -> Left i -{-| - Like 'execute', but 'IOException's are caught and converted to the error type @e@. - - Exit codes denoting errors are also converted to @e@ values. - -} -safeExecute :: (IOError -> e) -> (Int -> e) -> PipingPolicy e a -> CreateProcess -> IO (Either e a) -safeExecute exh ech pp cp = collapseEithers <$> (tryIOError $ execute pp cp) - where - collapseEithers = join . join . bimap exh (fmap (bimap ech id . exitCode)) - -{-| - A simpler version of 'safeExecute' that assumes the error type @e@ is 'String'. - -} -simpleSafeExecute :: PipingPolicy String a -> CreateProcess -> IO (Either String a) -simpleSafeExecute = safeExecute show (mappend "Exit code: " . show) - - terminateCarefully :: ProcessHandle -> IO () terminateCarefully pHandle = do mExitCode <- getProcessExitCode pHandle @@ -174,81 +187,128 @@ return $ Right (exitCode,r) {-| - A 'PipingPolicy' specifies what standard streams of the external process -should be piped, and how to consume them. + A 'PipingPolicy' determines what standard streams will be piped and what to +do with them. - Values of type @a@ denote successful consumption of the streams, values of -type @e@ denote errors. --} -data PipingPolicy e a = forall t. PipingPolicy (CreateProcess -> CreateProcess) (forall m. Applicative m => (t -> m t) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) (t -> (IO (Either e a), IO ())) + The user doesn't need to manually set the 'std_in', 'std_out' and 'std_err' +fields of the 'CreateProcess' record to 'CreatePipe', this is done +automatically. -instance Functor (PipingPolicy e) where - fmap f (PipingPolicy cpf prsm func) = PipingPolicy cpf prsm $ - (fmap (bimap (fmap (fmap f)) id) func) + A 'PipingPolicy' is parametrized by the type @e@ of errors that can abort +the processing of the streams. + -} +-- Knows that there is a stdin, stdout and a stderr, +-- but doesn't know anything about file handlers or CreateProcess. +data PipingPolicy e a = + PPNone a + | PPOutput (Producer ByteString IO () -> IO (Either e a)) + | PPError (Producer ByteString IO () -> IO (Either e a)) + | PPOutputError ((Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a)) + | PPInput ((Consumer ByteString IO (), IO ()) -> IO (Either e a)) + | PPInputOutput ((Consumer ByteString IO (), IO (),Producer ByteString IO ()) -> IO (Either e a)) + | PPInputError ((Consumer ByteString IO (), IO (), Producer ByteString IO ()) -> IO (Either e a)) + | PPInputOutputError ((Consumer ByteString IO (),IO (),Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a)) + deriving (Functor) instance Bifunctor PipingPolicy where - bimap f g (PipingPolicy cpf prsm func) = PipingPolicy cpf prsm $ - (fmap (bimap (fmap (bimap f g)) id) func) + bimap f g pp = case pp of + PPNone a -> PPNone $ g a + PPOutput action -> PPOutput $ fmap (fmap (bimap f g)) action + PPError action -> PPError $ fmap (fmap (bimap f g)) action + PPOutputError action -> PPOutputError $ fmap (fmap (bimap f g)) action + PPInput action -> PPInput $ fmap (fmap (bimap f g)) action + PPInputOutput action -> PPInputOutput $ fmap (fmap (bimap f g)) action + PPInputError action -> PPInputError $ fmap (fmap (bimap f g)) action + PPInputOutputError action -> PPInputOutputError $ fmap (fmap (bimap f g)) action {-| Do not pipe any standard stream. -} nopiping :: PipingPolicy e () -nopiping = PipingPolicy id nohandles (\() -> (return $ return (), return ())) +nopiping = PPNone () {-| - Pipe stderr and stdout. + Pipe @stdout@. +-} +pipeo :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a +pipeo (runSiphon -> siphonout) = PPOutput $ siphonout - See also the 'separated' and 'combined' functions. +{-| + Pipe @stderr@. -} -pipeoe :: (Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)) - -> PipingPolicy e a -pipeoe consumefunc = PipingPolicy changecp handlesoe handler - where handler (hout,herr) = - (,) (consumefunc (fromHandle hout) (fromHandle herr)) - (hClose hout `finally` hClose herr) - changecp cp = cp { std_out = CreatePipe - , std_err = CreatePipe - } +pipee :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a +pipee (runSiphon -> siphonout) = PPError $ siphonout {-| - Pipe stdin, stderr and stdout. + Pipe @stdout@ and @stderr@. +-} +pipeoe :: (Show e,Typeable e) => Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (a,b) +pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = + PPOutputError $ uncurry $ separated siphonout siphonerr - See also the 'separated' and 'combined' functions. +{-| + Pipe @stdout@ and @stderr@ and consume them combined as 'Text'. -} -pipeioe :: (Show e, Typeable e) - => (Consumer ByteString IO () -> IO (Either e a)) - -> (Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e b)) - -> PipingPolicy e (a,b) -pipeioe feeder consumefunc = PipingPolicy changecp handlesioe handler - where handler (hin,hout,herr) = - (,) (conceit (feeder (toHandle hin) `finally` hClose hin) - (consumefunc (fromHandle hout) (fromHandle herr))) - (hClose hin `finally` hClose hout `finally` hClose herr) - changecp cp = cp { std_in = CreatePipe - , std_out = CreatePipe - , std_err = CreatePipe - } +pipeoec :: (Show e,Typeable e) => LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e a +pipeoec policy1 policy2 (runSiphon -> siphon) = + PPOutputError $ uncurry $ combined policy1 policy2 siphon {-| - 'separate' should be used when we want to consume @stdout@ and @stderr@ -concurrently and independently. It constructs a function that can be plugged -into functions like 'pipeoe'. + Pipe @stdin@. +-} +pipei :: (Show e, Typeable e) => Pump ByteString e i -> PipingPolicy e i +pipei (Pump feeder) = PPInput $ \(consumer,cleanup) -> feeder consumer `finally` cleanup - If the consuming functions return with @a@ and @b@, the corresponding -streams keep being drained until the end. The combined value is not returned -until both @stdout@ and @stderr@ are closed by the external process. +{-| + Pipe @stdin@ and @stdout@. +-} +pipeio :: (Show e, Typeable e) + => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a) +pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) -> + (conceit (feeder consumer `finally` cleanup) (siphonout producer)) - However, if any of the consuming functions fails with @e@, the whole -computation fails immediately with @e@. - -} +{-| + Pipe @stdin@ and @stderr@. +-} +pipeie :: (Show e, Typeable e) + => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a) +pipeie (Pump feeder) (runSiphon -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) -> + (conceit (feeder consumer `finally` cleanup) (siphonerr producer)) + +{-| + Pipe @stdin@, @stdout@ and @stderr@. +-} +pipeioe :: (Show e, Typeable e) + => Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (i,a,b) +pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $ + \(consumer,cleanup,outprod,errprod) -> + (conceit (feeder consumer `finally` cleanup) + (separated siphonout siphonerr outprod errprod)) + where + flattenTuple (i, (a, b)) = (i,a,b) + +{-| + Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'. +-} +pipeioec :: (Show e, Typeable e) + => Pump ByteString e i -> LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e (i,a) +pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $ + \(consumer,cleanup,outprod,errprod) -> + (conceit (feeder consumer `finally` cleanup) + (combined policy1 policy2 siphon outprod errprod)) + separated :: (Show e, Typeable e) => (Producer ByteString IO () -> IO (Either e a)) -> (Producer ByteString IO () -> IO (Either e b)) -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e (a,b)) separated outfunc errfunc outprod errprod = - conceit (buffer_ outfunc outprod) (buffer_ errfunc errprod) + conceit (outfunc outprod) (errfunc errprod) +{-| + Defines how to decode a stream of bytes into text, including what to do + in presence of leftovers. Also defines how to manipulate each individual + line of text. + -} data LinePolicy e = LinePolicy ((FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> Producer ByteString IO () -> IO (Either e ())) instance Functor LinePolicy where @@ -257,144 +317,66 @@ {-| Constructs a 'LinePolicy'. - The first argument is a function function that decodes 'ByteString' into -'T.Text'. See the section /Decoding Functions/ in the documentation for the -"Pipes.Text" module. + The second argument is a 'Siphon' value that specifies how to handle +decoding failures. Passing @pure ()@ will ignore any leftovers. Passing +@unwanted ()@ will abort the computation if leftovers remain. - The second argument is a function that modifies each individual line. The -line is represented as a 'Producer' to avoid having to keep it wholly in -memory. If you want the lines unmodified, just pass @id@. Line prefixes are -easy to add using applicative notation: + The third argument is a function that modifies each individual line. + The line is represented as a 'Producer' to avoid having to keep it + wholly in memory. If you want the lines unmodified, just pass @id@. + Line prefixes are easy to add using applicative notation: > (\x -> yield "prefix: " *> x) - The third argument is a 'LeftoverPolicy' value that specifies how to handle -decoding failures. -} -linePolicy :: (forall r. Producer ByteString IO r -> Producer T.Text IO (Producer ByteString IO r)) +linePolicy :: (Show e, Typeable e) + => DecodingFunction ByteString Text + -> Siphon ByteString e () -> (forall r. Producer T.Text IO r -> Producer T.Text IO r) - -> (LeftoverPolicy () ByteString e) - -> LinePolicy e -linePolicy decoder transform lopo = LinePolicy $ \teardown producer -> do + -> LinePolicy e +linePolicy decoder lopo transform = LinePolicy $ \teardown producer -> do let freeLines = transFreeT transform . viewLines . decoder $ producer viewLines = getConst . T.lines Const - teardown freeLines >>= runLeftoverPolicy lopo () - -{-| - In the Pipes ecosystem, leftovers from decoding operations are often stored -in the result value of 'Producer's (as 'Producer's themselves). - - A 'LeftoverPolicy' receives a value @a@ and a producer of lefovers of type -@l@. Analyzing the producer, it may modify the value @a@ or fail outright, -depending of what the leftovers are. - -} -data LeftoverPolicy a l e = LeftoverPolicy { runLeftoverPolicy :: a -> Producer l IO () -> IO (Either e a) } - -instance Functor (LeftoverPolicy a l) where - fmap f (LeftoverPolicy x) = LeftoverPolicy $ fmap (fmap (fmap (bimap f id))) x - -instance Profunctor (LeftoverPolicy a) where - dimap ab cd (LeftoverPolicy pf) = LeftoverPolicy $ \a p -> liftM (bimap cd id) $ pf a $ p >-> P.map ab - -{-| - Never fails for any leftover. - -} -ignoreLeftovers :: LeftoverPolicy a l e -ignoreLeftovers = LeftoverPolicy $ pure . pure . pure - -{-| - Fails if it encounters any leftover, and constructs the error out of the -first undedcoded data. - - For simple error handling, just ignore the @a@ and the undecoded data: - - > (failOnLeftvoers (\_ _->"badbytes")) :: LeftoverPolicy (Producer b IO ()) String a - - For more detailed error handling, you may want to include the result until -the error @a@ and/or the first undecoded values @b@ in your custom error -datatype. - -} -failOnLeftovers :: (a -> b -> e) -> LeftoverPolicy a b e -failOnLeftovers errh = LeftoverPolicy $ \a remainingBytes -> do - r <- next remainingBytes - return $ case r of - Left () -> Right a - Right (somebytes,_) -> Left $ errh a somebytes - -{-| - The bytes from @stdout@ and @stderr@ are decoded into 'Text', splitted into -lines (maybe applying some transformation to each line) and then combined and -consumed by the function passed as argument. - - For both @stdout@ and @stderr@, a 'LinePolicy' must be supplied. - - Like with 'separated', the streams are drained to completion if no errors -happen, but the computation is aborted immediately if any error @e@ is -returned. - - 'combined' returns a function that can be plugged into funtions like 'pipeioe'. + teardown freeLines >>= runSiphon lopo - /Beware!/ 'combined' avoids situations in which a line emitted -in @stderr@ cuts a long line emitted in @stdout@, see -<http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here> for a description of the problem. To avoid this, the combined text -stream is locked while writing each individual line. But this means that if the -external program stops writing to a handle /while in the middle of a line/, -lines coming from the other handles won't get printed, either! - -} +-- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here combined :: (Show e, Typeable e) => LinePolicy e -> LinePolicy e -> (Producer T.Text IO () -> IO (Either e a)) -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a) combined (LinePolicy fun1) (LinePolicy fun2) combinedConsumer prod1 prod2 = - manyCombined [fmap (($prod1).buffer_) fun1, fmap (($prod2).buffer_) fun2] combinedConsumer - -manyCombined :: (Show e, Typeable e) - => [((FreeT (Producer T.Text IO) IO (Producer ByteString IO ())) -> IO (Producer ByteString IO ())) -> IO (Either e ())] - -> (Producer T.Text IO () -> IO (Either e a)) - -> IO (Either e a) -manyCombined actions consumer = do - (outbox, inbox, seal) <- spawn' Unbounded - mVar <- newMVar outbox - r <- conceit (mapConceit ($ iterTLines mVar) actions `finally` atomically seal) - (consumer (fromInput inbox) `finally` atomically seal) - return $ snd <$> r - where - iterTLines mvar = iterT $ \textProducer -> do - -- the P.drain bit was difficult to figure out!!! - join $ withMVar mvar $ \output -> do - runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain) - -{-| - Useful for constructing @stdout@ or @stderr@ consuming functions from a -'Consumer', to be plugged into 'separated' or 'combined'. - - You may need to use 'surely' for the types to fit. - -} -useConsumer :: Monad m => Consumer b m () -> Producer b m () -> m () -useConsumer consumer producer = runEffect $ producer >-> consumer - -{-| - Useful for constructing @stdin@ feeding functions from a 'Producer'. + manyCombined [fmap ($prod1) fun1, fmap ($prod2) fun2] combinedConsumer + where + manyCombined :: (Show e, Typeable e) + => [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())] + -> (Producer T.Text IO () -> IO (Either e a)) + -> IO (Either e a) + manyCombined actions consumer = do + (outbox, inbox, seal) <- spawn' Single + mVar <- newMVar outbox + runConceit $ + Conceit (mapConceit ($ iterTLines mVar) actions `finally` atomically seal) + *> + Conceit (consumer (fromInput inbox) `finally` atomically seal) + where + iterTLines mvar = iterT $ \textProducer -> do + -- the P.drain bit was difficult to figure out!!! + join $ withMVar mvar $ \output -> do + runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain) - You may need to use 'surely' for the types to fit. - -} -useProducer :: Monad m => Producer b m () -> Consumer b m () -> m () -useProducer producer consumer = runEffect (producer >-> consumer) +fromProducer :: Producer b IO () -> Pump b e () +fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (producer >-> consumer) -{-| - Useful when we want to plug in a handler that doesn't return an 'Either'. For -example folds from "Pipes.Prelude", or functions created from simple -'Consumer's with 'useConsumer'. +fromSafeProducer :: Producer b (SafeT IO) () -> Pump b e () +fromSafeProducer producer = Pump $ safely $ \consumer -> fmap pure $ runEffect (producer >-> consumer) - > surely = fmap (fmap Right) - -} -surely :: (Functor f0, Functor f1) => f0 (f1 a) -> f0 (f1 (Either e a)) -surely = fmap (fmap Right) +fromFallibleProducer :: Producer b (ExceptT e IO) () -> Pump b e () +fromFallibleProducer producer = Pump $ \consumer -> runExceptT $ runEffect (producer >-> hoist lift consumer) {-| Useful when we want to plug in a handler that does its work in the 'SafeT' @@ -405,81 +387,42 @@ -> t m l -> m x safely activity = runSafeT . activity . hoist lift -fallibly :: (MFunctor t, Monad m, Error e) - => (t (ErrorT e m) l -> (ErrorT e m) x) - -> t m l -> m (Either e x) -fallibly activity = runErrorT . activity . hoist lift - {-| - Usually, it is better to use a fold form "Pipes.Prelude" instead of this -function. But this function has the ability to return the monoidal result -accumulated up until the error happened. - - The first argument is a function that combines the initial error with the -monoidal result to build the definitive error value. If you want to discard the -results, use 'const' as the first argument. - -} -monoidally :: (MFunctor t,Monad m,Monoid w, Error e') - => (e' -> w -> e) - -> (t (ErrorT e' (WriterT w m)) l -> ErrorT e' (WriterT w m) ()) - -> t m l -> m (Either e w) -monoidally errh activity proxy = do - (r,w) <- runWriterT . runErrorT . activity . hoist (lift.lift) $ proxy - return $ case r of - Left e' -> Left $ errh e' w - Right () -> Right $ w - -{-| - Value to plug into 'separated' or 'combined' when we are not interested in -doing anything with the stream. It returns immediately with @()@. - - Notice that even if 'nop' returns immediately, 'separate' and -'combined' drain the streams to completion before returning. - -} -nop :: Applicative m => i -> m (Either e ()) -nop = pure . pure . pure $ () - -buffer :: (Show e, Typeable e) - => LeftoverPolicy a l e - -> (Producer b IO () -> IO (Either e a)) - -> Producer b IO (Producer l IO ()) -> IO (Either e a) -buffer policy activity producer = do - (outbox,inbox,seal) <- spawn' Unbounded - r <- conceit - (do feeding <- async $ runEffect $ - producer >-> (toOutput outbox >> P.drain) - Right <$> wait feeding `finally` atomically seal - ) - (activity (fromInput inbox) `finally` atomically seal) - case r of - Left e -> return $ Left e - Right (lp,r') -> runLeftoverPolicy policy r' lp - -buffer_ :: (Show e, Typeable e) - => (Producer ByteString IO () -> IO (Either e a)) - -> Producer ByteString IO () -> IO (Either e a) -buffer_ activity producer = do - (outbox,inbox,seal) <- spawn' Unbounded - r <- conceit - (do feeding <- async $ runEffect $ - producer >-> (toOutput outbox >> P.drain) - Right <$> wait feeding `finally` atomically seal - ) - (activity (fromInput inbox) `finally` atomically seal) - return $ fmap snd r + See the section /Non-lens decoding functions/ in the documentation for the +@pipes-text@ package. +-} +type DecodingFunction bytes text = forall r. Producer bytes IO r -> Producer text IO (Producer bytes IO r) {-| - Adapts a function that works with 'Producer's of decoded values so that it -works with 'Producer's of still undecoded values, by supplying a decoding -function and a 'LeftoverPolicy'. + Constructs a 'Siphon' that works on undecoded values out of a 'Siphon' that +works on decoded values. + + The two first arguments are a decoding function and a 'Siphon' that +determines how to handle leftovers. Pass @pure id@ to ignore leftovers. Pass +@unwanted id@ to abort the computation if leftovers remain. -} -encoding :: (Show e, Typeable e) - => (Producer b IO () -> Producer t IO (Producer b IO ())) - -> LeftoverPolicy a b e - -> (Producer t IO () -> IO (Either e a)) - -> Producer b IO () -> IO (Either e a) -encoding decoder policy activity producer = buffer policy activity $ decoder producer - +encoded :: (Show e, Typeable e) + => DecodingFunction bytes text + -> Siphon bytes e (a -> b) + -> Siphon text e a + -> Siphon bytes e b +encoded decoder policy activity = Halting $ \producer -> buffer policy activity $ decoder producer + where + buffer :: (Show e, Typeable e) + => Siphon bytes e (a -> b) + -> Siphon text e a + -> Producer text IO (Producer bytes IO ()) -> IO (Either e b) + buffer policy activity producer = do + (outbox,inbox,seal) <- spawn' Single + r <- conceit + (do feeding <- async $ runEffect $ + producer >-> (toOutput outbox >> P.drain) + Right <$> wait feeding `finally` atomically seal + ) + (runSiphon activity (fromInput inbox) `finally` atomically seal) + case r of + Left e -> return $ Left e + Right (leftovers,a) -> runSiphon (fmap ($a) policy) leftovers data WrappedError e = WrappedError e deriving (Show, Typeable) @@ -493,19 +436,6 @@ revealError action = catch (action >>= return . Right) (\(WrappedError e) -> return . Left $ e) -{-| - 'Conceit' is very similar to 'Control.Concurrent.Async.Concurrently' from the -@async@ package, but it has an explicit error type @e@. - - The 'Applicative' instance is used to run actions concurrently, wait until -they finish, and combine their results. - - However, if any of the actions fails with @e@ the other actions are -immediately cancelled and the whole computation fails with @e@. - - To put it another way: 'Conceit' behaves like 'Concurrently' for successes and -like 'race' for errors. --} newtype Conceit e a = Conceit { runConceit :: IO (Either e a) } instance Functor (Conceit e) where @@ -543,64 +473,396 @@ mapConceit :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b)) mapConceit f = revealError . mapConcurrently (elideError . f) +newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } + +instance Functor (Pump b e) where + fmap f (Pump x) = Pump $ fmap (fmap (fmap f)) x + +instance Bifunctor (Pump b) where + bimap f g (Pump x) = Pump $ fmap (liftM (bimap f g)) x + +instance (Show e, Typeable e) => Applicative (Pump b e) where + pure = Pump . pure . pure . pure + Pump fs <*> Pump as = + Pump $ \consumer -> do + (outbox1,inbox1,seal1) <- spawn' Single + (outbox2,inbox2,seal2) <- spawn' Single + runConceit $ + Conceit (runExceptT $ do + r1 <- ExceptT $ (fs $ toOutput outbox1) + `finally` atomically seal1 + r2 <- ExceptT $ (as $ toOutput outbox2) + `finally` atomically seal2 + return $ r1 r2 + ) + <* + Conceit (do + (runEffect $ + (fromInput inbox1 >> fromInput inbox2) >-> consumer) + `finally` atomically seal1 + `finally` atomically seal2 + return $ pure () + ) + +instance (Show e, Typeable e, Monoid a) => Monoid (Pump b e a) where + mempty = Pump . pure . pure . pure $ mempty + mappend s1 s2 = (<>) <$> s1 <*> s2 + {-| - 'Siphon' is a newtype around a function that does something with a -'Producer'. The applicative instance fuses the functions, so that each one -receives its own copy of the 'Producer' and runs concurrently with the others. -Like with 'Conceit', if any of the functions fails with @e@ the others are -immediately cancelled and the whole computation fails with @e@. + A 'Siphon' represents a computation that completely drains a producer, but +may fail early with an error of type @e@. - 'Siphon' and its accompanying functions are useful to run multiple -parsers from "Pipes.Parse" in parallel over the same 'Producer'. - -} -newtype Siphon b e a = Siphon { runSiphon :: Producer b IO () -> IO (Either e a) } + 'pure' creates a 'Siphon' that does nothing besides draining the +'Producer'. -instance Functor (Siphon b e) where - fmap f (Siphon x) = Siphon $ fmap (fmap (fmap f)) x + '<*>' executes its arguments concurrently. The 'Producer' is forked so + that each argument receives its own copy of the data. + -} +data Siphon b e a = + Trivial a + | Unhalting (forall r. Producer b IO r -> IO (Either e (a,r))) + | Halting (Producer b IO () -> IO (Either e a)) + deriving (Functor) instance Bifunctor (Siphon b) where - bimap f g (Siphon x) = Siphon $ fmap (liftM (bimap f g)) x + bimap f g s = case s of + Trivial a -> Trivial $ g a + Unhalting u -> Unhalting $ fmap (liftM (bimap f (bimap g id))) u + Halting h -> Halting $ fmap (liftM (bimap f g)) h instance (Show e, Typeable e) => Applicative (Siphon b e) where - pure = Siphon . pure . pure . pure - Siphon fs <*> Siphon as = - Siphon $ \producer -> do - (outbox1,inbox1,seal1) <- spawn' Unbounded - (outbox2,inbox2,seal2) <- spawn' Unbounded - r <- conceit (do - feeding <- async $ runEffect $ - producer >-> P.tee (toOutput outbox1 >> P.drain) - >-> (toOutput outbox2 >> P.drain) - sealing <- async $ wait feeding `finally` atomically seal1 - `finally` atomically seal2 - return $ Right () - ) - (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) - `finally` atomically seal1) - ((as $ fromInput inbox2) - `finally` atomically seal2) - ) - return $ fmap snd r + pure = Trivial + + s1 <*> s2 = case (s1,s2) of + (Trivial f, s2') -> fmap f s2' + (s1', Trivial a) -> fmap ($ a) s1' + (Halting fs, Halting as) -> fork fs as + (Halting fs, Unhalting as) -> fork fs (halting as) + (Unhalting fs, Halting as) -> fork (halting fs) as + (Unhalting fs, Unhalting as) -> fork (halting fs) (halting as) + where + fork fs as = + Halting $ \producer -> do + (outbox1,inbox1,seal1) <- spawn' Single + (outbox2,inbox2,seal2) <- spawn' Single + runConceit $ + Conceit (do + -- mmm who cancels these asyncs ?? + feeding <- async $ runEffect $ + producer >-> P.tee (toOutput outbox1 >> P.drain) + >-> (toOutput outbox2 >> P.drain) + -- is these async neccessary ?? + sealing <- async $ wait feeding `finally` atomically seal1 + `finally` atomically seal2 + return $ pure () + ) + *> + Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) + `finally` atomically seal1) + ((as $ fromInput inbox2) + `finally` atomically seal2) + ) + +halting :: (forall r. Producer b IO r -> IO (Either e (a,r))) + -> (Producer b IO () -> IO (Either e a)) +halting polyfunc = \producer -> + liftM (fmap fst) $ polyfunc producer + instance (Show e, Typeable e, Monoid a) => Monoid (Siphon b e a) where - mempty = Siphon . pure . pure . pure $ mempty + mempty = pure mempty mappend s1 s2 = (<>) <$> s1 <*> s2 -forkSiphon :: (Show e, Typeable e) - => (Producer b IO () -> IO (Either e x)) - -> (Producer b IO () -> IO (Either e y)) - -> Producer b IO () -> IO (Either e (x,y)) -forkSiphon c1 c2 = runSiphon $ (,) <$> Siphon c1 <*> Siphon c2 +fromConsumer :: Consumer b IO () -> Siphon b e () +fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> consumer -newtype SiphonL a b e = SiphonL { runSiphonL :: Producer b IO () -> IO (Either e a) } +fromSafeConsumer :: Consumer b (SafeT IO) () -> Siphon b e () +fromSafeConsumer consumer = siphon $ safely $ \producer -> fmap pure $ runEffect $ producer >-> consumer -instance Profunctor (SiphonL e) where - dimap ab cd (SiphonL pf) = SiphonL $ \p -> liftM (bimap cd id) $ pf $ p >-> P.map ab +fromFallibleConsumer :: Consumer b (ExceptT e IO) () -> Siphon b e () +fromFallibleConsumer consumer = siphon $ \producer -> runExceptT $ runEffect (hoist lift producer >-> consumer) -newtype SiphonR e b a = SiphonR { runSiphonR :: Producer b IO () -> IO (Either e a) } +{-| + Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'. + -} +fromParser :: Parser b IO (Either e a) -> Siphon b e a +fromParser parser = siphon $ Pipes.Parse.evalStateT parser -instance Profunctor (SiphonR e) where - dimap ab cd (SiphonR pf) = SiphonR $ \p -> liftM (fmap cd) $ pf $ p >-> P.map ab +runSiphon :: (Show e, Typeable e) + => Siphon b e a + -> (Producer b IO () -> IO (Either e a)) +runSiphon s = case s of + Trivial a -> \producer -> (runEffect $ producer >-> P.drain) >> (pure . pure $ a) + Unhalting u -> halting u -- no need to re-buffer + Halting h -> buffer_ h + +{-| + Builds a 'Siphon' out of a computation that does something with + a 'Producer', but may fail with an error of type @e@. + + Even if the original computation doesn't completely drain the 'Producer', + the constructed 'Siphon' will. +-} +siphon :: (Producer b IO () -> IO (Either e a)) + -> Siphon b e a +siphon = Halting + + +{-| + Builds a 'Siphon' out of a computation that drains a 'Producer' completely, +but may fail with an error of type @e@. +-} +siphon' :: (forall r. Producer b IO r -> IO (Either e (a,r))) -> Siphon b e a +siphon' = Unhalting + +buffer_ :: (Show e, Typeable e) + => (Producer b IO () -> IO (Either e a)) + -> Producer b IO () -> IO (Either e a) +buffer_ activity producer = do + (outbox,inbox,seal) <- spawn' Single + runConceit $ + Conceit (do feeding <- async $ runEffect $ + producer >-> (toOutput outbox >> P.drain) + Right <$> wait feeding `finally` atomically seal + ) + *> + Conceit (activity (fromInput inbox) `finally` atomically seal) + +fromFold :: (Producer b IO () -> IO a) -> Siphon b e a +fromFold aFold = siphon $ fmap (fmap pure) $ aFold + +{-| + Builds a 'Siphon' out of a computation that folds a 'Producer' and drains it completely. +-} +fromFold' :: (forall r. Producer b IO r -> IO (a,r)) -> Siphon b e a +fromFold' aFold = siphon' $ fmap (fmap pure) aFold + +fromFold'_ :: (forall r. Producer b IO r -> IO r) -> Siphon b e () +fromFold'_ aFold = fromFold' $ fmap (fmap ((,) ())) aFold + +{-| + Constructs a 'Siphon' that aborts the computation if the underlying +'Producer' produces anything. + -} +unwanted :: a -> Siphon b b a +unwanted a = Halting $ \producer -> do + r <- next producer + return $ case r of + Left () -> Right a + Right (b,_) -> Left b + +executePipeline :: PipingPolicy Void a -> CreatePipeline Void -> IO a +executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline + + +{-| + Similar to 'executeFallibly', but instead of a single process it + executes a (possibly branching) pipeline of external processes. + + The 'PipingPolicy' argument views the pipeline as a synthetic process + for which @stdin@ is the @stdin@ of the first stage, @stdout@ is the + @stdout@ of the leftmost terminal stage among those closer to the root, + and @stderr@ is a combination of the @stderr@ streams of all the + stages. + + The combined @stderr@ stream always has UTF-8 encoding. + + This function has a limitation compared to the standard UNIX pipelines. + If a downstream process terminates early without error, the upstream + processes are not notified and keep going. There is no SIGPIPE-like + functionality, in other words. + -} +executePipelineFallibly :: (Show e,Typeable e) => PipingPolicy e a -> CreatePipeline e -> IO (Either e a) +executePipelineFallibly policy pipeline = case policy of + PPNone a -> fmap (fmap (const a)) $ + executePipelineInternal + (\o _ -> mute $ pipeo o) + (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipei i) + (\i _ -> mute $ pipei i) + pipeline + PPOutput action -> do + (outbox, inbox, seal) <- spawn' Single + runConceit $ + (Conceit $ action $ fromInput inbox) + <* + (Conceit $ executePipelineInternal + (\o _ -> pipeo o) + (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ outbox)) + (\i _ -> mute $ pipei i) + pipeline + `finally` atomically seal + ) + PPError action -> do + (eoutbox, einbox, eseal) <- spawn' Single + errf <- errorSiphonUTF8 <$> newMVar eoutbox + runConceit $ + (Conceit $ action $ fromInput einbox) + <* + (Conceit $ executePipelineInternal + (\o l -> mute $ pipeoe o (errf l)) + (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeie i (errf l)) + (\i l -> mute $ pipeie i (errf l)) + pipeline + `finally` atomically eseal) + PPOutputError action -> do + (outbox, inbox, seal) <- spawn' Single + (eoutbox, einbox, eseal) <- spawn' Single + errf <- errorSiphonUTF8 <$> newMVar eoutbox + runConceit $ + (Conceit $ action $ (fromInput inbox,fromInput einbox)) + <* + (Conceit $ executePipelineInternal + (\o l -> mute $ pipeoe o (errf l)) + (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ outbox) (errf l)) + (\i l -> mute $ pipeie i (errf l)) + pipeline + `finally` atomically seal `finally` atomically eseal + ) + PPInput action -> do + (outbox, inbox, seal) <- spawn' Single + runConceit $ + (Conceit $ action (toOutput outbox,atomically seal)) + <* + (Conceit $ executePipelineInternal + (\o _ -> mute $ pipeio (fromProducer . fromInput $ inbox) o) + (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipei i) + (\i _ -> mute $ pipei i) + pipeline + `finally` atomically seal + ) + PPInputOutput action -> do + (ioutbox, iinbox, iseal) <- spawn' Single + (ooutbox, oinbox, oseal) <- spawn' Single + runConceit $ + (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox)) + <* + (Conceit $ executePipelineInternal + (\o _ -> mute $ pipeio (fromProducer . fromInput $ iinbox) o) + (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ ooutbox)) + (\i _ -> mute $ pipei i) + pipeline + `finally` atomically iseal `finally` atomically oseal + ) + PPInputError action -> do + (outbox, inbox, seal) <- spawn' Single + (eoutbox, einbox, eseal) <- spawn' Single + errf <- errorSiphonUTF8 <$> newMVar eoutbox + runConceit $ + (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox)) + <* + (Conceit $ executePipelineInternal + (\o l -> mute $ pipeioe (fromProducer . fromInput $ inbox) o (errf l)) + (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeie i (errf l)) + (\i l -> mute $ pipeie i (errf l)) + pipeline + `finally` atomically seal `finally` atomically eseal + ) + PPInputOutputError action -> do + (ioutbox, iinbox, iseal) <- spawn' Single + (ooutbox, oinbox, oseal) <- spawn' Single + (eoutbox, einbox, eseal) <- spawn' Single + errf <- errorSiphonUTF8 <$> newMVar eoutbox + runConceit $ + (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox)) + <* + (Conceit $ executePipelineInternal + (\o l -> mute $ pipeioe (fromProducer . fromInput $ iinbox) o (errf l)) + (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ ooutbox) (errf l)) + (\i l -> mute $ pipeie i (errf l)) + pipeline + `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal + ) + where + mute = fmap (const ()) + + errorSiphonUTF8 :: MVar (Output ByteString) -> LinePolicy e -> Siphon ByteString e () + errorSiphonUTF8 mvar (LinePolicy fun) = Halting $ fun iterTLines + where + iterTLines = iterT $ \textProducer -> do + -- the P.drain bit was difficult to figure out!!! + join $ withMVar mvar $ \output -> do + runEffect $ (textProducer <* P.yield (singleton '\n')) + >-> P.map Data.Text.Encoding.encodeUtf8 + >-> (toOutput output >> P.drain) + +{-| + An individual stage in a process pipeline. + + The 'LinePolicy' field defines how to handle @stderr@ when @stderr@ is + piped. + + Also required is a function that determines if the returned exit code + represents an error or not. This is necessary because some programs use + non-standard exit codes. + -} +data Stage e = Stage + { + processDefinition :: CreateProcess + , stderrLinePolicy :: LinePolicy e + , exitCodePolicy :: Int -> Maybe e + } deriving (Functor) + +{-| + Any stage beyond the first in a process pipeline. + + Incoming data is passed through the 'Pipe' before being fed to the process. + + Use 'cat' (the identity 'Pipe' from 'Pipes') if no pre-processing is required. + -} +data SubsequentStage e = SubsequentStage (forall a.Pipe ByteString ByteString (ExceptT e IO) a) (Stage e) + +instance Functor (SubsequentStage) where + fmap f (SubsequentStage bs s) = SubsequentStage (hoist (mapExceptT $ liftM (bimap f id)) bs) (fmap f s) + +data CreatePipeline e = CreatePipeline (Stage e) (NonEmpty (Tree (SubsequentStage e))) deriving (Functor) + +{-| + Builds a (possibly branching) pipeline assuming that @stderr@ has the same +encoding in all the stages, that no computation is perfored between the stages, +and that any exit code besides 'ExitSuccess' in a stage actually represents an +error. + -} +simplePipeline :: DecodingFunction ByteString Text -> CreateProcess -> NonEmpty (Tree (CreateProcess)) -> CreatePipeline String +simplePipeline decoder initial forest = CreatePipeline (simpleStage initial) (fmap (fmap simpleSubsequentStage) forest) + where + simpleStage cp = Stage cp simpleLinePolicy simpleErrorPolicy + simpleSubsequentStage = SubsequentStage P.cat . simpleStage + simpleLinePolicy = linePolicy decoder (pure ()) id + simpleErrorPolicy = Just . ("Exit failure: " ++) . show + +executePipelineInternal :: (Show e,Typeable e) + => (Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ()) + -> (Pump ByteString e () -> Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ()) + -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ()) + -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ()) + -> CreatePipeline e + -> IO (Either e ()) +executePipelineInternal ppinitial ppmiddle ppend ppend' (CreatePipeline (Stage cp lpol ecpol) a) = + blende ecpol <$> executeFallibly (ppinitial (runNonEmpty ppend ppend' a) lpol) cp + where + runTree ppend ppend' (Node (SubsequentStage pipe (Stage cp lpol ecpol)) forest) = case forest of + [] -> Halting $ \producer -> + blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ hoist lift producer >-> pipe) lpol) cp + c1 : cs -> Halting $ \producer -> + blende ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ hoist lift producer >-> pipe) (runNonEmpty ppend ppend' (c1 :| cs)) lpol) cp + + runNonEmpty ppend ppend' (b :| bs) = + runTree ppend ppend' b <* Prelude.foldr (<*) (pure ()) (runTree ppend' ppend' <$> bs) + + blende :: (Int -> Maybe e) -> Either e (ExitCode,()) -> Either e () + blende f (Right (ExitFailure i,())) = case f i of + Nothing -> Right () + Just e -> Left e + blende _ (Right (ExitSuccess,())) = Right () + blende _ (Left e) = Left e {- $reexports
− src/System/Process/Streaming/Tutorial.hs
@@ -1,266 +0,0 @@- --- | --- ------------------------------------------------------------------------------ - - -module System.Process.Streaming.Tutorial ( - -- * Introduction - -- $introduction - - -- * Stdin and stderr to different files - -- $stdinstderr - - -- * Missing executable - -- $missingexec - - -- * Combining stdout and stderr - -- $combinelines - - -- * Running two parsers in parallel - -- $forkProd - - -- * Aborting an execution - -- $fastExit - - -- * Feeding stdin, collecting stdout as text - -- $cat - - -- * Collecting stdout and stderr as bytestring - -- $bscollect - - -- * Counting words - -- $wordcount - - -- * ghci - -- $ghci - ) where - -{- $introduction -These examples require the @OverloadedStrings@ extension. - -Some preliminary imports: - -> module Main where -> -> import Data.Bifunctor -> import Data.Monoid -> import qualified Data.Attoparsec.Text as A -> import Control.Applicative -> import Control.Monad -> import Control.Lens (view) -> import Pipes -> import qualified Pipes.ByteString as B -> import qualified Pipes.Prelude as P -> import qualified Pipes.Parse as P -> import qualified Pipes.Attoparsec as P -> import qualified Pipes.Text as T -> import qualified Pipes.Text.Encoding as T -> import qualified Pipes.Text.IO as T -> import qualified Pipes.Group as G -> import qualified Pipes.Safe as S -> import qualified Pipes.Safe.Prelude as S -> import System.IO -> import System.IO.Error -> import System.Exit -> import System.Process.Streaming - --} - - -{- $stdinstderr - -Using 'separated' to consume @stdout@ and @stderr@ concurrently, and functions -from @pipes-safe@ to write the files. - -> example1 :: IO (Either String ((),())) -> example1 = simpleSafeExecute -> (pipeoe $ separated (consume "stdout.log") (consume "stderr.log")) -> (shell "{ echo ooo ; echo eee 1>&2 ; }") -> where -> consume file = surely . safely . useConsumer $ -> S.withFile file WriteMode B.toHandle - --} - - -{- $missingexec - -Missing executables and other 'IOException's are converted to an error type @e@ -and returned in the 'Left' of an 'Either': - -> example2 :: IO (Either String ()) -> example2 = simpleSafeExecute nopiping (proc "fsdfsdf" []) - -Returns: - ->>> Left "fsdfsdf: createProcess: runInteractiveProcess: exec: does not exist (No such file or directory)" - --} - - -{- $combinelines - -Here we use 'combined' to process 'stdout' and 'stderr' together. - -Notice that they are consumed together as 'Text'. We have to specify a decoding -function for each stream, and a 'LeftoverPolicy' as well. - -We also add a prefix to the lines coming from @stderr@. - -> example3 :: IO (Either String ()) -> example3 = simpleSafeExecute -> (pipeoe $ combined -> (linePolicy T.decodeIso8859_1 id policy) -> (linePolicy T.decodeIso8859_1 annotate policy) -> (surely . safely . useConsumer $ -> S.withFile "combined.txt" WriteMode T.toHandle)) -> (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") -> where -> policy = failOnLeftovers $ \_ _->"badbytes" -> annotate x = P.yield "errprefix: " *> x - --} - - -{- $forkProd - -Plugging parsers from @pipes-parse@ into 'separated' or 'combined' is easy -because running 'evalStateT' on a parser returns a function that consumes a -'Producer'. - -In this example we define two Attoparsec Text parsers and we convert them to -Pipes parsers using function 'parse' from package @pipes-attoparsec@. - -Stdout is decoded to Text and parsed by the two parsers in parallel using the -auxiliary 'forkSiphon' function. The results are aggregated in a tuple. - -Stderr is ignored using the 'nop' function. - -> parseChars :: Char -> A.Parser [Char] -> parseChars c = fmap mconcat $ -> many (A.notChar c) *> A.many1 (some (A.char c) <* many (A.notChar c)) -> -> parser1 = parseChars 'o' -> -> parser2 = parseChars 'a' -> -> example4 ::IO (Either String (([Char], [Char]),())) -> example4 = simpleSafeExecute -> (pipeoe $ separated -> (encoding T.decodeIso8859_1 (failOnLeftovers $ \_ _->"badbytes") $ -> forkSiphon (adapt parser1) (adapt parser2)) -> nop) -> (shell "{ echo ooaaoo ; echo aaooaoa; }") -> where -> adapt p = P.evalStateT $ do -> r <- P.parse p -> return $ case r of -> Just (Right r') -> Right r' -> _ -> Left "parse error" - -Returns: - ->>> Right (("ooooooo","aaaaaa"),()) - --} - - -{- $fastExit - -If any function consuming a standard stream returns with an error value @e@, -the external program is terminated and the computation returns immediately with -@e@. - -> example5 ::IO (Either String ((),())) -> example5 = simpleSafeExecute -> (pipeoe $ separated (\_ -> return $ Left "fast return!") nop) -> (shell "sleep 10s") - -Returns: - ->>> Left "fast return!" - -If we change the stdout consuming function to 'nop', 'example5' waits 10 -seconds. --} - - -{- $cat - -In this example we invoke the @cat@ command, feeding its input stream with a -'ByteString'. - -We decode stdout to Text and collect the whole output using a fold from -@pipes-text@. - -Plugging folds defined in "Pipes.Prelude" (or @pipes-bytestring@ or -@pipes-text@) into 'separated' or 'combined' is easy because the folds -return functions that consume 'Producer's. - -Notice that @stdin@ is written concurrently with the reading of @stdout@. It is -not the case that @sdtin@ is written first and then @stdout@ is read. - -> example6 = simpleSafeExecute -> (pipeioe -> (surely . useProducer $ yield "aaaaaa\naaaaa") -> (separated -> (encoding T.decodeIso8859_1 ignoreLeftovers $ surely $ T.toLazyM) -> nop)) -> (shell "cat") - -Returns: - ->>> Right ((),("aaaaaa\naaaaa",())) - --} - -{- $bscollect - -In this example we collect @stdout@ and @stderr@ as lazy bytestrings, using a -fold defined in @pipes-bytestring@. - -> example7 = simpleSafeExecute -> (pipeoe $ separated (surely B.toLazyM) (surely B.toLazyM)) -> (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") - -Returns: - ->>> Right ("ooo\nppp\n","eee\nffff\n") --} - - -{- $wordcount - - In this example we count words emitted to @stdout@ in a streaming fashion, -without having to keep whole words in memory. - - We use a lens from @pipes-text@ to split the text into words, and a trivial -fold from @pipes-group@ to create a 'Producer' of 'Int' values. Then we sum the -ints using a fold from "Pipes.Prelude". - -> example8 = simpleSafeExecute -> (pipeoe $ separated -> (encoding T.decodeIso8859_1 ignoreLeftovers $ surely $ -> P.sum . G.folds const () (const 1) . view T.words) -> nop) -> (shell "{ echo aaa ; echo bbb ; echo ccc ; }") - --} - -{- $ghci - -Sometimes it's useful to launch external programs during a ghci session, like -this: - ->>> a <- async $ execute nopiping (proc "xeyes" []) - -Cancelling the async causes the termination of the external program: - ->>> cancel a - -Waiting for the async returns the result: - ->>> wait a - --}
+ tests/test.hs view
@@ -0,0 +1,255 @@+{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ViewPatterns #-} + +module Main where + +import Test.Tasty +import Test.Tasty.HUnit + +import Data.Bifunctor +import Data.Monoid +import Data.Foldable +import Data.List.NonEmpty +import Data.ByteString +import Data.ByteString.Lazy as BL +import Data.Text.Lazy as TL +import Data.Typeable +import Data.Tree +import qualified Data.Attoparsec.Text as A +import Control.Applicative +import Control.Monad +import Control.Monad.Trans.Except +import Control.Lens (view) +import Control.Concurrent.Async +import Pipes +import qualified Pipes.ByteString as B +import qualified Pipes.Prelude as P +import qualified Pipes.Parse as P +import qualified Pipes.Attoparsec as P +import qualified Pipes.Text as T +import qualified Pipes.Text.Encoding as T +import qualified Pipes.Text.IO as T +import qualified Pipes.Group as G +import qualified Pipes.Safe as S +import qualified Pipes.Safe.Prelude as S +import System.IO +import System.IO.Error +import System.Exit +import System.Directory +import System.Process.Streaming + +main = defaultMain tests + +tests :: TestTree +tests = testGroup "Tests" + [ testCollectStdoutStderrAsByteString + , testFeedStdinCollectStdoutAsText + , testCombinedStdoutStderr + , testInterruptExecution + , testFailIfAnythingShowsInStderr + , testTwoTextParsersInParallel + , testCountWords + , testBasicPipeline + , testBranchingPipeline + ] + +------------------------------------------------------------------------------- +testCollectStdoutStderrAsByteString :: TestTree +testCollectStdoutStderrAsByteString = testCase "collectStdoutStderrAsByteString" $ do + r <- collectStdoutStderrAsByteString + case r of + (ExitSuccess,("ooo\nppp\n","eee\nffff\n")) -> return () + _ -> assertFailure "oops" + +collectStdoutStderrAsByteString :: IO (ExitCode,(BL.ByteString,BL.ByteString)) +collectStdoutStderrAsByteString = execute + (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) + (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") + + +------------------------------------------------------------------------------- +testFeedStdinCollectStdoutAsText :: TestTree +testFeedStdinCollectStdoutAsText = testCase "feedStdinCollectStdoutAsText" $ do + r <- feedStdinCollectStdoutAsText + case r of + (ExitSuccess,((),"aaaaaa\naaaaa")) -> return () + _ -> assertFailure "oops" + +feedStdinCollectStdoutAsText :: IO (ExitCode, ((), Text)) +feedStdinCollectStdoutAsText = execute + (pipeio (fromProducer $ yield "aaaaaa\naaaaa") + (encoded T.decodeIso8859_1 (pure id) $ fromFold T.toLazyM)) + (shell "cat") + +------------------------------------------------------------------------------- + +testCombinedStdoutStderr :: TestTree +testCombinedStdoutStderr = testCase "testCombinedStdoutStderr" $ do + r <- combinedStdoutStderr + case r of + (ExitSuccess,TL.lines -> ls) -> do + assertEqual "line count" (Prelude.length ls) 4 + assertBool "expected lines" $ + getAll $ foldMap (All . flip Prelude.elem ls) $ + [ "ooo" + , "ppp" + , "errprefix: eee" + , "errprefix: ffff" + ] + _ -> assertFailure "oops" + +combinedStdoutStderr :: IO (ExitCode,TL.Text) +combinedStdoutStderr = execute + (pipeoec (linePolicy T.decodeIso8859_1 (pure ()) id) + (linePolicy T.decodeIso8859_1 (pure ()) annotate) + (fromFold T.toLazyM)) + (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") + where + annotate x = P.yield "errprefix: " *> x + +------------------------------------------------------------------------------- + +testInterruptExecution :: TestTree +testInterruptExecution = localOption (mkTimeout $ 5*(10^6)) $ + testCase "interruptExecution" $ do + r <- interruptExecution + case r of + Left "interrupted" -> return () + _ -> assertFailure "oops" + +interruptExecution :: IO (Either String (ExitCode,())) +interruptExecution = executeFallibly + (pipeo . siphon $ \_ -> runExceptT . throwE $ "interrupted") + (shell "sleep 100s") + +------------------------------------------------------------------------------- + +testFailIfAnythingShowsInStderr :: TestTree +testFailIfAnythingShowsInStderr = localOption (mkTimeout $ 5*(10^6)) $ + testCase "failIfAnythingShowsInStderr" $ do + r <- failIfAnythingShowsInStderr + case r of + Left "morestuff\n" -> return () + _ -> assertFailure "oops" + +failIfAnythingShowsInStderr :: IO (Either T.ByteString (ExitCode,())) +failIfAnythingShowsInStderr = executeFallibly + (pipee (unwanted ())) + (shell "{ echo morestuff 1>&2 ; sleep 100s ; }") + +------------------------------------------------------------------------------- + +testTwoTextParsersInParallel :: TestTree +testTwoTextParsersInParallel = testCase "twoTextParsersInParallel" $ do + r <- twoTextParsersInParallel + case r of + Right (ExitSuccess,("ooooooo","aaaaaa")) -> return () + _ -> assertFailure "oops" + +parseChars :: Char -> A.Parser [Char] +parseChars c = fmap mconcat $ + many (A.notChar c) *> A.many1 (some (A.char c) <* many (A.notChar c)) + +parser1 = parseChars 'o' + +parser2 = parseChars 'a' + +twoTextParsersInParallel :: IO (Either String (ExitCode,([Char], [Char]))) +twoTextParsersInParallel = executeFallibly + (pipeo (encoded T.decodeIso8859_1 (pure id) $ + (,) <$> adapt parser1 <*> adapt parser2)) + (shell "{ echo ooaaoo ; echo aaooaoa; }") + where + adapt p = fromParser $ do + r <- P.parse p + return $ case r of + Just (Right r') -> Right r' + _ -> Left "parse error" + +------------------------------------------------------------------------------- + +testCountWords :: TestTree +testCountWords = testCase "testCountWords" $ do + r <- countWords + case r of + (ExitSuccess,3) -> return () + _ -> assertFailure "oops" + +countWords :: IO (ExitCode,Int) +countWords = execute + (pipeo (encoded T.decodeIso8859_1 (pure id) $ + fromFold $ P.sum . G.folds const () (const 1) . view T.words)) + (shell "{ echo aaa ; echo bbb ; echo ccc ; }") + +------------------------------------------------------------------------------- +testBasicPipeline :: TestTree +testBasicPipeline = testCase "basicPipeline" $ do + r <- basicPipeline + case r of + Right ((),"aaaccc\n") -> return () + _ -> assertFailure "oops" + +basicPipeline :: IO (Either String ((),BL.ByteString)) +basicPipeline = executePipelineFallibly + (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") + (fromFold B.toLazyM)) + (simplePipeline T.decodeUtf8 (shell "grep aaa") (pure . pure $ shell "grep ccc")) + +------------------------------------------------------------------------------- + +testBranchingPipeline :: TestTree +testBranchingPipeline = testCase "branchingPipeline" $ do + exists <- doesFileExist branchingPipelineFile + when exists $ removeFile branchingPipelineFile + r <- branchingPipeline + case r of + ("ppp\v","eee\nffff\n") -> return () + _ -> assertFailure "oops" + fileContents <- withFile branchingPipelineFile ReadMode $ \hIn -> do + B.toLazyM $ B.fromHandle hIn + assertBool "file contexts" $ BL.isPrefixOf "yyy" fileContents + +branchingPipelineFile :: String +branchingPipelineFile = "dist/test/process-streaming-pipeline-text.txt" + +branchingPipeline :: IO (BL.ByteString, BL.ByteString) +branchingPipeline = executePipeline + (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) + (CreatePipeline rootStage . fromList $ + [ Node branch1 [pure terminalStage1] , Node branch2 [pure terminalStage2] ] ) + where + succStage = SubsequentStage (P.map (Data.ByteString.map succ)) + + rootStage :: (Show e, Typeable e) => Stage e + rootStage = Stage (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ; echo ffff 1>&2 ; }") + (linePolicy T.decodeIso8859_1 (pure ()) id) + (\_ -> Nothing) + + branch1 :: (Show e, Typeable e) => SubsequentStage e + branch1 = SubsequentStage cat $ + Stage (shell "grep ooo") + (linePolicy T.decodeIso8859_1 (pure ()) id) + (\_ -> Nothing) + + branch2 :: (Show e, Typeable e) => SubsequentStage e + branch2 = SubsequentStage cat $ + Stage (shell "grep xxx") + (linePolicy T.decodeIso8859_1 (pure ()) id) + (\_ -> Nothing) + + terminalStage1 :: (Show e, Typeable e) => SubsequentStage e + terminalStage1 = succStage $ + Stage (shell "tr -d b") + (linePolicy T.decodeIso8859_1 (pure ()) id) + (\_ -> Nothing) + + terminalStage2 :: (Show e, Typeable e) => SubsequentStage e + terminalStage2 = succStage $ + Stage (shell $ "cat > " ++ branchingPipelineFile) + (linePolicy T.decodeIso8859_1 (pure ()) id) + (\_ -> Nothing) + + + + +