System.Process.Streaming

This module contains helper functions and types built on top of System.Process and Pipes.

They provide concurrent, buffered (to avoid deadlocks) streaming access to the inputs and outputs of system processes.

There's also an emphasis in having error conditions explicit in the types, instead of throwing exceptions.

Regular Consumers, Parsers from pipes-parse and folds from Pipes.Prelude (also folds from pipes-bytestring and pipes-text) can be used to consume the output streams of the external processes.

Execution

execute :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode, a))

Executes an external process. The standard streams are piped and consumed in a way defined by the PipingPolicy argument.

This fuction re-throws any IOExceptions it encounters.

If the consumption of the standard streams fails with e, the whole computation is immediately aborted and e is returned. (An exception is not thrown in this case.).

If an error e or an exception happens, the external process is terminated.

exitCode :: (ExitCode, a) -> Either Int a

safeExecute :: (IOError -> e) -> (Int -> e) -> PipingPolicy e a -> CreateProcess -> IO (Either e a)

Like execute, but IOExceptions are caught and converted to the error type e.

Exit codes denoting errors are also converted to e values.

simpleSafeExecute :: PipingPolicy String a -> CreateProcess -> IO (Either String a)

A simpler version of safeExecute that assumes the error type e is String.

Piping standard streams

data PipingPolicy e a

A PipingPolicy specifies what standard streams of the external process should be piped, and how to consume them.

Values of type a denote successful consumption of the streams, values of type e denote errors.

Instances

Bifunctor PipingPolicy

Functor (PipingPolicy e)

nopiping :: PipingPolicy e ()

Do not pipe any standard stream.

pipeoe :: (Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)) -> PipingPolicy e a

Pipe stderr and stdout.

See also the separated and combined functions.

pipeioe :: (Show e, Typeable e) => (Consumer ByteString IO () -> IO (Either e a)) -> (Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e b)) -> PipingPolicy e (a, b)

Pipe stdin, stderr and stdout.

See also the separated and combined functions.

Separated stdout/stderr

separated :: (Show e, Typeable e) => (Producer ByteString IO () -> IO (Either e a)) -> (Producer ByteString IO () -> IO (Either e b)) -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e (a, b))

separate should be used when we want to consume stdout and stderr concurrently and independently. It constructs a function that can be plugged into functions like pipeoe.

If the consuming functions return with a and b, the corresponding streams keep being drained until the end. The combined value is not returned until both stdout and stderr are closed by the external process.

However, if any of the consuming functions fails with e, the whole computation fails immediately with e.

Stdout/stderr combined as text

combined :: (Show e, Typeable e) => LinePolicy e -> LinePolicy e -> (Producer Text IO () -> IO (Either e a)) -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)

The bytes from stdout and stderr are decoded into Text, splitted into lines (maybe applying some transformation to each line) and then combined and consumed by the function passed as argument.

For both stdout and stderr, a LinePolicy must be supplied.

Like with separated, the streams are drained to completion if no errors happen, but the computation is aborted immediately if any error e is returned.

combined returns a function that can be plugged into funtions like pipeioe.

Beware! combined avoids situations in which a line emitted in stderr cuts a long line emitted in stdout, see here for a description of the problem. To avoid this, the combined text stream is locked while writing each individual line. But this means that if the external program stops writing to a handle while in the middle of a line, lines coming from the other handles won't get printed, either!

data LinePolicy e

Instances

Functor LinePolicy

linePolicy :: (forall r. Producer ByteString IO r -> Producer Text IO (Producer ByteString IO r)) -> (forall r. Producer Text IO r -> Producer Text IO r) -> LeftoverPolicy () ByteString e -> LinePolicy e

Constructs a LinePolicy.

The first argument is a function function that decodes ByteString into Text. See the section Decoding Functions in the documentation for the Pipes.Text module.

The second argument is a function that modifies each individual line. The line is represented as a Producer to avoid having to keep it wholly in memory. If you want the lines unmodified, just pass id. Line prefixes are easy to add using applicative notation:

 (\x -> yield "prefix: " *> x)

The third argument is a LeftoverPolicy value that specifies how to handle decoding failures.

Decoding and leftovers

encoding :: (Show e, Typeable e) => (Producer b IO () -> Producer t IO (Producer b IO ())) -> LeftoverPolicy a b e -> (Producer t IO () -> IO (Either e a)) -> Producer b IO () -> IO (Either e a)

Adapts a function that works with Producers of decoded values so that it works with Producers of still undecoded values, by supplying a decoding function and a LeftoverPolicy.

data LeftoverPolicy a l e

In the Pipes ecosystem, leftovers from decoding operations are often stored in the result value of Producers (as Producers themselves).

A LeftoverPolicy receives a value a and a producer of lefovers of type l. Analyzing the producer, it may modify the value a or fail outright, depending of what the leftovers are.

Constructors

LeftoverPolicy

