diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,12 @@ +0.9.0.0 +------- +- All Siphon functionality has been moved to new package pipes-transduce. +- Siphons correspond to the "Fold1" datatype in pipes-transduce. +- "Piping" is now the "Streams" Applicative. +- Exit code handling has been merged into the "Streams" Applicative. +- Removed pipeline functions. +- Removed some lenses, added others. + 0.7.2.1 ------- - Asynchronous cancellation now works on Windows (issue #8) diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,22 +1,27 @@ process-streaming ================= -Exploring how to interact with system processes using a streaming library -(pipes). +A library for interacting with system processes in a streaming fashion. The basic goals: -- Concurrent, streaming access to stdin, stdout and stderr. +- Concurrent, streaming access to stdin, stdout and stderr... -- Easy integration with regular consumers, parsers from pipes-parse and various - folds. +- ...all the while preventing deadlocks caused by mishandling of the streams. -- Avoid launching exceptions: use Either or similar solution to signal non-IO - related error conditions. +- Easy integration with consumers from + [pipes](http://hackage.haskell.org/package/pipes), parsers from + [pipes-parse](http://hackage.haskell.org/package/pipes-parse) and folds from + [foldl](http://hackage.haskell.org/package/foldl). -Relevant thread in the Haskell Pipes Google Group: +- Facilitate the use of sum types to signal failures, when desired. -https://groups.google.com/forum/#!searchin/haskell-pipes/pipes$20process/haskell-pipes/JFfyquj5HAg/Lxz7p50JOh4J +- No fussing around with process handles: wait for the process by waiting for + the IO action, terminate the process by killing the thread executing the IO + action. + +A [relevant thread](https://groups.google.com/forum/#!searchin/haskell-pipes/pipes$20process/haskell-pipes/JFfyquj5HAg/Lxz7p50JOh4J) in the Haskell Pipes Google Group. + ## Possible alternatives in Hackage diff --git a/process-streaming.cabal b/process-streaming.cabal --- a/process-streaming.cabal +++ b/process-streaming.cabal @@ -1,5 +1,5 @@ name: process-streaming -version: 0.7.2.2 +version: 0.9.0.0 license: BSD3 license-file: LICENSE data-files: @@ -19,34 +19,31 @@ default-language: Haskell2010 hs-source-dirs: src exposed-modules: - System.Process.Streaming.Extended - System.Process.Streaming.Internal - System.Process.Streaming.Tutorial System.Process.Streaming + System.Process.Streaming.Text System.Process.Lens other-modules: ghc-options: -Wall -threaded -O2 build-depends: - base >= 4.4 && < 5, - transformers >= 0.2 && < 0.5, - transformers-compat >= 0.3, - free >= 4.2 && < 5, - bifunctors >= 4.1 && < 6, - process >= 1.2.0 && < 1.3, - pipes >= 4.1.3 && < 4.2, - pipes-bytestring >= 2.1.0 && < 2.2, - pipes-text >= 0.0.0.10 && < 0.0.2, - bytestring, - text >= 0.11.2 && < 1.3, - pipes-concurrency >= 2.0.2 && < 3, - pipes-safe >= 2.2.0 && < 3, - pipes-parse >=3.0.1 && <3.1, - void >= 0.6 && < 1.0, - containers >= 0.4, - semigroups >= 0.15 && < 0.20, - conceit >= 0.3.2.0 && < 0.4.0.0, - contravariant >= 1.2, - foldl >= 1.0.7 + base >= 4.4 && < 5 + , process >= 1.2.3.0 + , bytestring + , text >= 0.11.2 + , void >= 0.6 + , transformers >= 0.2 + , transformers-compat >= 0.3 + , free >= 4.2 + , kan-extensions > 4.2 + , profunctors >= 5 + , bifunctors >= 4.1 + , pipes >= 4.1.3 + , pipes-safe >= 2.2.0 + , pipes-parse >=3.0.1 + , pipes-concurrency >= 2.0.2 + , pipes-bytestring >=2.1.0 + , pipes-text >= 0.0.0.10 + , pipes-transduce >= 0.3.3.0 + , conceit >= 0.3.2.0 Test-suite test default-language: @@ -60,31 +57,33 @@ ghc-options: -Wall -threaded build-depends: base >= 4.4 && < 5 - , transformers >= 0.2 && < 0.5 + , process >= 1.2.3.0 + , transformers >= 0.2 , transformers-compat >= 0.3 - , free >= 4.2 && < 5 - , bifunctors >= 4.1 && < 6 - , 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.3 - , pipes-concurrency >= 2.0.3 && < 3 - , pipes-safe >= 2.2.0 && < 3 - , pipes-parse >=3.0.1 && <3.1 + , bytestring + , text >= 0.11.2 + , void >= 0.6 + , free >= 4.2 + , bifunctors >= 4.1 + , foldl >= 1.1 + , pipes >= 4.1.3 + , pipes-safe >= 2.2.0 + , pipes-parse >=3.0.1 + , pipes-concurrency >= 2.0.2 + , pipes-bytestring >=2.1.0 + , pipes-text >= 0.0.0.10 + , pipes-group >= 1.0.1 + , pipes-attoparsec >= 0.5 + , pipes-transduce >= 0.3.3.0 , exceptions >= 0.6.0 && < 1.0 - , void >= 0.6 && < 1.0 , semigroups >= 0.15 && < 0.20 , 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 + , lens-family-core >= 1.1 , directory >= 1.2 + , tasty >= 0.9 + , tasty-hunit >= 0.9 + , process-streaming test-suite doctests type: exitcode-stdio-1.0 @@ -94,33 +93,35 @@ build-depends: base >= 4.4 && < 5 - , transformers >= 0.2 && < 0.5 + , process >= 1.2.3.0 + , transformers >= 0.2 , transformers-compat >= 0.3 + , bytestring + , text >= 0.11.2 + , void >= 0.6 , free >= 4.2 && < 5 - , bifunctors >= 4.1 && < 6 - , 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.3 - , pipes-concurrency >= 2.0.3 && < 3 - , pipes-safe >= 2.2.0 && < 3 - , pipes-parse >=3.0.1 && <3.1 - , exceptions >= 0.6.0 && < 1 - , void >= 0.6 && < 1.0 - , semigroups >= 0.15 && < 0.20 + , bifunctors >= 4.1 + , foldl >= 1.1 + , pipes >= 4.1.3 + , pipes-safe >= 2.2.0 + , pipes-parse >=3.0.1 + , pipes-concurrency >= 2.0.2 + , pipes-bytestring >=2.1.0 + , pipes-text >= 0.0.0.10 + , pipes-group >= 1.0.1 + , pipes-attoparsec >= 0.5 + , pipes-transduce >= 0.3.3.0 + , exceptions >= 0.6.0 + , semigroups >= 0.15 , 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 + , lens-family-core >= 1.1 , directory >= 1.2 , filepath , doctest >= 0.9.1 + , tasty >= 0.9 + , tasty-hunit >= 0.9 + , process-streaming Source-repository head type: git diff --git a/src/System/Process/Lens.hs b/src/System/Process/Lens.hs --- a/src/System/Process/Lens.hs +++ b/src/System/Process/Lens.hs @@ -2,13 +2,13 @@ -- | -- Lenses and traversals for 'CreateProcess' and related types. -- --- These are provided as a convenience and aren't required to use the other --- modules of the package. +-- These are provided as a convenience and aren't at all required to use the +-- other modules of this package. -- ----------------------------------------------------------------------------- -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE CPP #-} module System.Process.Lens ( _cmdspec @@ -16,39 +16,33 @@ , _RawCommand , _cwd , _env - , streams + , std_streams + , _std_in + , _std_out + , _std_err , _close_fds , _create_group , _delegate_ctlc - , handles - , nohandles - , handleso - , handlese - , handlesoe - , handlesi - , handlesio - , handlesie - , handlesioe +#if MIN_VERSION_process(1,3,0) + , _detach_console + , _create_new_console + , _new_session +#endif ) where -import Data.Maybe -import Data.Functor.Identity -import Data.Monoid -import Data.Traversable import Control.Applicative -import System.IO import System.Process {-| > _cmdspec :: Lens' CreateProcess CmdSpec -} _cmdspec :: forall f. Functor f => (CmdSpec -> f CmdSpec) -> CreateProcess -> f CreateProcess -_cmdspec f c = setCmdSpec c <$> f (cmdspec c) +_cmdspec f x = setCmdSpec x <$> f (cmdspec x) where setCmdSpec c cmdspec' = c { cmdspec = cmdspec' } {-| - > _ShellCommand :: Prism' CmdSpec String + > _ShellCommand :: Traversal' CmdSpec String -} _ShellCommand :: forall m. Applicative m => (String -> m String) -> CmdSpec -> m CmdSpec _ShellCommand f quad = case impure quad of @@ -59,7 +53,7 @@ impure x = Left x {-| - > _RawCommand :: Prism' CmdSpec (FilePath,[String]) + > _RawCommand :: Traversal' CmdSpec (FilePath,[String]) -} _RawCommand :: forall m. Applicative m => ((FilePath,[String]) -> m (FilePath,[String])) -> CmdSpec -> m CmdSpec _RawCommand f quad = case impure quad of @@ -74,7 +68,7 @@ > _cwd :: Lens' CreateProcess (Maybe FilePath) -} _cwd :: forall f. Functor f => (Maybe FilePath -> f (Maybe FilePath)) -> CreateProcess -> f CreateProcess -_cwd f c = setCwd c <$> f (cwd c) +_cwd f x = setCwd x <$> f (cwd x) where setCwd c cwd' = c { cwd = cwd' } @@ -82,17 +76,17 @@ > _env :: Lens' CreateProcess (Maybe [(String,String)]) -} _env :: forall f. Functor f => (Maybe [(String, String)] -> f (Maybe [(String, String)])) -> CreateProcess -> f CreateProcess -_env f c = setEnv c <$> f (env c) +_env f x = setEnv x <$> f (env x) where setEnv c env' = c { env = env' } {-| A lens for the @(std_in,std_out,std_err)@ triplet. - > streams :: Lens' CreateProcess (StdStream,StdStream,StdStream) + > std_streams :: Lens' CreateProcess (StdStream,StdStream,StdStream) -} -streams :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess -streams f c = setStreams c <$> f (getStreams c) +std_streams :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess +std_streams f x = setStreams x <$> f (getStreams x) where getStreams c = (std_in c,std_out c, std_err c) setStreams c (s1,s2,s3) = c { std_in = s1 @@ -100,133 +94,54 @@ , std_err = s3 } +_std_in :: forall f. Functor f => (StdStream -> f (StdStream)) -> CreateProcess -> f CreateProcess +_std_in f x = setStreams x <$> f (getStreams x) + where + getStreams c = std_in c + setStreams c s1 = c { std_in = s1 } + +_std_out :: forall f. Functor f => (StdStream -> f (StdStream)) -> CreateProcess -> f CreateProcess +_std_out f x = setStreams x <$> f (getStreams x) + where + getStreams c = std_out c + setStreams c s1 = c { std_out = s1 } + +_std_err :: forall f. Functor f => (StdStream -> f (StdStream)) -> CreateProcess -> f CreateProcess +_std_err f x = setStreams x <$> f (getStreams x) + where + getStreams c = std_err c + setStreams c s1 = c { std_err = s1 } + _close_fds :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess -_close_fds f c = set_close_fds c <$> f (close_fds c) +_close_fds f x = set_close_fds x <$> f (close_fds x) where - set_close_fds c cwd' = c { close_fds = cwd' } + set_close_fds c v = c { close_fds = v } _create_group :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess -_create_group f c = set_create_group c <$> f (create_group c) +_create_group f x = set_create_group x <$> f (create_group x) where - set_create_group c cwd' = c { create_group = cwd' } + set_create_group c v = c { create_group = v } _delegate_ctlc :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess -_delegate_ctlc f c = set_delegate_ctlc c <$> f (delegate_ctlc c) +_delegate_ctlc f x = set_delegate_ctlc x <$> f (delegate_ctlc x) where - set_delegate_ctlc c cwd' = c { delegate_ctlc = cwd' } - -{-| - A 'Lens' for the return value of 'createProcess' that focuses on the handles. + set_delegate_ctlc c v = c { delegate_ctlc = v } - > 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) +#if MIN_VERSION_process(1,3,0) +_detach_console :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_detach_console f x = set_detach_console x <$> f (detach_console x) where - setHandles (c1'',c2'',c3'',c4'') (c1',c2',c3') = (c1',c2',c3',c4'') - 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 - Right r -> fmap justify (f r) - where - impure (Nothing, Nothing, Nothing) = Right () - impure x = Left x - justify () = (Nothing, Nothing, Nothing) - - -{-| - A 'Prism' that matches when only @stdin@ has been piped. - - > 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 - Right r -> fmap justify (f r) - where - impure (Just h1, Just h2, Just h3) = Right (h1, h2, h3) - impure x = Left x - justify (h1, h2, h3) = (Just h1, Just h2, Just h3) - -{-| - A 'Prism' that matches when only @stdout@ and @stderr@ have been piped. - - > 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 - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Nothing, Just h2, Just h3) = Right (h2, h3) - impure x = Left x - justify (h2, h3) = (Nothing, Just h2, Just h3) - -{-| - A 'Prism' that matches when only @stdout@ has been piped. + set_detach_console c v = c { detach_console = v } - > 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) +_create_new_console :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_create_new_console f x = set_create_new_console x <$> f (create_new_console x) + where + set_create_new_console c v = c { create_new_console = v } -{-| - A 'Prism' that matches when only @stderr@ has been piped. +_new_session :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_new_session f x = set_new_session x <$> f (new_session x) + where + set_new_session c v = c { new_session = v } +#endif - > 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) diff --git a/src/System/Process/Streaming.hs b/src/System/Process/Streaming.hs --- a/src/System/Process/Streaming.hs +++ b/src/System/Process/Streaming.hs @@ -1,1033 +1,475 @@ - --- | --- This module contains helper functions and types built on top of --- "System.Process" and "Pipes". --- --- They provide concurrent, streaming access to the inputs and outputs of --- system processes. --- --- Error conditions other than 'IOException's are made explicit --- in the types. --- --- Regular 'Consumer's, 'Parser's from @pipes-parse@ and various folds can --- be used to consume the output streams of the external processes. --- ------------------------------------------------------------------------------ - -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE ViewPatterns #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE FlexibleInstances #-} - - -module System.Process.Streaming ( - -- * Execution - executeFallibly - , execute - -- * Piping the standard streams - , Piping - , nopiping - , pipeo - , pipee - , pipeoe - , pipeoec - , pipei - , pipeio - , pipeie - , pipeioe - , pipeioec - - -- * Pumping bytes into stdin - , Pump - , fromProducer - , fromProducerM - , fromSafeProducer - , fromFallibleProducer - , fromFoldable - , fromEnumerable - , fromLazyBytes - -- * Siphoning bytes out of stdout/stderr - , Siphon - , siphon - , siphon' - , fromFold - , fromFold' - , fromFold'_ - , fromConsumer - , fromConsumer' - , fromConsumerM - , fromConsumerM' - , fromSafeConsumer - , fromFallibleConsumer - , fromParser - , fromParserM - , fromFoldl - , fromFoldlIO - , fromFoldlM - , intoLazyBytes - , intoLazyText - , intoList - , unwanted - , DecodingFunction - , encoded - , SiphonOp (..) - , contramapFoldable - , contramapEnumerable - , contraproduce - , contraencoded - , Splitter - , splitter - , splitIntoLines - , tweakSplits - , rejoin - , nest - -- * Handling lines - , Lines - , toLines - , tweakLines - , prefixLines - -- * Throwing exceptions - , unwantedX - , LeftoverException (..) - , leftoverX - , _leftoverX - -- * Pipelines - , executePipelineFallibly - , executePipeline - --, simplePipeline - , Stage - , stage - , pipefail - , inbound - -- * Re-exports - -- $reexports - , module System.Process - , T.decodeUtf8 - , T.decodeAscii - , T.decodeIso8859_1 - ) where - -import qualified Data.ByteString.Lazy as BL -import Data.Functor.Contravariant -import Data.Functor.Contravariant.Divisible -import Data.Monoid -import Data.Foldable -import Data.Typeable -import Data.Tree -import qualified Data.Text.Lazy as TL -import Data.Text -import Data.Text.Encoding hiding (decodeUtf8) -import Data.Void -import Data.List.NonEmpty -import Control.Applicative -import Control.Applicative.Lift -import Control.Monad -import Control.Monad.Trans.Free hiding (Pure) -import qualified Control.Monad.Trans.Free as FREE -import Control.Monad.Trans.Except -import qualified Control.Foldl as L -import Control.Exception -import Control.Concurrent -import Control.Concurrent.Conceit -import Pipes -import qualified Pipes as P -import qualified Pipes.Prelude as P -import Pipes.ByteString -import Pipes.Parse -import qualified Pipes.Text as T -import qualified Pipes.Text.Encoding as T -import Pipes.Concurrent -import Pipes.Safe (SafeT, runSafeT) -import System.IO -import System.Process -import System.Process.Lens -import System.Exit - -import System.Process.Streaming.Internal - -{-| - A simplified version of 'executeFallibly' for when the error type unifies - with `Void`. Note however that this function may still throw exceptions. - -} -execute :: Piping 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 'Piping' argument. - - This function re-throws any 'IOException's it encounters. - - Besides exceptions, if the consumption of the standard streams fails - with @e@, the whole computation is immediately aborted and @e@ is - returned. - - If an exception or an error @e@ happen, the external process is -terminated. - -} -executeFallibly :: Piping 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 - (mi,mout,merr,phandle) <- createProcess record - case getFirst . getConst . somePrism (Const . First . Just) $ (mi,mout,merr) of - Nothing -> - throwIO (userError "stdin/stdout/stderr handle unwantedly null") - `finally` - terminateCarefully phandle - Just t -> do - latch <- newEmptyMVar - let (action,cleanup) = allocator t - innerRace = _runConceit $ - (_Conceit (takeMVar latch >> terminateOnError phandle action)) - <|> - (_Conceit (onException (putMVar latch () >> _runConceit Control.Applicative.empty) - (terminateCarefully phandle))) - -- 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 innerRace `onException` terminateCarefully phandle) `finally` cleanup - - - -terminateCarefully :: ProcessHandle -> IO () -terminateCarefully pHandle = do - mExitCode <- getProcessExitCode pHandle - case mExitCode of - Nothing -> catch - (terminateProcess pHandle) - (\(_::IOException) -> return ()) - Just _ -> return () - -terminateOnError :: ProcessHandle - -> IO (Either e a) - -> IO (Either e (ExitCode,a)) -terminateOnError pHandle action = do - result <- action - case result of - Left e -> do - terminateCarefully pHandle - return $ Left e - Right r -> do - exitCode <- waitForProcess pHandle - return $ Right (exitCode,r) - -{-| - Do not pipe any standard stream. --} -nopiping :: Piping e () -nopiping = PPNone () - -{-| - Pipe @stdout@. --} -pipeo :: Siphon ByteString e a -> Piping e a -pipeo (runSiphonDumb -> siphonout) = PPOutput $ siphonout - -{-| - Pipe @stderr@. --} -pipee :: Siphon ByteString e a -> Piping e a -pipee (runSiphonDumb -> siphonout) = PPError $ siphonout - -{-| - Pipe @stdout@ and @stderr@. --} -pipeoe :: Siphon ByteString e a -> Siphon ByteString e b -> Piping e (a,b) -pipeoe (runSiphonDumb -> siphonout) (runSiphonDumb -> siphonerr) = - PPOutputError $ uncurry $ separated siphonout siphonerr - -{-| - Pipe @stdout@ and @stderr@ and consume them combined as 'Text'. --} -pipeoec :: Lines e -> Lines e -> Siphon Text e a -> Piping e a -pipeoec policy1 policy2 (runSiphonDumb -> s) = - PPOutputError $ uncurry $ combined policy1 policy2 s - -{-| - Pipe @stdin@. --} -pipei :: Pump ByteString e i -> Piping e i -pipei (Pump feeder) = PPInput $ \(consumer,cleanup) -> feeder consumer `finally` cleanup - -{-| - Pipe @stdin@ and @stdout@. --} -pipeio :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a) -pipeio (Pump feeder) (runSiphonDumb -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) -> - (conceit (feeder consumer `finally` cleanup) (siphonout producer)) - -{-| - Pipe @stdin@ and @stderr@. --} -pipeie :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a) -pipeie (Pump feeder) (runSiphonDumb -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) -> - (conceit (feeder consumer `finally` cleanup) (siphonerr producer)) - -{-| - Pipe @stdin@, @stdout@ and @stderr@. --} -pipeioe :: Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> Piping e (i,a,b) -pipeioe (Pump feeder) (runSiphonDumb -> siphonout) (runSiphonDumb -> 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 :: Pump ByteString e i -> Lines e -> Lines e -> Siphon Text e a -> Piping e (i,a) -pipeioec (Pump feeder) policy1 policy2 (runSiphonDumb -> s) = PPInputOutputError $ - \(consumer,cleanup,outprod,errprod) -> - (conceit (feeder consumer `finally` cleanup) - (combined policy1 policy2 s outprod errprod)) - -separated :: (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 (outfunc outprod) (errfunc errprod) - -fromProducer :: Producer b IO r -> Pump b e () -fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) - -fromProducerM :: MonadIO m => (m () -> IO (Either e a)) -> Producer b m r -> Pump b e a -fromProducerM whittle producer = Pump $ \consumer -> whittle $ runEffect (mute producer >-> hoist liftIO consumer) - -fromSafeProducer :: Producer b (SafeT IO) r -> Pump b e () -fromSafeProducer = fromProducerM (fmap pure . runSafeT) - -fromFallibleProducer :: Producer b (ExceptT e IO) r -> Pump b e () -fromFallibleProducer = fromProducerM runExceptT - -fromFoldable :: Foldable f => f b -> Pump b e () -fromFoldable = fromProducer . each - -fromEnumerable :: Enumerable t => t IO b -> Pump b e () -fromEnumerable = fromProducer . every - -fromLazyBytes :: BL.ByteString -> Pump ByteString e () -fromLazyBytes = fromProducer . fromLazy - - -{-| - Collects incoming 'BS.ByteString' values into a lazy 'BL.ByteString'. --} -intoLazyBytes :: Siphon ByteString e BL.ByteString -intoLazyBytes = fromFoldl (fmap BL.fromChunks L.list) - -{-| - Collects incoming 'Data.Text' values into a lazy 'TL.Text'. --} -intoLazyText :: Siphon Text e TL.Text -intoLazyText = fromFoldl (fmap TL.fromChunks L.list) - -intoList :: Siphon b e [b] -intoList = fromFoldl L.list - -{-| - 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 f = Siphon (Other (Nonexhaustive f)) - -{-| - Builds a 'Siphon' out of a computation that drains a 'Producer' completely, -but may fail with an error of type @e@. - - This functions incurs in less overhead than 'siphon'. --} -siphon' :: (forall r. Producer b IO r -> IO (Either e (a,r))) -> Siphon b e a -siphon' f = Siphon (Other (Exhaustive f)) - -{-| - Useful in combination with folds from the pipes prelude, or more - specialized folds like 'Pipes.Text.toLazyM' from @pipes-text@ and - 'Pipes.ByteString.toLazyM' from @pipes-bytestring@. --} -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 - - -{-| - Builds a 'Siphon' out of a pure fold from the @foldl@ package. --} -fromFoldl :: L.Fold b a -> Siphon b e a -fromFoldl aFold = fromFold' $ L.purely P.fold' aFold - -{-| - Builds a 'Siphon' out of a monadic fold from the @foldl@ package that - works in the IO monad. --} -fromFoldlIO :: L.FoldM IO b a -> Siphon b e a -fromFoldlIO aFoldM = fromFold' $ L.impurely P.foldM' aFoldM - - -{-| - Builds a 'Siphon' out of a monadic fold from the @foldl@ package. --} -fromFoldlM :: MonadIO m - => (forall r. m (a,r) -> IO (Either e (c,r))) - -> L.FoldM m b a - -> Siphon b e c -fromFoldlM whittle aFoldM = siphon' $ \producer -> - whittle $ L.impurely P.foldM' aFoldM (hoist liftIO producer) - -fromConsumer :: Consumer b IO () -> Siphon b e () -fromConsumer consumer = fromFold $ \producer -> runEffect $ producer >-> consumer - -{-| - Builds a 'Siphon' out of a 'Consumer' with a polymorphic return type - (one example is 'toHandle' from @pipes-bytestring@). --} -fromConsumer' :: Consumer b IO Void -> Siphon b e () -fromConsumer' consumer = fromFold'_$ \producer -> runEffect $ producer >-> fmap absurd consumer - -fromConsumerM :: MonadIO m - => (m () -> IO (Either e a)) - -> Consumer b m () - -> Siphon b e a -fromConsumerM whittle consumer = siphon $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> consumer - -fromConsumerM' :: MonadIO m - => (forall r. m r -> IO (Either e (a,r))) - -> Consumer b m Void - -> Siphon b e a -fromConsumerM' whittle consumer = siphon' $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> fmap absurd consumer - -fromSafeConsumer :: Consumer b (SafeT IO) Void -> Siphon b e () -fromSafeConsumer = fromConsumerM' (fmap (\r -> Right ((),r)) . runSafeT) - -fromFallibleConsumer :: Consumer b (ExceptT e IO) Void -> Siphon b e () -fromFallibleConsumer = fromConsumerM' (fmap (fmap (\r -> ((), r))) . runExceptT) - -{-| - Turn a 'Parser' from @pipes-parse@ into a 'Siphon'. - -} -fromParser :: Parser b IO (Either e a) -> Siphon b e a -fromParser parser = siphon' $ \producer -> drainage $ Pipes.Parse.runStateT parser producer - where - drainage m = do - (a,leftovers) <- m - r <- runEffect (leftovers >-> P.drain) - case a of - Left e -> return (Left e) - Right a' -> return (Right (a',r)) - -{-| - Turn a 'Parser' from @pipes-parse@ into a 'Siphon'. - -} -fromParserM :: MonadIO m - => (forall r. m (a,r) -> IO (Either e (c,r))) - -> Parser b m a -> Siphon b e c -fromParserM f parser = siphon' $ \producer -> f $ drainage $ (Pipes.Parse.runStateT parser) (hoist liftIO producer) - where - drainage m = do - (a,leftovers) <- m - r <- runEffect (leftovers >-> P.drain) - return (a,r) - -{-| - Constructs a 'Siphon' that aborts the computation with an explicit error - if the underlying 'Producer' produces anything. - -} -unwanted :: a -> Siphon b b a -unwanted a = siphon' $ \producer -> do - n <- next producer - return $ case n of - Left r -> Right (a,r) - Right (b,_) -> Left b - -{-| - Like 'unwanted', but throws an exception instead of using the explicit - error type. --} -unwantedX :: Exception ex => (b -> ex) -> a -> Siphon b e a -unwantedX f a = siphon' $ \producer -> do - n <- next producer - case n of - Left r -> return $ Right (a,r) - Right (b,_) -> throwIO (f b) - -{-| - Exception that carries a message and a sample of the leftover data. --} -data LeftoverException b = LeftoverException String b deriving (Typeable) - -instance (Typeable b) => Exception (LeftoverException b) - -instance (Typeable b) => Show (LeftoverException b) where - show (LeftoverException msg _) = - "[Leftovers of type " ++ typeName (Proxy::Data.Typeable.Proxy b) ++ "]" ++ msg' - where - typeName p = showsTypeRep (typeRep p) [] - msg' = case msg of - [] -> [] - _ -> " " ++ msg - -{-| - Throws 'LeftoverException' if any data comes out of the underlying - producer, and returns 'id' otherwise. --} -leftoverX :: String - -- ^ Error message - -> Siphon ByteString e (a -> a) -leftoverX msg = unwantedX (LeftoverException msg') id - where - msg' = "leftoverX." ++ case msg of - "" -> "" - _ -> " " ++ msg - -{-| - Like 'leftoverX', but doesn't take an error message. --} -_leftoverX :: Siphon ByteString e (a -> a) -_leftoverX = unwantedX (LeftoverException msg) id - where - msg = "_leftoverX." - -{-| - 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) - -{-| - Constructs a 'Siphon' that works on encoded values out of a 'Siphon' that -works on decoded values. - -} -encoded :: DecodingFunction bytes text - -- ^ A decoding function. - -> Siphon bytes e (a -> b) - -- ^ A 'Siphon' that determines how to handle decoding leftovers. - -- Pass @pure id@ to ignore leftovers. Pass @unwanted id@ to abort - -- the computation with an explicit error if leftovers remain. Pass - -- '_leftoverX' to throw a 'LeftoverException' if leftovers remain. - -> Siphon text e a - -> Siphon bytes e b -encoded decoder (Siphon (unLift -> policy)) (Siphon (unLift -> activity)) = - Siphon (Other internal) - where - internal = Exhaustive $ \producer -> runExceptT $ do - (a,leftovers) <- ExceptT $ exhaustive activity $ decoder producer - (f,r) <- ExceptT $ exhaustive policy leftovers - pure (f a,r) - - -{-| - Like encoded, but works on 'SiphonOp's. - -} -contraencoded :: DecodingFunction bytes text - -- ^ A decoding function. - -> Siphon bytes e (a -> b) - -- ^ A 'Siphon' that determines how to handle decoding leftovers. - -- Pass @pure id@ to ignore leftovers. Pass @unwanted id@ to abort - -- the computation with an explicit error if leftovers remain. Pass - -- '_leftoverX' to throw a 'LeftoverException' if leftovers remain. - -> SiphonOp e a text - -> SiphonOp e b bytes -contraencoded decoder leftovers (SiphonOp siph) = SiphonOp $ - encoded decoder leftovers siph - - -{-| - Build a 'Splitter' out of a function that splits a 'Producer' while - preserving streaming. - - See the section /FreeT Transformations/ in the documentation for the - /pipes-text/ package, and also the documentation for the /pipes-group/ - package. --} -splitter :: (forall r. Producer b IO r -> FreeT (Producer b IO) IO r) -> Splitter b -splitter = Splitter - -{-| - Specifies a transformation that will be applied to each individual - split, represented as a 'Producer'. --} -tweakSplits :: (forall r. Producer b IO r -> Producer b IO r) -> Splitter b -> Splitter b -tweakSplits f (Splitter s) = Splitter $ fmap (transFreeT f) s - - -{-| - Flattens the 'Splitter', returning a function from 'Producer' to - 'Producer' which can be passed to functions like 'contraproduce'. --} -rejoin :: forall b r. Splitter b -> Producer b IO r -> Producer b IO r -rejoin (Splitter f) = go . f - where - -- code copied from the "concats" function from the pipes-group package - go f = do - x <- lift (runFreeT f) - case x of - FREE.Pure r -> return r - Free p -> do - f' <- p - go f' - - -splitIntoLines :: Splitter T.Text -splitIntoLines = splitter $ getConst . T.lines Const - - -{-| - Process each individual split created by a 'Splitter' using a 'Siphon'. - -} -nest :: Splitter b -> Siphon b Void a -> SiphonOp e r a -> SiphonOp e r b -nest (Splitter sp) nested = - contraproduce $ \producer -> iterT runRow (hoistFreeT lift $ sp producer) - where - runRow p = do - (r, innerprod) <- lift $ fmap (either absurd id) (runSiphon nested p) - P.yield r >> innerprod - -{-| - A newtype wrapper with functions for working on the inputs of - a 'Siphon', instead of the outputs. - -} -newtype SiphonOp e a b = SiphonOp { getSiphonOp :: Siphon b e a } - --- | 'contramap' carn turn a 'SiphonOp' for bytes into a 'SiphonOp' for text. -instance Contravariant (SiphonOp e a) where - contramap f (SiphonOp (Siphon s)) = SiphonOp . Siphon $ case s of - Pure p -> Pure p - Other o -> Other $ case o of - Exhaustive e -> Exhaustive $ \producer -> - e $ producer >-> P.map f - Nonexhaustive ne -> Nonexhaustive $ \producer -> - ne $ producer >-> P.map f - --- | 'divide' builds a 'SiphonOp' for a composite out of the 'SiphonOp's --- for the parts. -instance Monoid a => Divisible (SiphonOp e a) where - divide divider siphonOp1 siphonOp2 = contramap divider . SiphonOp $ - (getSiphonOp (contramap fst siphonOp1)) - `mappend` - (getSiphonOp (contramap snd siphonOp2)) - conquer = SiphonOp (pure mempty) - --- | 'choose' builds a 'SiphonOp' for a sum out of the 'SiphonOp's --- for the branches. -instance Monoid a => Decidable (SiphonOp e a) where - choose chooser (SiphonOp s1) (SiphonOp s2) = - contramap chooser . SiphonOp $ - (contraPipeMapL s1) - `mappend` - (contraPipeMapR s2) - where - contraPipeMapL (Siphon s) = Siphon $ case s of - Pure p -> Pure p - Other o -> Other $ case o of - Exhaustive e -> Exhaustive $ \producer -> - e $ producer >-> allowLefts - Nonexhaustive ne -> Nonexhaustive $ \producer -> - ne $ producer >-> allowLefts - contraPipeMapR (Siphon s) = Siphon $ case s of - Pure p -> Pure p - Other o -> Other $ case o of - Exhaustive e -> Exhaustive $ \producer -> - e $ producer >-> allowRights - Nonexhaustive ne -> Nonexhaustive $ \producer -> - ne $ producer >-> allowRights - allowLefts = do - e <- await - case e of - Left l -> Pipes.yield l >> allowLefts - Right _ -> allowLefts - allowRights = do - e <- await - case e of - Right r -> Pipes.yield r >> allowRights - Left _ -> allowRights - lose f = SiphonOp . Siphon . Other . Nonexhaustive $ \producer -> do - n <- next producer - return $ case n of - Left () -> Right mempty - Right (b,_) -> Right (absurd (f b)) - -{-| - Useful to weed out unwanted inputs to a 'Siphon', by returning @[]@. --} -contramapFoldable :: Foldable f => (a -> f b) -> SiphonOp e r b -> SiphonOp e r a -contramapFoldable unwinder = contramapEnumerable (Select . each . unwinder) - -contramapEnumerable :: Enumerable t => (a -> t IO b) -> SiphonOp e r b -> SiphonOp e r a -contramapEnumerable unwinder (getSiphonOp -> s) = SiphonOp $ - siphon' $ runSiphon s . flip for (enumerate . toListT . unwinder) - -contraproduce :: (forall r. Producer a IO r -> Producer b IO r) -> SiphonOp e r b -> SiphonOp e r a -contraproduce f (getSiphonOp -> s) = SiphonOp $ siphon' $ runSiphon s . f - -{-| - Specifies a transformation that will be applied to each line of text, - represented as a 'Producer'. --} -tweakLines :: (forall r. Producer T.Text IO r -> Producer T.Text IO r) -> Lines e -> Lines e -tweakLines lt' (Lines tear lt) = Lines tear (lt' . lt) - - -{-| - Specifies a prefix that will be calculated and appended for each line of - text. --} -prefixLines :: IO T.Text -> Lines e -> Lines e -prefixLines tio = tweakLines (\p -> liftIO tio *> p) - - -{-| - Constructs a 'Lines' value. - -} -toLines :: DecodingFunction ByteString Text - -- ^ A decoding function for lines of text. - -> Siphon ByteString e (() -> ()) - -- ^ A 'Siphon' that determines how to handle decoding leftovers. - -- Pass @pure id@ to ignore leftovers. Pass @unwanted id@ to abort - -- the computation with an explicit error if leftovers remain. Pass - -- '_leftoverX' to throw a 'LeftoverException' if leftovers remain. - -> Lines e -toLines decoder lopo = Lines - (\tweaker tear producer -> do - let freeLines = transFreeT tweaker - . viewLines - . decoder - $ producer - viewLines = getConst . T.lines Const - tear freeLines >>= runSiphonDumb (fmap ($()) lopo)) - id - - -{-| - A simplified version of 'executePipelineFallibly' for when the error type - unifies with `Void`. Note however that this function may still throw - exceptions. - -} -executePipeline :: Piping Void a -> Tree (Stage 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. - - 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 :: Piping e a - -- ^ - -- Views the pipeline as a single process - -- for which @stdin@ is the @stdin@ of the first stage and @stdout@ is the - -- @stdout@ of the leftmost terminal stage closer to the root. - -- @stderr@ is a combination of the @stderr@ streams of all the - -- stages. The combined @stderr@ stream always has UTF-8 encoding. - -> Tree (Stage e) - -- ^ A (possibly branching) pipeline of processes. - -- Each process' stdin is fed with the stdout of - -- its parent in the tree. - -> IO (Either e a) -executePipelineFallibly policy (Node (Stage cp lpol ecpol _) []) = case policy of - PPNone _ -> blende ecpol <$> executeFallibly policy cp - PPOutput _ -> blende ecpol <$> executeFallibly policy cp - PPError action -> do - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action $ fromInput einbox) - <* - (Conceit $ blende ecpol <$> executeFallibly (pipee (errf lpol)) cp `finally` atomically eseal) - PPOutputError action -> do - (outbox, inbox, seal) <- spawn' (bounded 1) - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action $ (fromInput inbox,fromInput einbox)) - <* - (Conceit $ blende ecpol <$> executeFallibly - (pipeoe (fromConsumer.toOutput $ outbox) (errf lpol)) cp - `finally` atomically seal `finally` atomically eseal - ) - PPInput _ -> blende ecpol <$> executeFallibly policy cp - PPInputOutput _ -> blende ecpol <$> executeFallibly policy cp - PPInputError action -> do - (outbox, inbox, seal) <- spawn' (bounded 1) - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox)) - <* - (Conceit $ blende ecpol <$> executeFallibly - (pipeie (fromProducer . fromInput $ inbox) (errf lpol)) cp - `finally` atomically seal `finally` atomically eseal - ) - PPInputOutputError action -> do - (ioutbox, iinbox, iseal) <- spawn' (bounded 1) - (ooutbox, oinbox, oseal) <- spawn' (bounded 1) - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox)) - <* - (Conceit $ blende ecpol <$> executeFallibly - (pipeioe (fromProducer . fromInput $ iinbox) - (fromConsumer . toOutput $ ooutbox) - (errf lpol) - ) - cp - `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal - ) -executePipelineFallibly policy (Node s (s':ss)) = - let pipeline = CreatePipeline s $ s' :| ss - in 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' (bounded 1) - 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' (bounded 1) - 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' (bounded 1) - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - 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' (bounded 1) - 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' (bounded 1) - (ooutbox, oinbox, oseal) <- spawn' (bounded 1) - 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' (bounded 1) - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - 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' (bounded 1) - (ooutbox, oinbox, oseal) <- spawn' (bounded 1) - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - 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 - ) - -errorSiphonUTF8 :: MVar (Output ByteString) -> Lines e -> Siphon ByteString e () -errorSiphonUTF8 mvar (Lines fun twk) = siphon (fun twk 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) - -mute :: Functor f => f a -> f () -mute = fmap (const ()) - - -{-| - Builds a 'Stage'. --} -stage :: Lines e - -- ^ How to handle lines coming from stderr for this 'Stage'. - -> (ExitCode -> Either e ()) - -- ^ Does the 'ExitCode' for this 'Stage' represent an error? (Some - -- programs return non-standard exit codes.) - -> CreateProcess - -- ^ A process definition. - -> Stage e -stage lp ec cp = Stage cp lp ec (hoist lift) - -{-| - Applies a transformation to the stream of bytes flowing into a stage from previous stages. - - This function is ignored for first stages. --} -inbound :: (forall r. Producer ByteString (ExceptT e IO) r -> Producer ByteString (ExceptT e IO) r) - -> Stage e -> Stage e -inbound f (Stage a b c d) = Stage a b c (f . d) - -data CreatePipeline e = CreatePipeline (Stage e) (NonEmpty (Tree (Stage e))) deriving (Functor) - -executePipelineInternal :: (Siphon ByteString e () -> Lines e -> Piping e ()) - -> (Pump ByteString e () -> Siphon ByteString e () -> Lines e -> Piping e ()) - -> (Pump ByteString e () -> Lines e -> Piping e ()) - -> (Pump ByteString e () -> Lines e -> Piping 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 (Stage _cp _lpol _ecpol pipe) forest) = case forest of - [] -> siphon $ \producer -> - blende _ecpol <$> executeFallibly (_ppend (fromFallibleProducer $ pipe producer) _lpol) _cp - c1 : cs -> siphon $ \producer -> - blende _ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ pipe producer) (runNonEmpty _ppend _ppend' (c1 :| cs)) _lpol) _cp - - runNonEmpty _ppend _ppend' (b :| bs) = - runTree _ppend _ppend' b <* Prelude.foldr (<*) (pure ()) (runTree _ppend' _ppend' <$> bs) - -blende :: (ExitCode -> Either e ()) -> Either e (ExitCode,a) -> Either e a -blende f r = r >>= \(ec,a) -> f ec *> pure a - -{-| - Converts any 'ExitFailure' to the left side of an 'Either'. --} -pipefail :: ExitCode -> Either Int () -pipefail ec = case ec of - ExitSuccess -> Right () - ExitFailure i -> Left i - - -{- $reexports - -"System.Process" is re-exported for convenience. +-- | +-- This module contains helper functions and types built on top of +-- "System.Process". +-- +-- They provide concurrent, streaming access to the inputs and outputs of +-- external processes. +-- +-- 'Consumer's from @pipes@, 'Parser's from @pipes-parse@ and 'Fold's from +-- @foldl@ can be used to consume the standard streams, by +-- means of the auxiliary 'Fold1' datatype which is re-exported from +-- @pipes-transduce@. +----------------------------------------------------------------------------- + +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE ViewPatterns #-} +{-# LANGUAGE FlexibleInstances #-} + + +module System.Process.Streaming ( + -- * Execution + execute + , executeFallibly + -- * CreateProcess helpers + , piped + -- * The Streams Applicative + , Streams + -- * Feeding stdin + , feedBytes + , feedLazyBytes + , feedUtf8 + , feedLazyUtf8 + , feedProducer + , feedProducerM + , feedSafeProducer + , feedFallibleProducer + , feedCont + -- * Consuming stdout and stderr + -- $folds + , foldOut + , foldErr + , Pipes.Transduce.ByteString.intoLazyBytes + , foldOutErr + -- * Handling exit codes + , exitCode + , validateExitCode + , withExitCode + -- * A GHCi idiom + -- $ghci + -- * Re-exports + -- $reexports + , module System.Process + , module Pipes.Transduce + ) where + +import qualified Data.ByteString.Lazy +import Data.Monoid +import Data.Foldable +import Data.Bifunctor +import Data.ByteString +import Data.Text +import qualified Data.Text.Encoding +import qualified Data.Text.Lazy +import qualified Data.Text.Lazy.Encoding +import Data.Void +import Data.Functor.Day +import Data.Profunctor (Star(..)) +import Control.Applicative +import Control.Applicative.Lift +import Control.Monad +import Control.Monad.Trans.Except +import Control.Exception (onException,catch,IOException,mask,finally) +import Control.Concurrent +import Control.Concurrent.Conceit +import Pipes +import qualified Pipes.Prelude +import Pipes.ByteString (fromHandle,toHandle,fromLazy) +import Pipes.Concurrent +import Pipes.Safe (SafeT,runSafeT) +import Pipes.Transduce +import Pipes.Transduce.ByteString +import System.IO +import System.Process +import System.Exit + +{- $setup + +>>> :set -XOverloadedStrings +>>> import Control.Exception (throwIO) +>>> import qualified System.Process.Streaming.Text as PT + +-} + +{-| + Execute an external program described by the 'CreateProcess' record. + + The 'Streams' Applicative specifies how to handle the standard + streams and the exit code. Since 'Streams' is an Applicative, a simple + invocation of 'execute' could be + +>>> execute (piped (shell "echo foo")) (pure ()) + + which would discard the program's stdout and stderr, and ignore the exit + code. To actually get the exit code: + +>>> execute (piped (shell "echo foo")) exitCode +ExitSuccess + + To collect stdout as a lazy 'Data.ByteString.Lazy.ByteString' along with the + exit code: + +>>> execute (piped (shell "echo foo")) (liftA2 (,) (foldOut intoLazyBytes) exitCode) +("foo\n",ExitSuccess) + + 'execute' respects all the fields of the 'CreateProcess' record. If stdout is + not piped, but a handler is defined for it, the handler will see an empty + stream: + +>>> execute ((shell "echo foo"){ std_out = Inherit }) (foldOut intoLazyBytes) +foo +"" + + No effort is made to catch exceptions thrown during execution: + +>>> execute (piped (shell "echo foo")) (foldOut (withCont (\_ -> throwIO (userError "oops")))) +*** Exception: user error (oops) + + However, care is taken to automatically terminate the external process if an + exception (including asynchronous ones) or other type of error happens. + This means we can terminate the external process by killing + the thread that is running 'execute': + +>>> forkIO (execute (piped (shell "sleep infinity")) (pure ())) >>= killThread + + -} +execute :: CreateProcess -> Streams Void a -> IO a +execute cprocess pp = either absurd id <$> executeFallibly cprocess pp + + +{-| Like 'execute', but allows the handlers in the 'Streams' Applicative to interrupt the execution of the external process by returning a 'Left' value, in addition to throwing exceptions. This is sometimes more convenient: + +>>> executeFallibly (piped (shell "sleep infinity")) (foldOut (withFallibleCont (\_ -> pure (Left "oops")))) +Left "oops" + +>>> executeFallibly (piped (shell "exit 1")) validateExitCode +Left 1 + + The first type parameter of 'Streams' is the error type. If it is never used, it remains polymorphic and may unify with 'Void' (as required by 'execute'). + +-} +executeFallibly :: + CreateProcess + -> Streams e a + -> IO (Either e a) +executeFallibly record (Streams streams) = mask $ \restore -> do + (mstdin,mstdout,mstderr,phandle) <- createProcess record + let (clientx,cleanupx) = case mstdin of + Nothing -> (pure (),pure ()) + Just handle -> (toHandle handle,hClose handle) + (producer1x,cleanup1) = case mstdout of + Nothing -> (pure (),pure()) + Just handle -> (fromHandle handle,hClose handle) + (producer2x,cleanup2) = case mstderr of + Nothing -> (pure (),pure ()) + Just handle -> (fromHandle handle,hClose handle) + streams' = + dap + . trans1 + ( + flip catchE (\e -> liftIO (terminateCarefully phandle) *> throwE e) + . ExceptT + . runConceit + . dap + . trans1 (\f -> Conceit (feed1Fallibly f clientx `finally` cleanupx)) + . trans2 (\f -> Conceit (fmap (fmap (\(x,_,_) -> x)) (Pipes.Transduce.fold2Fallibly f producer1x producer2x))) + ) + . trans2 (\exitCodeHandler -> do + c <- liftIO (waitForProcess phandle) + runStar exitCodeHandler c) + $ streams + -- The following is a workaround for the fact that, in Windows, IO actions + -- are not interruptible: https://ghc.haskell.org/trac/ghc/ticket/7353 + -- + -- This is annoying because we want to kill the external process in case + -- of an asynchronous exception. + -- + -- What we do is create a thread with the sole purpose of being a "soft + -- target" for the asychronous exception. That thread installs an exception + -- handler that kills the external process. + -- + -- We must be sure that the handler is installed before anything is read + -- form a standard stream, that's why we use the a MVar. + latch <- newEmptyMVar + let killable = _runConceit $ + (_Conceit (takeMVar latch >> runExceptT streams')) + <|> + (_Conceit (onException (putMVar latch () >> _runConceit Control.Applicative.empty) + (terminateCarefully phandle))) + -- 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 killable `onException` terminateCarefully phandle) `finally` cleanup1 `finally` cleanup2 + +terminateCarefully :: ProcessHandle -> IO () +terminateCarefully pHandle = do + mExitCode <- getProcessExitCode pHandle + case mExitCode of + Nothing -> catch + (terminateProcess pHandle) + (\(_::IOException) -> return ()) + Just _ -> return () + +{-| Sets 'std_in', 'std_out' and 'std_err' in the 'CreateProcess' record to + 'CreatePipe'. + + Any unpiped stream will appear to the 'Streams' handlers as empty. + +-} +piped :: CreateProcess -> CreateProcess +piped cmd = cmd { std_in = CreatePipe, + std_out = CreatePipe, + std_err = CreatePipe } + +newtype Feed1 b e a = Feed1 (Lift (Feed1_ b e) a) deriving (Functor) + +newtype Feed1_ b e a = Feed1_ { runFeed1_ :: Consumer b IO () -> IO (Either e a) } deriving Functor + +instance Bifunctor (Feed1_ b) where + bimap f g (Feed1_ x) = Feed1_ $ fmap (liftM (bimap f g)) x + +{-| + 'first' is useful to massage errors. +-} +instance Bifunctor (Feed1 b) where + bimap f g (Feed1 x) = Feed1 (case x of + Pure a -> Pure (g a) + Other o -> Other (bimap f g o)) + +instance Applicative (Feed1 b e) where + pure a = Feed1 (pure a) + Feed1 fa <*> Feed1 a = Feed1 (fa <*> a) + +{-| + 'pure' writes nothing to @stdin@. + '<*>' sequences the writes to @stdin@. +-} +instance Applicative (Feed1_ b e) where + pure = Feed1_ . pure . pure . pure + Feed1_ fs <*> Feed1_ as = + Feed1_ $ \consumer -> do + (outbox1,inbox1,seal1) <- spawn' (bounded 1) + (outbox2,inbox2,seal2) <- spawn' (bounded 1) + 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 + runExceptT $ pure () + ) + +instance (Monoid a) => Monoid (Feed1 b e a) where + mempty = pure mempty + mappend s1 s2 = (<>) <$> s1 <*> s2 + +feed1Fallibly :: Feed1 b e a -> Consumer b IO () -> IO (Either e a) +feed1Fallibly (Feed1 (unLift -> s)) = runFeed1_ s + +{-| Feed any 'Foldable' container of strict 'Data.ByteString's to @stdin@. + +-} +feedBytes :: Foldable f => f ByteString -> Streams e () +feedBytes = feedProducer . each + +{-| Feed a lazy 'Data.Lazy.ByteString' to @stdin@. + +-} +feedLazyBytes :: Data.ByteString.Lazy.ByteString -> Streams e () +feedLazyBytes = feedProducer . fromLazy + +{-| Feed any 'Foldable' container of strict 'Data.Texts's to @stdin@, encoding + the texts as UTF8. + +-} +feedUtf8 :: Foldable f => f Text -> Streams e () +feedUtf8 = feedProducer . (\p -> p >-> Pipes.Prelude.map Data.Text.Encoding.encodeUtf8) . each + +{-| Feed a lazy 'Data.Lazy.Text' to @stdin@, encoding it as UTF8. + +-} +feedLazyUtf8 :: Data.Text.Lazy.Text -> Streams e () +feedLazyUtf8 = feedProducer . fromLazy . Data.Text.Lazy.Encoding.encodeUtf8 + +feedProducer :: Producer ByteString IO () -> Streams e () +feedProducer producer = liftFeed1 . Feed1 . Other . Feed1_ $ \consumer -> fmap pure $ runEffect (void producer >-> consumer) +feedProducerM :: MonadIO m => (m () -> IO (Either e a)) -> Producer ByteString m r -> Streams e a +feedProducerM whittle producer = liftFeed1 . Feed1 . Other . Feed1_ $ \consumer -> whittle $ runEffect (void producer >-> hoist liftIO consumer) + +feedSafeProducer :: Producer ByteString (SafeT IO) () -> Streams e () +feedSafeProducer = feedProducerM (fmap pure . runSafeT) + +feedFallibleProducer :: Producer ByteString (ExceptT e IO) () -> Streams e () +feedFallibleProducer = feedProducerM runExceptT + + +{-| Feed @stdin@ by running a pipes 'Consumer'. This allows bracketing + functions like 'withFile' inside the handler. + +-} +feedCont :: (Consumer ByteString IO () -> IO (Either e a)) -> Streams e a +feedCont = liftFeed1 . Feed1 . Other . Feed1_ + +{- $folds + + These functions take as parameters the 'Pipes.Transduce.Fold1' and + 'Pipes.Transduce.Fold2' datatypes defined in the @pipes-transduce@ package. + + A convenience 'intoLazyBytes' 'Pipes.Transduce.Fold1' that collects a stream into a + lazy 'Data.ByteString.Lazy.ByteString' is re-exported. +-} + +{-| Consume standard output. + +-} +foldOut :: Fold1 ByteString e r -> Streams e r +foldOut = liftFold2 . Pipes.Transduce.liftFirst + +{-| Consume standard error. + +-} +foldErr :: Fold1 ByteString e r -> Streams e r +foldErr = liftFold2 . Pipes.Transduce.liftSecond + +{-| Consume standard output and error together. See also the 'combine' function + re-exported from "Pipes.Transduce". + +-} +foldOutErr :: Fold2 ByteString ByteString e r -> Streams e r +foldOutErr = liftFold2 + +{-| Simply returns the 'ExitCode'. + +-} +exitCode :: Streams e ExitCode +exitCode = liftExitCodeValidation (Star pure) + + +{-| Fails with the error code when 'ExitCode' is not 'ExitSuccess'. + +-} +validateExitCode :: Streams Int () +validateExitCode = liftExitCodeValidation . Star . fmap ExceptT . fmap pure $ validation + where + validation ExitSuccess = Right () + validation (ExitFailure i) = Left i + +withExitCode :: (ExitCode -> IO (Either e a)) -> Streams e a +withExitCode = liftExitCodeValidation . Star . fmap ExceptT + +liftFeed1 :: Feed1 ByteString e a -> Streams e a +liftFeed1 f = Streams $ + day + (day (const . const <$> f) + (pure ())) + (pure ()) + +liftFold2 :: Fold2 ByteString ByteString e a -> Streams e a +liftFold2 f2 = Streams $ + day + (day (pure const) + f2) + (pure ()) + +liftExitCodeValidation :: (Star (ExceptT e IO) ExitCode a) -> Streams e a +liftExitCodeValidation v = Streams $ + swapped + (day + (const <$> v) + (pure ())) + +{-| The type of handlers that write to piped @stdin@, consume piped @stdout@ and + @stderr@, and work with the process exit code, eventually returning a value of + type @a@, except when an error @e@ interrups the execution. + + Example of a complex handler: + +>>> :{ + execute (piped (shell "{ cat ; echo eee 1>&2 ; }")) $ + (\_ _ o e oe ec -> (o,e,oe,ec)) + <$> + feedBytes (Just "aaa") + <*> + feedBytes (Just "bbb") + <*> + foldOut intoLazyBytes + <*> + foldErr intoLazyBytes + <*> + foldOutErr (combined (PT.lines PT.utf8x) (PT.lines PT.utf8x) PT.intoLazyText) + <*> + exitCode + :} +("aaabbb","eee\n","aaabbb\neee\n",ExitSuccess) + +-} +newtype Streams e r = + Streams (Day (Day (Feed1 ByteString e) + (Fold2 ByteString ByteString e)) + (Star (ExceptT e IO) ExitCode) + r) + deriving (Functor) + +{-| 'first' is useful to massage errors. + +-} +instance Bifunctor Streams where + bimap f g (Streams d) = Streams $ fmap g $ + trans1 + ( trans1 (first f) + . trans2 (first f) + ) + . trans2 (\(Star starfunc) -> Star (withExceptT f <$> starfunc)) + $ d + + +{-| + 'pure' writes nothing to @stdin@, discards the data coming from @stdout@ and @stderr@, and ignores the exit code. + + '<*>' combines handlers by sequencing the writes to @stdin@, and making concurrent reads from @stdout@ and @stderr@. +-} +instance Applicative (Streams e) where + pure a = Streams (pure a) + + Streams f <*> Streams x = Streams (f <*> x) + +instance (Monoid a) => Monoid (Streams e a) where + mempty = pure mempty + mappend s1 s2 = (<>) <$> s1 <*> s2 + +{- $ghci + +Within GHCi, it's easy to use this module to launch external programs. + +If the program is long-running, do it in a new thread to avoid locking GHCi, +and keep track of the thread id: + +@ +ghci> r <- forkIO $ execute (piped (proc "C:\/Program Files (x86)\/Vim\/vim74\/gvim.exe" [])) (pure ()) +@ + +Kill the thread to kill the long-running process: + +@ +ghci> killThread r +@ + +-} + +{- $reexports + +"System.Process" is re-exported in its entirety. + +"Pipes.Transduce" from the @pipes-transduce@ package is re-exported in its entirety. -} diff --git a/src/System/Process/Streaming/Extended.hs b/src/System/Process/Streaming/Extended.hs deleted file mode 100644 --- a/src/System/Process/Streaming/Extended.hs +++ /dev/null @@ -1,154 +0,0 @@ -{-| --} - -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE ViewPatterns #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} - -module System.Process.Streaming.Extended ( - Piap - , piapi - , piapo - , piape - , piapoe - , samei - , sameo - , samee - , sameioe - , toPiping - , pumpFromHandle - , siphonToHandle - , module System.Process.Streaming - ) where - -import Data.Text -import Control.Applicative -import Control.Exception -import Control.Concurrent.Conceit -import Pipes.ByteString -import System.IO - -import System.Process.Streaming -import System.Process.Streaming.Internal - -toPiping :: Piap e a -> Piping e a -toPiping (Piap f) = PPInputOutputError f - -{-| - Do stuff with @stdout@. --} -piapo :: Siphon ByteString e a -> Piap e a -piapo s = Piap $ \(consumer, cleanup, producer1, producer2) -> do - let nullInput = runPump (pure ()) consumer `finally` cleanup - drainOutput = runSiphonDumb s producer1 - drainError = runSiphonDumb (pure ()) producer2 - runConceit $ - (\_ r _ -> r) - <$> - Conceit nullInput - <*> - Conceit drainOutput - <*> - Conceit drainError - -{-| - Do stuff with @stderr@. --} -piape :: Siphon ByteString e a -> Piap e a -piape s = Piap $ \(consumer, cleanup, producer1, producer2) -> do - let nullInput = runPump (pure ()) consumer `finally` cleanup - drainOutput = runSiphonDumb (pure ()) producer1 - drainError = runSiphonDumb s producer2 - runConceit $ - (\_ _ r -> r) - <$> - Conceit nullInput - <*> - Conceit drainOutput - <*> - Conceit drainError - -{-| - Do stuff with @stdin@. --} -piapi :: Pump ByteString e a -> Piap e a -piapi p = Piap $ \(consumer, cleanup, producer1, producer2) -> do - let nullInput = runPump p consumer `finally` cleanup - drainOutput = runSiphonDumb (pure ()) producer1 - drainError = runSiphonDumb (pure ()) producer2 - runConceit $ - (\r _ _ -> r) - <$> - Conceit nullInput - <*> - Conceit drainOutput - <*> - Conceit drainError - -{-| - Do stuff with @stdout@ and @stderr@ combined as 'Text'. --} -piapoe :: Lines e -> Lines e -> Siphon Text e a -> Piap e a -piapoe policy1 policy2 s = Piap $ \(consumer, cleanup, producer1, producer2) -> do - let nullInput = runPump (pure ()) consumer `finally` cleanup - combination = combined policy1 policy2 (runSiphonDumb s) producer1 producer2 - runConceit $ - (\_ r -> r) - <$> - Conceit nullInput - <*> - Conceit combination - -{-| - Pipe @stdin@ to the created process' @stdin@. --} -samei :: Piap e () -samei = piapi $ pumpFromHandle System.IO.stdin - -{-| - Pipe the created process' @stdout@ to @stdout@. --} -sameo :: Piap e () -sameo = piapo $ siphonToHandle System.IO.stdout - -{-| - Pipe the created process' @stderr@ to @stderr@. --} -samee :: Piap e () -samee = piape $ siphonToHandle System.IO.stderr - -sameioe :: Piap e () -sameioe = samei *> sameo *> samee - -pumpFromHandle :: Handle -> Pump ByteString e () -pumpFromHandle = fromProducer . fromHandle - -siphonToHandle :: Handle -> Siphon ByteString e () -siphonToHandle = fromConsumer . toHandle - - ---{-| --- More general than '_nestEnumerable' in that the 'Siphon's that consume each --- stream of @b@s can depend on the @a@s. ----} ---nestEnumerable :: Enumerable t => (a -> t IO b) -> (a -> Client (SiphonOp e () b) a (ExceptT e IO) Void) -> SiphonOp e () a ---nestEnumerable unwinder siphonClient = SiphonOp $ siphon' $ \producer -> runExceptT . runEffect $ fmap ((,) ()) $ --- hoist lift producer >>~ (retag >~> (fmap absurd . siphonClient)) --- where --- retag a = do --- s <- respond a --- _ <- lift . ExceptT $ runSiphonDumb (getSiphonOp s) (enumerate . toListT . unwinder $ a) --- request () >>= retag --- --- ---{-| --- For each incoming @a@, use a different 'Siphon' to consume the --- corresponding stream of @b@s. ----} ---_nestEnumerable :: Enumerable t => (a -> t IO b) -> Producer (SiphonOp e () b) (ExceptT e IO) Void -> SiphonOp e () a ---_nestEnumerable unwinder siphonProducer = SiphonOp $ siphon' $ \producer -> runExceptT . runEffect $ fmap ((,) ()) $ --- for (P.zip (hoist lift producer) (fmap absurd siphonProducer)) $ \(a, siph) -> --- lift . ExceptT $ runSiphonDumb (getSiphonOp siph) (enumerate . toListT . unwinder $ a) diff --git a/src/System/Process/Streaming/Internal.hs b/src/System/Process/Streaming/Internal.hs deleted file mode 100644 --- a/src/System/Process/Streaming/Internal.hs +++ /dev/null @@ -1,408 +0,0 @@ -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE ViewPatterns #-} - -module System.Process.Streaming.Internal ( - Piping(..), - Piap(..), - Pump(..), - Siphon(..), - runSiphon, - runSiphonDumb, - Siphon_(..), - exhaustive, - Lines(..), - Splitter(..), - combined, - manyCombined, - Stage(..) - ) where - -import Data.Bifunctor -import Data.Monoid -import Data.Text -import Control.Applicative -import Control.Applicative.Lift -import Control.Monad -import Control.Monad.Trans.Free hiding (Pure) -import Control.Monad.Trans.Except -import Control.Exception -import Control.Concurrent -import Control.Concurrent.Conceit -import Pipes -import qualified Pipes as P -import qualified Pipes.Prelude as P -import Pipes.ByteString -import qualified Pipes.Text as T -import Pipes.Concurrent -import System.Process -import System.Exit - -{-| - A 'Piping' determines what standard streams will be piped and what to -do with them. - - 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. - - A 'Piping' 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 Piping 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) - - -{-| - 'first' is useful to massage errors. --} -instance Bifunctor Piping where - 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 - - -{-| - 'Pump's are actions that write data into a process' @stdin@. - -} -newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor - -{-| - 'first' is useful to massage errors. --} -instance Bifunctor (Pump b) where - bimap f g (Pump x) = Pump $ fmap (liftM (bimap f g)) x - - -{-| - 'pure' writes nothing to @stdin@. - - '<*>' sequences the writes to @stdin@. --} -instance Applicative (Pump b e) where - pure = Pump . pure . pure . pure - Pump fs <*> Pump as = - Pump $ \consumer -> do - (outbox1,inbox1,seal1) <- spawn' (bounded 1) - (outbox2,inbox2,seal2) <- spawn' (bounded 1) - 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 - runExceptT $ pure () - ) - -instance (Monoid a) => Monoid (Pump b e a) where - mempty = Pump . pure . pure . pure $ mempty - mappend s1 s2 = (<>) <$> s1 <*> s2 - --- instance IsString b => IsString (Pump b e ()) where --- fromString = fromProducer . P.yield . fromString - -{-| - An alternative to `Piping` for defining what to do with the - standard streams. 'Piap' is an instance of 'Applicative', unlike - 'Piping'. - - With 'Piap', the standard streams are always piped. The values of - @std_in@, @std_out@ and @std_err@ in the 'CreateProcess' record are - ignored. - -} -newtype Piap e a = Piap { runPiap :: (Consumer ByteString IO (), IO (), Producer ByteString IO (), Producer ByteString IO ()) -> IO (Either e a) } deriving (Functor) - -instance Bifunctor Piap where - bimap f g (Piap action) = Piap $ fmap (fmap (bimap f g)) action - - -{-| - 'pure' creates a 'Piap' that writes nothing to @stdin@ and drains - @stdout@ and @stderr@, discarding the data. - - '<*>' schedules the writes to @stdin@ sequentially, and the reads from - @stdout@ and @stderr@ concurrently. --} -instance Applicative (Piap e) where - pure a = Piap $ \(consumer, cleanup, producer1, producer2) -> do - let nullInput = runPump (pure ()) consumer `finally` cleanup - drainOutput = runSiphonDumb (pure ()) producer1 - drainError = runSiphonDumb (pure ()) producer2 - runConceit $ - (\_ _ _ -> a) - <$> - Conceit nullInput - <*> - Conceit drainOutput - <*> - Conceit drainError - - (Piap fa) <*> (Piap fb) = Piap $ \(consumer, cleanup, producer1, producer2) -> do - latch <- newEmptyMVar :: IO (MVar ()) - (ioutbox, iinbox, iseal) <- spawn' (bounded 1) - (ooutbox, oinbox, oseal) <- spawn' (bounded 1) - (eoutbox, einbox, eseal) <- spawn' (bounded 1) - (ioutbox2, iinbox2, iseal2) <- spawn' (bounded 1) - (ooutbox2, oinbox2, oseal2) <- spawn' (bounded 1) - (eoutbox2, einbox2, eseal2) <- spawn' (bounded 1) - let reroutei = runEffect (fromInput (iinbox <> iinbox2) >-> consumer) - `finally` atomically iseal - `finally` atomically iseal2 - `finally` cleanup - rerouteo = runEffect (producer1 >-> toOutput (ooutbox <> ooutbox2)) - `finally` atomically oseal - `finally` atomically oseal2 - reroutee = runEffect (producer2 >-> toOutput (eoutbox <> eoutbox2)) - `finally` atomically eseal - `finally` atomically eseal2 - deceivedf = fa - (toOutput ioutbox, - atomically iseal `finally` putMVar latch (), - fromInput oinbox, - fromInput einbox) - `finally` atomically iseal - `finally` atomically oseal - `finally` atomically eseal - deceivedx = fb - (liftIO (takeMVar latch) *> toOutput ioutbox2, - atomically iseal2, - fromInput oinbox2, - fromInput einbox2) - `finally` atomically iseal2 - `finally` atomically oseal2 - `finally` atomically eseal2 - runConceit $ - _Conceit reroutei - *> - _Conceit rerouteo - *> - _Conceit reroutee - *> - (Conceit deceivedf <*> Conceit deceivedx) - - -{-| - A 'Siphon' represents a computation that completely drains - a 'Producer', but which may fail early with an error of type @e@. - -} -newtype Siphon b e a = Siphon (Lift (Siphon_ b e) a) deriving (Functor) - - -{-| - 'pure' creates a 'Siphon' that does nothing besides draining the -'Producer'. - - '<*>' executes its arguments concurrently. The 'Producer' is forked so - that each argument receives its own copy of the data. --} -instance Applicative (Siphon b e) where - pure a = Siphon (pure a) - (Siphon fa) <*> (Siphon a) = Siphon (fa <*> a) - -data Siphon_ b e a = - Exhaustive (forall r. Producer b IO r -> IO (Either e (a,r))) - | Nonexhaustive (Producer b IO () -> IO (Either e a)) - deriving (Functor) - - -instance Applicative (Siphon_ b e) where - pure a = Exhaustive $ \producer -> do - r <- runEffect (producer >-> P.drain) - pure (pure (a,r)) - s1 <*> s2 = bifurcate (nonexhaustive s1) (nonexhaustive s2) - where - bifurcate fs as = Exhaustive $ \producer -> do - (outbox1,inbox1,seal1) <- spawn' (bounded 1) - (outbox2,inbox2,seal2) <- spawn' (bounded 1) - runConceit $ - (,) - <$> - Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) - `finally` atomically seal1) - ((as $ fromInput inbox2) - `finally` atomically seal2) - ) - <*> - _Conceit ((runEffect $ - producer >-> P.tee (toOutput outbox1 >> P.drain) - >-> (toOutput outbox2 >> P.drain)) - `finally` atomically seal1 `finally` atomically seal2 - ) - -nonexhaustive :: Siphon_ b e a -> Producer b IO () -> IO (Either e a) -nonexhaustive (Exhaustive e) = \producer -> liftM (fmap fst) (e producer) -nonexhaustive (Nonexhaustive u) = u - -exhaustive :: Siphon_ b e a -> Producer b IO r -> IO (Either e (a,r)) -exhaustive s = case s of - Exhaustive e -> e - Nonexhaustive activity -> \producer -> do - (outbox,inbox,seal) <- spawn' (bounded 1) - runConceit $ - (,) - <$> - Conceit (activity (fromInput inbox) `finally` atomically seal) - <*> - _Conceit (runEffect (producer >-> (toOutput outbox >> P.drain)) - `finally` atomically seal - ) - -runSiphon :: Siphon b e a -> Producer b IO r -> IO (Either e (a,r)) -runSiphon (Siphon (unLift -> s)) = exhaustive s - -runSiphonDumb :: Siphon b e a -> Producer b IO () -> IO (Either e a) -runSiphonDumb (Siphon (unLift -> s)) = nonexhaustive $ case s of - Exhaustive _ -> s - Nonexhaustive _ -> Exhaustive (exhaustive s) - - -instance Bifunctor (Siphon_ b) where - bimap f g s = case s of - Exhaustive u -> Exhaustive $ fmap (liftM (bimap f (bimap g id))) u - Nonexhaustive h -> Nonexhaustive $ fmap (liftM (bimap f g)) h - -{-| - 'first' is useful to massage errors. --} -instance Bifunctor (Siphon b) where - bimap f g (Siphon s) = Siphon $ case s of - Pure a -> Pure (g a) - Other o -> Other (bimap f g o) - -instance (Monoid a) => Monoid (Siphon b e a) where - mempty = pure mempty - mappend s1 s2 = (<>) <$> s1 <*> s2 - - - --- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here -combined :: Lines e - -> Lines e - -> (Producer T.Text IO () -> IO (Either e a)) - -> Producer ByteString IO () - -> Producer ByteString IO () - -> IO (Either e a) -combined (Lines fun1 twk1) (Lines fun2 twk2) combinedConsumer prod1 prod2 = - manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer - -manyCombined :: [(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' (bounded 1) - 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) - -{-| - A configuration parameter used in functions that combine lines of text from - multiple streams. - -} - -data Lines e = Lines - { - teardown :: (forall r. Producer T.Text IO r -> Producer T.Text IO r) - -> (FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) - -> Producer ByteString IO () -> IO (Either e ()) - , lineTweaker :: forall r. Producer T.Text IO r -> Producer T.Text IO r - } - --- | 'fmap' maps over the encoding error. -instance Functor Lines where - fmap f (Lines func lt) = Lines (\x y z -> fmap (bimap f id) $ func x y z) lt - - -newtype Splitter b = Splitter { getSplitter :: forall r. Producer b IO r -> FreeT (Producer b IO) IO r } - - -{-| - An individual stage in a process pipeline. - -} -data Stage e = Stage - { - _processDefinition :: CreateProcess - , _stderrLines :: Lines e - , _exitCodePolicy :: ExitCode -> Either e () - , _inbound :: forall r. Producer ByteString IO r - -> Producer ByteString (ExceptT e IO) r - } - -instance Functor (Stage) where - fmap f (Stage a b c d) = Stage a (fmap f b) (bimap f id . c) (hoist (mapExceptT $ liftM (bimap f id)) . d) - - - - diff --git a/src/System/Process/Streaming/Text.hs b/src/System/Process/Streaming/Text.hs new file mode 100644 --- /dev/null +++ b/src/System/Process/Streaming/Text.hs @@ -0,0 +1,99 @@ +{-| + This module re-exports functions useful for treating the @stdout@ and @stderr@ + streams as text. + + It is better to import it qualified: + +>>> import qualified System.Process.Streaming.Text as PT + +-} +module System.Process.Streaming.Text ( + -- * Examples + -- $examples + -- * Re-exports + -- $reexports + module Pipes.Transduce.Text + ) where + +import Pipes.Transduce.Text + +{- $setup + +>>> :set -XOverloadedStrings +>>> import qualified Data.Text.Lazy +>>> import qualified Pipes +>>> import Pipes.Transduce +>>> import Control.Applicative +>>> import System.Process.Streaming +>>> import qualified System.Process.Streaming.Text as PT +>>> import qualified Control.Foldl as L + +-} + +{- $examples + + To process a standard stream as text, use the 'utf8' or 'utf8x' + 'Transducer's, applying them with 'transduce1' to a 'Fold1' that accepts + text. + + The result will be a a 'Fold1' that accepts raw 'ByteStrings': + +>>> execute (piped (shell "echo foo")) (foldOut (transduce1 PT.utf8x PT.intoLazyText)) +"foo\n" + + The difference between 'utf8' and 'utf8x' is that the former uses an 'Either' + to signal encoding errors, while the latter uses an exception. + +>>> executeFallibly (piped (shell "echo foo")) (foldOut (transduce1 PT.utf8 PT.intoLazyText)) +Right "foo\n" + + 'foldedLines' lets you consume the lines that appear in a stream as lazy + 'Data.Text.Lazy.Text's. Here we collect them in a list using the + 'Control.Foldl.list' 'Control.Foldl.Fold' from the @foldl@ package: + +>>> :{ + execute (piped (shell "{ echo foo ; echo bar ; }")) + . foldOut + . transduce1 PT.utf8x + . transduce1 PT.foldedLines + $ withFold L.list + :} +["foo","bar"] + + Sometimes we want to consume the lines in @stdout@ and @stderr@ as a single + stream. We can do this with 'System.Process.Streaming.foldOutErr' and the 'Pipes.Transduce.combined' function. + + 'combined' takes one 'Pipes.Transduce.Transducer' for @stdout@ and another + for @stderr@, that known how to decode each stream into text, then break + the text into lines. + + The resulting lines are consumed using a 'Fold1': + +>>> :{ + execute (piped (shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }")) + . foldOutErr + . combined (PT.lines PT.utf8x) (PT.lines PT.utf8x) + $ intoLazyText + :} +"ooo\neee\n" + + We can also tag each line with its provenance, using 'Pipes.Transduce.groups': + +>>> :{ + let tag prefix = groups (\producer -> Pipes.yield prefix *> producer) + in + execute (piped (shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }")) + . foldOutErr + . combined (tag "+" (PT.lines PT.utf8x)) (tag "-" (PT.lines PT.utf8x)) + $ intoLazyText + :} +"+ooo\n-eee\n" + +-} + +{- $reexports + +"Pipes.Transduce.Text" from the @pipes-transduce@ package is re-exported in its entirety. + +-} + diff --git a/src/System/Process/Streaming/Tutorial.hs b/src/System/Process/Streaming/Tutorial.hs deleted file mode 100644 --- a/src/System/Process/Streaming/Tutorial.hs +++ /dev/null @@ -1,238 +0,0 @@ - -{-| - @process-streaming@ uses the 'CreateProcess' record to describe the - program to be executed. The user doesn't need to set the 'std_in', - 'std_out' and 'std_err' fields, as these are set automatically according - to the 'Piping'. - - 'Piping' is a datatype that specifies what standard streams to pipe and - what to do with them. It has several constructors, one for each possible - combination of streams. - - Constructors for 'Piping' usually take 'Siphon's as parameters. - A 'Siphon' specifies what to do with a particular standard stream. - - 'Siphon's can be built from each of the typical ways of consuming - a 'Producer' in the @pipes@ ecosystem: - - * Regular 'Consumer's (with 'fromConsumer', 'fromConsumerM'). - - * Folds from the @pipes@ 'Pipes.Prelude' or specialized folds from - @pipes-bytestring@ or @pipes-text@ (with 'fromFold', 'fromFold''). - - * 'Parser's from @pipes-parse@ (with 'fromParser' and 'fromParserM'). - - * 'Applicative' folds from the @foldl@ package (with 'fromFoldl', - 'fromFoldlIO' and 'fromFoldlM'). - - * In general, any computation that does something with a 'Producer' (with - 'siphon' and 'siphon''). - - 'Siphon's have an explicit error type; when a 'Siphon' reading one of the - standard streams fails, the external program is immediately terminated and - the error value is returned. - - A 'Siphon' reading a stream always consumes the whole stream. If the user - wants to interrupt the computation early, he can return a failure (or - throw an exception). - - 'Siphon's have an 'Applicative' instance. 'pure' creates a 'Siphon' that - drains a stream but does nothing with the data. --} - -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE ViewPatterns #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} - -module System.Process.Streaming.Tutorial ( - -- * Collecting @stdout@ as a lazy ByteString - -- $collstdout - - -- * Collecting @stdout@ and @stderr@ independently - -- $collstdoutstderr - - -- * Collecting @stdout@ as a lazy Text - -- $collstdouttext - - -- * Consuming @stdout@ and @stderr@ combined as Text - -- $collstdoutstderrtext - - -- * Feeding @stdin@, consuming @stdout@ - -- $feedstdincollstdout - - -- * Early termination - -- $earlytermination - - -- * Collecting @stdout@ as lines of Text - -- $collstdouttextlines - ) where - -import System.Process.Streaming - -{- $setup - ->>> :set -XOverloadedStrings ->>> import Data.Bifunctor ->>> import Data.Monoid ->>> import Data.ByteString.Lazy as BL ->>> 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.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 - --} - -{- $collstdout - -We can do this with the 'toLazyM' fold from @pipes-bytestring@: - ->>> execute (pipeo (fromFold B.toLazyM)) (shell "echo ooo") -(ExitSuccess,"ooo\n") - -But the 'intoLazyBytes' 'Siphon' is easier to use: - ->>> execute (pipeo intoLazyBytes) (shell "echo ooo") -(ExitSuccess,"ooo\n") - -'Siphon's are functors, so if we wanted to collect the output as a strict -'ByteString', we could do - ->>> execute (pipeo (BL.toStrict <$> intoLazyBytes)) (shell "echo ooo") -(ExitSuccess,"ooo\n") - -Of course, collecting the output in this way breaks streaming. But this is OK -if the output is small. --} - - - -{- $collstdoutstderr - -We can use 'pipeoe' collect @stdout@ and @stderr@ concurrently: - ->>> execute (pipeoe intoLazyBytes intoLazyBytes) (shell "{ echo ooo ; echo eee 1>&2 ; }") -(ExitSuccess,("ooo\n","eee\n")) - --} - - -{- $collstdouttext - -If we want to consume @stdout@ as text, we need to use the 'encoded' -function. 'encoded' takes as parameters a decoding function (here -'decodeUtf8') and a 'Siphon' that specifies how to handle the leftovers. It -returns a function that converts a 'Siphon' for text into a 'Siphon' for -bytes. - -In the example we pass @pure id@ as the leftover-handling 'Siphon'. This -means "drain all the undecoded data remaining in the stream and return -unchanged the result of 'intoLazyText'". In other words: ignore any -leftovers. - ->>> execute (pipeo (encoded decodeUtf8 (pure id) intoLazyText)) (shell "echo ooo") -(ExitSuccess,"ooo\n") - -But suppose we want to interrupt the execution of the program when we -encounter a decoding error. In that case, we can pass @unwanted id@ as the -leftover-handling 'Siphon'. 'unwanted' constructs a 'Siphon' that fails -when the stream produces any output at all, meaning it will fail if any -leftovers remain. 'unwanted' uses the first leftovers that apear in the -stream as the error value. So, in this example the error type will be -'ByteString': - ->>> executeFallibly (pipeo (encoded decodeUtf8 (unwanted id) intoLazyText)) (shell "echo ooo") -Right (ExitSuccess,"ooo\n") - -Notice also that we had to switch from 'execute' to 'executeFallibly'. This -is because, for the first time in the tutorial, we actually have a need for -the error type. 'execute' only works when the error type unifies with 'Void'. - -Beware: even if the error type is 'Void', exceptions can still be thrown. --} - - -{- $collstdoutstderrtext - -Sometimes we want to consume both @stdout@ and @stderr@, not independently, -but combined into a single stream. We can use 'pipeoec' for that. - -'pipeoec' takes as parameter a 'Siphon' for text, and two 'Lines' values -that know how to decode the bytes coming from @stdout@ and @stderr@ into -lines of text. - ->>> :{ - let - lin = toLines decodeUtf8 (pure id) - program = shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }" - in execute (pipeoec lin lin intoLazyText) program - :} -(ExitSuccess,"ooo\neee\n") - -We may wish to tag each line in the combined stream with its provenance. This can be done by using 'tweakLines' to modify each 'Lines' argument. - ->>> :{ - let - lin = toLines decodeUtf8 (pure id) - lin_stdout = tweakLines (\p -> P.yield "O" *> p) lin - lin_stderr = tweakLines (\p -> P.yield "E" *> p) lin - program = shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }" - in execute (pipeoec lin_stdout lin_stderr intoLazyText) program - :} -(ExitSuccess,"Oooo\nEeee\n") - --} - -{- $feedstdincollstdout - -We can feed bytes to @stdin@ while we read @stdout@ or @stderr@. We use the -'Pump' datatype for that. - ->>> execute (pipeio (fromLazyBytes "iii") intoLazyBytes) (shell "cat") -(ExitSuccess,((),"iii")) --} - - -{- $earlytermination - -An example of how returning a failure from a 'Siphon' interrupts the whole -computation and terminates the external program. - ->>> executeFallibly (pipeo (siphon (\_ -> return (Left "oops")))) (shell "sleep infinity") -Left "oops" --} - - -{- $collstdouttextlines - - ->>> :{ - let - osiphon - = getSiphonOp - $ contraencoded decodeUtf8 (pure id) - $ nest splitIntoLines intoLazyText - $ SiphonOp intoList - in - execute (pipeo osiphon) (shell "{ echo aa ; echo bb ; }") - :} -(ExitSuccess,["aa","bb"]) --} - - diff --git a/tests/doctests.hs b/tests/doctests.hs --- a/tests/doctests.hs +++ b/tests/doctests.hs @@ -1,11 +1,9 @@ module Main where -import Control.Applicative -import Control.Monad -import Data.List -import System.Directory -import System.FilePath import Test.DocTest main :: IO () -main = doctest ["src/System/Process/Streaming/Tutorial.hs"] +main = doctest [ + "src/System/Process/Streaming.hs", + "src/System/Process/Streaming/Text.hs" + ] diff --git a/tests/test.hs b/tests/test.hs --- a/tests/test.hs +++ b/tests/test.hs @@ -21,7 +21,7 @@ import Control.Monad import Control.Monad.Trans.Except import Control.Exception -import Control.Lens (view) +import Lens.Family (view) import Pipes import qualified Pipes.ByteString as B import qualified Pipes.Prelude as P @@ -37,8 +37,13 @@ import System.IO.Error import System.Exit import System.Directory -import System.Process.Streaming.Extended +import System.Process.Streaming +import Pipes.Transduce +import Pipes.Transduce.Text +import qualified Pipes.Transduce.Text as PT +import Pipes.Transduce.ByteString +main :: IO () main = defaultMain tests tests :: TestTree @@ -50,14 +55,11 @@ , testFailIfAnythingShowsInStderr , testTwoTextParsersInParallel , testCountWords - , testSingletonPipeline - , testBasicPipeline - , testBranchingPipeline , testDrainageDeadlock , testAlternatingWithCombined , testDecodeFailure - , testDecodeFailureX - , testPiapIn + , testMultipleFeeds + , testMultipleFeedsNoPiped ] ------------------------------------------------------------------------------- @@ -65,13 +67,14 @@ testCollectStdoutStderrAsByteString = testCase "collectStdoutStderrAsByteString" $ do r <- collectStdoutStderrAsByteString case r of - (ExitSuccess,("ooo\nppp\n","eee\nffff\n")) -> return () + ("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 ; }") +collectStdoutStderrAsByteString :: IO (BL.ByteString,BL.ByteString) +collectStdoutStderrAsByteString = + execute + (piped (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }")) + (liftA2 (,) (foldOut intoLazyBytes) (foldErr intoLazyBytes)) ------------------------------------------------------------------------------- @@ -79,14 +82,14 @@ testFeedStdinCollectStdoutAsText = testCase "feedStdinCollectStdoutAsText" $ do r <- feedStdinCollectStdoutAsText case r of - (ExitSuccess,((),"aaaaaa\naaaaa")) -> return () + "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") +feedStdinCollectStdoutAsText :: IO Text +feedStdinCollectStdoutAsText = + execute + (piped (shell "cat")) + (feedUtf8 (Just "aaaaaa\naaaaa") *> foldOut (transduce1 utf8x intoLazyText)) ------------------------------------------------------------------------------- @@ -94,7 +97,7 @@ testCombinedStdoutStderr = testCase "testCombinedStdoutStderr" $ do r <- combinedStdoutStderr case r of - (ExitSuccess,TL.lines -> ls) -> do + (TL.lines -> ls) -> do assertEqual "line count" (Prelude.length ls) 4 assertBool "expected lines" $ getAll $ foldMap (All . flip Prelude.elem ls) $ @@ -103,14 +106,11 @@ , "errprefix: eee" , "errprefix: ffff" ] - _ -> assertFailure "oops" -combinedStdoutStderr :: IO (ExitCode,TL.Text) +combinedStdoutStderr :: IO TL.Text combinedStdoutStderr = execute - (pipeoec (toLines T.decodeIso8859_1 (pure id)) - (tweakLines annotate $ toLines T.decodeIso8859_1 (pure id)) - (fromFold T.toLazyM)) - (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") + (piped (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }")) + (foldOutErr (combined (PT.lines utf8x) (groups annotate . PT.lines $ utf8x) intoLazyText)) where annotate x = P.yield "errprefix: " *> x @@ -124,10 +124,10 @@ Left "interrupted" -> return () _ -> assertFailure "oops" -interruptExecution :: IO (Either String (ExitCode,())) +interruptExecution :: IO (Either String ()) interruptExecution = executeFallibly - (pipeo . siphon $ \_ -> runExceptT . throwE $ "interrupted") - (shell "sleep 100s") + (piped (shell "sleep 100s")) + (foldOut $ withFallibleCont $ \_ -> runExceptT . throwE $ "interrupted") ------------------------------------------------------------------------------- @@ -139,10 +139,10 @@ Left "morestuff\n" -> return () _ -> assertFailure "oops" -failIfAnythingShowsInStderr :: IO (Either T.ByteString (ExitCode,())) +failIfAnythingShowsInStderr :: IO (Either T.ByteString ()) failIfAnythingShowsInStderr = executeFallibly - (pipee (unwanted ())) - (shell "{ echo morestuff 1>&2 ; sleep 100s ; }") + (piped (shell "{ echo morestuff 1>&2 ; sleep 100s ; }")) + (foldErr trip) ------------------------------------------------------------------------------- @@ -150,179 +150,100 @@ testTwoTextParsersInParallel = testCase "twoTextParsersInParallel" $ do r <- twoTextParsersInParallel case r of - Right (ExitSuccess,("ooooooo","aaaaaa")) -> return () + Right ("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 :: A.Parser [Char] parser1 = parseChars 'o' +parser2 :: A.Parser [Char] parser2 = parseChars 'a' -twoTextParsersInParallel :: IO (Either String (ExitCode,([Char], [Char]))) -twoTextParsersInParallel = executeFallibly - (pipeo (encoded T.decodeIso8859_1 (pure id) $ +twoTextParsersInParallel :: IO (Either String ([Char], [Char])) +twoTextParsersInParallel = + executeFallibly + (piped (shell "{ echo ooaaoo ; echo aaooaoa; }")) + (foldOut (transduce1 utf8x $ (,) <$> adapt parser1 <*> adapt parser2)) - (shell "{ echo ooaaoo ; echo aaooaoa; }") where - adapt p = fromParser $ do + adapt p = withParser $ 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 ; }") - -------------------------------------------------------------------------------- -testSingletonPipeline :: TestTree -testSingletonPipeline = testCase "singletonPipeline" $ do - r <- singletonPipeline - case r of - (Right ("ooo\nppp\n","eee\nffff\n")) -> return () - _ -> assertFailure "oops" - -singletonPipeline :: IO (Either Int (BL.ByteString,BL.ByteString)) -singletonPipeline = executePipelineFallibly - (pipeoe (fromFold B.toLazyM) - (fromFold B.toLazyM)) - (pure $ stage (toLines T.decodeUtf8 (pure id)) pipefail $ - shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }" - ) - -------------------------------------------------------------------------------- -testBasicPipeline :: TestTree -testBasicPipeline = testCase "basicPipeline" $ do - r <- basicPipeline - case r of - Right ((),"aaaccc\n") -> return () + 3 -> return () _ -> assertFailure "oops" -basicPipeline :: IO (Either Int ((),BL.ByteString)) -basicPipeline = executePipelineFallibly - (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") - (fromFold B.toLazyM)) - (fmap (stage (toLines T.decodeUtf8 (pure id)) pipefail) $ - Node (shell "grep aaa") [Node (shell "grep ccc") []] ) - -------------------------------------------------------------------------------- - -testBranchingPipeline :: TestTree -testBranchingPipeline = testCase "branchingPipeline" $ do - exists <- doesFileExist branchingPipelineFile - when exists $ removeFile branchingPipelineFile - r <- branchingPipeline - case r of - (Right ("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 (Either Int (BL.ByteString, BL.ByteString)) -branchingPipeline = executePipelineFallibly - (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) - (Node rootStage - [ Node branch1 [Node terminalStage1 []] - , Node branch2 [Node terminalStage2 []] - ] - ) - where - succStage = P.map (Data.ByteString.map succ) - - rootStage :: Stage Int - rootStage = stage (toLines T.decodeIso8859_1 (pure id)) - pipefail - (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ; echo ffff 1>&2 ; }") - - branch1 :: Stage Int - branch1 = stage (toLines T.decodeIso8859_1 (pure id)) - pipefail - (shell "grep ooo") - branch2 :: Stage Int - branch2 = stage (toLines T.decodeIso8859_1 (pure id)) - pipefail - (shell "grep xxx") - - terminalStage1 :: Stage Int - terminalStage1 = inbound (\p -> p >-> succStage) $ - stage (toLines T.decodeIso8859_1 (pure id)) - pipefail - (shell "tr -d b") - - terminalStage2 :: Stage Int - terminalStage2 = inbound (\p -> p >-> succStage) $ - stage (toLines T.decodeIso8859_1 (pure id)) - pipefail - (shell $ "cat > " ++ branchingPipelineFile) +countWords :: IO Int +countWords = + execute + (piped (shell "{ echo aaa ; echo bbb ; echo ccc ; }")) + (foldOut (transduce1 PT.utf8x $ + withCont $ P.sum . G.folds const () (const 1) . view T.words)) ------------------------------------------------------------------------------- testDrainageDeadlock :: TestTree testDrainageDeadlock = localOption (mkTimeout $ 20*(10^6)) $ testCase "drainageDeadlock" $ do - execute nopiping $ shell "chmod u+x tests/alternating.sh" + execute (piped (shell "chmod u+x tests/alternating.sh")) (pure ()) r <- drainageDeadlock case r of - (ExitSuccess,((),())) -> return () + ExitSuccess -> return () _ -> assertFailure "oops" -- A bug caused some streams not to be drained, and this caused problems -- due to full output buffers. -drainageDeadlock :: IO (ExitCode,((),())) -drainageDeadlock = execute - (pipeoe (pure ()) (fromFold $ \producer -> next producer >> pure ())) - (proc "tests/alternating.sh" []) - +drainageDeadlock :: IO ExitCode +drainageDeadlock = + execute + (piped (proc "tests/alternating.sh" [])) + (foldErr (withCont $ \producer -> next producer >> pure ()) *> exitCode) ------------------------------------------------------------------------------- testAlternatingWithCombined :: TestTree testAlternatingWithCombined = localOption (mkTimeout $ 20*(10^6)) $ testCase "testAlternatingWithCombined" $ do - execute nopiping $ shell "chmod u+x tests/alternating.sh" + execute (piped (shell "chmod u+x tests/alternating.sh")) (pure ()) r <- alternatingWithCombined case r of - (ExitSuccess,80000) -> return () - _ -> assertFailure "oops" + 80000 -> return () + _ -> assertFailure $ "unexpected lines (1) " ++ show r r <- alternatingWithCombined2 case r of - (ExitSuccess,(80000,80000)) -> return () - _ -> assertFailure "oops" + (80000,80000) -> return () + _ -> assertFailure $ "unexpected lines (2) " ++ show r -alternatingWithCombined :: IO (ExitCode,Integer) -alternatingWithCombined = execute - (pipeoec lp lp countLines) - (proc "tests/alternating.sh" []) +alternatingWithCombined :: IO Integer +alternatingWithCombined = + execute + (piped (proc "tests/alternating.sh" [])) + (foldOutErr (combined lp lp countLines)) where - lp = toLines T.decodeIso8859_1 (pure id) - countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines + lp = PT.lines PT.utf8x + countLines = withCont $ P.sum . G.folds const () (const 1) . view T.lines -alternatingWithCombined2 :: IO (ExitCode,(Integer,Integer)) -alternatingWithCombined2 = execute - (pipeoec lp lp $ (,) <$> countLines <*> countLines) - (proc "tests/alternating.sh" []) +alternatingWithCombined2 :: IO (Integer,Integer) +alternatingWithCombined2 = + execute + (piped (proc "tests/alternating.sh" [])) + (foldOutErr (combined lp lp (liftA2 (,) countLines countLines))) where - lp = toLines T.decodeIso8859_1 (pure id) - countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines + lp = PT.lines PT.utf8x + countLines = withCont $ P.sum . G.folds const () (const 1) . view T.lines ------------------------------------------------------------------------------- @@ -331,55 +252,56 @@ testCase "testDecodeFailure" $ do r <- decodeFailure case r of - Left nonAscii -> return () + Left _ -> return () _ -> assertFailure "oops" nonAscii :: BL.ByteString nonAscii = TL.encodeUtf8 "\x4e2d" -decodeFailure :: IO (Either T.ByteString (ExitCode,((),TL.Text))) -decodeFailure = executeFallibly - (pipeio (fromLazyBytes ("aaaaaaaa" <> nonAscii)) - (encoded decodeAscii (unwanted id) intoLazyText)) - (shell "cat") - - -testDecodeFailureX :: TestTree -testDecodeFailureX = localOption (mkTimeout $ 20*(10^6)) $ - testCase "testDecodeFailureX" $ do - r <- tryLeftoverB decodeFailureX - case r of - Left (LeftoverException _ nonAscii) -> return () - _ -> assertFailure "oops" - -tryLeftoverB :: IO a -> IO (Either (LeftoverException Data.ByteString.ByteString) a) -tryLeftoverB = Control.Exception.try - -decodeFailureX :: IO (ExitCode,((),TL.Text)) -decodeFailureX = execute - (pipeio (fromLazyBytes ("aaaaaaaa" <> nonAscii)) - (encoded decodeAscii _leftoverX intoLazyText)) - (shell "cat") +decodeFailure :: IO (Either T.ByteString TL.Text) +decodeFailure = + executeFallibly + (piped (shell "cat")) + (feedLazyBytes ("aaaaaaaa" <> nonAscii) *> + foldOut (transduce1 (decoder T.decodeAscii) intoLazyText)) ------------------------------------------------------------------------------- -testPiapIn :: TestTree -testPiapIn = localOption (mkTimeout $ 5*(10^6)) $ - testCase "piapIn" $ do - r <- piapIn - case r of - (ExitSuccess,"aaabbb") -> return () - _ -> assertFailure "oops" +testMultipleFeeds :: TestTree +testMultipleFeeds = testCase "testMultipleFeeds" $ do + r <- multipleFeeds + case r of + "firstline1secondline2thirdline3" -> return () + _ -> assertFailure (show r) -piapIn :: IO (ExitCode, BL.ByteString) -piapIn = execute piping (shell "cat") - where - piping = toPiping $ - piapi (fromLazyBytes "aaa") - *> - piapi (fromLazyBytes "bbb") - *> - piapo intoLazyBytes +multipleFeeds :: IO BL.ByteString +multipleFeeds = + execute + (piped (shell "cat")) + (feedBytes ["first","line1"] + *> + feedBytes ["second","line2"] + *> + feedBytes ["third","line3"] + *> + foldOut intoLazyBytes) -------------------------------------------------------------------------------- +testMultipleFeedsNoPiped :: TestTree +testMultipleFeedsNoPiped = localOption (mkTimeout $ 1*(10^6)) $ + testCase "testMultipleFeedsNoPiped" $ do + r <- multipleFeedsNoPiped + case r of + "foo\n" -> return () + _ -> assertFailure (show r) +multipleFeedsNoPiped :: IO BL.ByteString +multipleFeedsNoPiped = + execute + ((piped (shell "echo foo")) { std_in = Inherit }) + (feedBytes ["first","line1"] + *> + feedBytes ["second","line2"] + *> + feedBytes ["third","line3"] + *> + foldOut intoLazyBytes)