Fields

runLeftoverPolicy :: a -> Producer l IO () -> IO (Either e a)

Instances

Profunctor (LeftoverPolicy a)

Functor (LeftoverPolicy a l)

ignoreLeftovers :: LeftoverPolicy a l e

Never fails for any leftover.

failOnLeftovers :: (a -> b -> e) -> LeftoverPolicy a b e

Fails if it encounters any leftover, and constructs the error out of the first undedcoded data.

For simple error handling, just ignore the a and the undecoded data:

 (failOnLeftvoers (\_ _->"badbytes")) :: LeftoverPolicy (Producer b IO ()) String a

For more detailed error handling, you may want to include the result until the error a and/or the first undecoded values b in your custom error datatype.

Construction of feeding/consuming functions

useConsumer :: Monad m => Consumer b m () -> Producer b m () -> m ()

Useful for constructing stdout or stderr consuming functions from a Consumer, to be plugged into separated or combined.

You may need to use surely for the types to fit.

useProducer :: Monad m => Producer b m () -> Consumer b m () -> m ()

Useful for constructing stdin feeding functions from a Producer.

You may need to use surely for the types to fit.

surely :: (Functor f0, Functor f1) => f0 (f1 a) -> f0 (f1 (Either e a))

Useful when we want to plug in a handler that doesn't return an Either. For example folds from Pipes.Prelude, or functions created from simple Consumers with useConsumer.

 surely = fmap (fmap Right)

safely :: (MFunctor t, MonadMask m, MonadIO m) => (t (SafeT m) l -> SafeT m x) -> t m l -> m x

Useful when we want to plug in a handler that does its work in the SafeT transformer.

fallibly :: (MFunctor t, Monad m, Error e) => (t (ErrorT e m) l -> ErrorT e m x) -> t m l -> m (Either e x)

monoidally :: (MFunctor t, Monad m, Monoid w, Error e') => (e' -> w -> e) -> (t (ErrorT e' (WriterT w m)) l -> ErrorT e' (WriterT w m) ()) -> t m l -> m (Either e w)

Usually, it is better to use a fold form Pipes.Prelude instead of this function. But this function has the ability to return the monoidal result accumulated up until the error happened.

The first argument is a function that combines the initial error with the monoidal result to build the definitive error value. If you want to discard the results, use const as the first argument.

nop :: Applicative m => i -> m (Either e ())

Value to plug into separated or combined when we are not interested in doing anything with the stream. It returns immediately with ().

Notice that even if nop returns immediately, separate and combined drain the streams to completion before returning.

Concurrency helpers

newtype Conceit e a

Conceit is very similar to Concurrently from the async package, but it has an explicit error type e.

The Applicative instance is used to run actions concurrently, wait until they finish, and combine their results.

However, if any of the actions fails with e the other actions are immediately cancelled and the whole computation fails with e.

To put it another way: Conceit behaves like Concurrently for successes and like race for errors.

Constructors

Conceit

Fields

runConceit :: IO (Either e a)

Instances

Bifunctor Conceit

Functor (Conceit e)

(Show e, Typeable e) => Applicative (Conceit e)

(Show e, Typeable e) => Alternative (Conceit e)

(Show e, Typeable e, Monoid a) => Monoid (Conceit e a)

conceit :: (Show e, Typeable e) => IO (Either e a) -> IO (Either e b) -> IO (Either e (a, b))

mapConceit :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b))

Works similarly to mapConcurrently from the async package, but if any of the computations fails with e, the others are immediately cancelled and the whole computation fails with e.

newtype Siphon b e a

Siphon is a newtype around a function that does something with a Producer. The applicative instance fuses the functions, so that each one receives its own copy of the Producer and runs concurrently with the others. Like with Conceit, if any of the functions fails with e the others are immediately cancelled and the whole computation fails with e.

Siphon and its accompanying functions are useful to run multiple parsers from Pipes.Parse in parallel over the same Producer.

Constructors

Siphon

Fields

runSiphon :: Producer b IO () -> IO (Either e a)

Instances

Bifunctor (Siphon b)

Functor (Siphon b e)

(Show e, Typeable e) => Applicative (Siphon b e)

(Show e, Typeable e, Monoid a) => Monoid (Siphon b e a)

forkSiphon :: (Show e, Typeable e) => (Producer b IO () -> IO (Either e x)) -> (Producer b IO () -> IO (Either e y)) -> Producer b IO () -> IO (Either e (x, y))

newtype SiphonL a b e

Constructors

SiphonL

Fields

runSiphonL :: Producer b IO () -> IO (Either e a)

Instances

Profunctor (SiphonL e)

newtype SiphonR e b a

Constructors

SiphonR

Fields

runSiphonR :: Producer b IO () -> IO (Either e a)

Instances

Profunctor (SiphonR e)

Re-exports

System.Process is re-exported for convenience.

module System.Process