process-streaming 0.0.1.1 → 0.3.0.0
raw patch · 5 files changed
+434/−373 lines, 5 filesdep +bifunctorsdep +profunctorsdep −eitherdep −mmorphdep ~asyncdep ~basedep ~exceptions
Dependencies added: bifunctors, profunctors
Dependencies removed: either, mmorph
Dependency ranges changed: async, base, exceptions, mtl, pipes, pipes-safe, process, text, transformers
Files
- CHANGELOG +11/−0
- process-streaming.cabal +12/−12
- src/System/Process/Lens.hs +67/−49
- src/System/Process/Streaming.hs +283/−249
- src/System/Process/Streaming/Tutorial.hs +61/−63
CHANGELOG view
@@ -1,3 +1,14 @@+0.3.0.0 +------- + +- Renamed many functions (the number-based naming schema is gone). +- Decisions about how to consume the standard streams were moved to the new +"PipingPolicy" type. +- "ForkProd" is now "Siphon". +- "execute" can now throw IOExceptions (but see "safeExecute"). +- Some complicated types were hidden behind data declarations. +- Functor, Bifunctor and Profunctor instances for some types. + 0.0.1.1 -------
process-streaming.cabal view
@@ -1,5 +1,5 @@ name: process-streaming -version: 0.0.1.1 +version: 0.3.0.0 license: BSD3 license-file: LICENSE data-files: @@ -23,21 +23,21 @@ System.Process.Lens other-modules: build-depends: - base >= 4.4 && < 5, - transformers >= 0.3.0.0 && < 0.4, - mtl >= 2.1.1 && < 2.2, - either >= 4.1 && < 5, + base >= 4.4 && < 4.8, + transformers >= 0.2.0.0 && < 0.5, + mtl >= 2.0.1 && < 2.3, free >= 4.2 && < 5, - async >= 2.0.1.3 && < 3, - process >= 1.2.0.0 && < 1.3, - text >= 0.11.3.1 && < 1.2, - pipes >= 4.1.0 && < 4.2, + bifunctors >= 4.1 && < 5, + profunctors >= 3.1.1 && < 4.1, + async >= 2.0.1 && < 2.1, + process >= 1.2.0 && < 1.3, + pipes >= 4.0 && < 4.2, pipes-bytestring >= 2.0.0 && < 2.1, pipes-text >= 0.0.0.9 && < 0.0.2, + text >= 0.11.2 && < 1.2, pipes-concurrency >= 2.0.2 && < 3, - pipes-safe >= 2.0.2 && < 3, - mmorph >= 1.0.0 && < 2, - exceptions >= 0.3.3 && < 4 + pipes-safe >= 2.2.0 && < 3, + exceptions >= 0.6.0 && < 0.7 default-language: Haskell2010
src/System/Process/Lens.hs view
@@ -8,24 +8,25 @@ {-# LANGUAGE RankNTypes #-} module System.Process.Lens ( - _cmdspec, - _ShellCommand, - _RawCommand, - _cwd, - _env, - stream3, - pipe3, - pipe2, - pipe2h, - handle3, - handle2, + _cmdspec + , _ShellCommand + , _RawCommand + , _cwd + , _env + , streams + , _close_fds + , _create_group + , _delegate_ctlc + , handles + , nohandles + , handlesioe + , handlesoe ) where import Data.Maybe import Data.Functor.Identity import Data.Monoid import Data.Traversable -import Data.Text import Control.Applicative import System.IO import System.Process @@ -80,62 +81,79 @@ {-| A lens for the @(std_in,std_out,std_err)@ triplet. - > stream3 :: Lens' CreateProcess (StdStream,StdStream,StdStream) + > streams :: Lens' CreateProcess (StdStream,StdStream,StdStream) -} -stream3 :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess -stream3 f c = setStreams c <$> f (getStreams c) +streams :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess +streams f c = setStreams c <$> f (getStreams c) where - getStreams c = (std_in c,std_out c, std_err c) - setStreams c (s1,s2,s3) = c { std_in = s1 - , std_out = s2 - , std_err = s3 - } -{-| - > pipe3 = (CreatePipe,CreatePipe,CreatePipe) --} -pipe3 :: (StdStream,StdStream,StdStream) -pipe3 = (CreatePipe,CreatePipe,CreatePipe) + getStreams c = (std_in c,std_out c, std_err c) + setStreams c (s1,s2,s3) = c { std_in = s1 + , std_out = s2 + , std_err = s3 + } -{-| - Specifies @CreatePipe@ for @std_out@ and @std_err@; @std_in@ is set to 'Inherit'. +_close_fds :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_close_fds f c = set_close_fds c <$> f (close_fds c) + where + set_close_fds c cwd' = c { close_fds = cwd' } - > pipe3 = (Inherit,CreatePipe,CreatePipe) - -} -pipe2 :: (StdStream,StdStream,StdStream) -pipe2 = (Inherit,CreatePipe,CreatePipe) +_create_group :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_create_group f c = set_create_group c <$> f (create_group c) + where + set_create_group c cwd' = c { create_group = cwd' } + +_delegate_ctlc :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_delegate_ctlc f c = set_delegate_ctlc c <$> f (delegate_ctlc c) + where + set_delegate_ctlc c cwd' = c { delegate_ctlc = cwd' } + {-| - Specifies @CreatePipe@ for @std_out@ and @std_err@; @std_in@ is taken as -parameter. + A 'Lens' for the return value of 'createProcess' that focuses on the handles. + + > handlesioe :: Lens' (Maybe Handle, Maybe Handle, Maybe Handle,ProcessHandle) (Maybe Handle, Maybe Handle, Maybe Handle) -} -pipe2h :: Handle -> (StdStream,StdStream,StdStream) -pipe2h handle = (UseHandle handle,CreatePipe,CreatePipe) +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) + where + setHandles (c1'',c2'',c3'',c4'') (c1',c2',c3') = (c1',c2',c3',c4'') + getHandles (c1'',c2'',c3'',c4'') = (c1'',c2'',c3'') + +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' for the return value of 'createProcess' that removes the 'Maybe's from @stdin@, @stdout@ and @stderr@ or fails to match if any of them is 'Nothing'. + A 'Prism' that removes the 'Maybe's from @stdin@, @stdout@ and @stderr@ or fails to match if any of them is 'Nothing'. - > handle3 :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> ((Handle, Handle, Handle), ProcessHandle) + > handlesioe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) -> (Handle, Handle, Handle) -} -handle3 :: forall m. Applicative m => (((Handle, Handle, Handle), ProcessHandle) -> m ((Handle, Handle, Handle), ProcessHandle)) -> (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -handle3 f quad = case impure quad of +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, phandle) = Right ((h1, h2, h3), phandle) - impure x = Left x - justify ((h1, h2, h3), phandle) = (Just h1, Just h2, Just h3, phandle) + 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' for the return value of 'createProcess' that removes the 'Maybe's from @stdout@ and @stderr@ or fails to match if any of them is 'Nothing'. + A 'Prism' that removes the 'Maybe's from @stdout@ and @stderr@ or fails to match if any of them is 'Nothing'. - > handle2 :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> ((Handle, Handle), ProcessHandle) + > handlesoe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) -> (Handle, Handle) -} -handle2 :: forall m. Applicative m => (((Handle, Handle), ProcessHandle) -> m ((Handle, Handle), ProcessHandle)) -> (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -handle2 f quad = case impure quad of +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, phandle) = Right ((h2, h3), phandle) - impure x = Left x - justify ((h2, h3), phandle) = (Nothing, Just h2, Just h3, phandle) + impure (Nothing, Just h2, Just h3) = Right (h2, h3) + impure x = Left x + justify (h2, h3) = (Nothing, Just h2, Just h3)
src/System/Process/Streaming.hs view
@@ -9,61 +9,71 @@ -- There's also an emphasis in having error conditions explicit in the types, -- instead of throwing exceptions. -- --- See the functions 'execute' and 'execute3' for an entry point. Then the --- functions 'separate' and 'combineLines' that handle the consumption of --- stdout and stderr. --- -- Regular 'Consumer's, 'Parser's from @pipes-parse@ and folds from --- "Pipes.Prelude" (also folds @pipes-bytestring@ and @pipes-text@) can be used --- to consume the output streams of the external processes. +-- "Pipes.Prelude" (also folds from @pipes-bytestring@ and @pipes-text@) can be +-- used to consume the output streams of the external processes. -- ----------------------------------------------------------------------------- {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE RankNTypes #-} module System.Process.Streaming ( - -- * Execution helpers + -- * Execution execute - , execute3 , exitCode - , separate + , safeExecute + , simpleSafeExecute + -- * Piping standard streams + , PipingPolicy + , nopiping + , pipeoe + , pipeioe - -- * Execution with combined stdout/stderr + -- * Separated stdout/stderr + , separated + + -- * Stdout/stderr combined as text + , combined , LinePolicy , linePolicy - , LeftoverPolicy + + -- * Decoding and leftovers + , encoding + , LeftoverPolicy(..) , ignoreLeftovers , failOnLeftovers - , combineLines - -- * Constructing feeding/consuming functions + -- * Construction of feeding/consuming functions , useConsumer , useProducer , surely , safely , fallibly , monoidally - , exceptionally , nop - , encoding -- * Concurrency helpers - , Conc (..) - , conc - , mapConc - , ForkProd (..) - , forkProd + , Conceit (..) + , conceit + , mapConceit + , Siphon (..) + , forkSiphon + , SiphonL (..) + , SiphonR (..) + -- * Re-exports -- $reexports , module System.Process ) where import Data.Maybe +import Data.Bifunctor +import Data.Profunctor import Data.Functor.Identity import Data.Either -import Data.Either.Combinators import Data.Monoid import Data.Traversable import Data.Typeable @@ -71,10 +81,8 @@ import Control.Applicative import Control.Monad import Control.Monad.Trans.Free -import Control.Monad.Trans.Either import Control.Monad.Error import Control.Monad.State -import Control.Monad.Morph import Control.Monad.Writer.Strict import qualified Control.Monad.Catch as C import Control.Exception @@ -89,78 +97,62 @@ import Pipes.Concurrent import Pipes.Safe (SafeT, runSafeT) import System.IO +import System.IO.Error import System.Process import System.Process.Lens import System.Exit {-| - This function takes as arguments a 'CreateProcess' record, an exception -handler, and a consuming function for two 'Producers' associated to @stdout@ -and @stderr@, respectively. + Executes an external process. The standard streams are piped and consumed in +a way defined by the 'PipingPolicy' argument. - 'execute' tries to avoid launching exceptions, and represents all errors as -@e@ values. + This fuction re-throws any 'IOException's it encounters. - If the consuming function fails with @e@, the whole computation is -immediately aborted and @e@ is returned. + 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 or asynchronous exception happens, the external process is + If an error @e@ or an exception happens, the external process is terminated. - - This function sets the @std_out@ and @std_err@ fields in the 'CreateProcess' -record to 'CreatePipe'. -} -execute :: (Show e, Typeable e) - => CreateProcess - -> (IOException -> e) - -> (Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)) - -> IO (Either e (ExitCode,a)) -execute spec ehandler consumefunc = do - executeX handle2 spec' ehandler $ \(hout,herr) -> - (,) (consumefunc (fromHandle hout) (fromHandle herr)) - (hClose hout `finally` hClose herr) - where - spec' = spec { std_out = CreatePipe - , std_err = CreatePipe - } +execute :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a)) +execute (PipingPolicy tr somePrism action) procSpec = mask $ \restore -> do + (min,mout,merr,phandle) <- createProcess (tr procSpec) + case getFirst . getConst . somePrism (Const . First . Just) $ (min,mout,merr) of + Nothing -> + throwIO (userError "stdin/stdout/stderr handle unexpectedly null") + `finally` + terminateCarefully phandle + Just t -> + let (a, cleanup) = action t in + -- Handles must be closed *after* terminating the process, because a close + -- operation may block if the external process has unflushed bytes in the stream. + (restore (terminateOnError phandle a) `onException` terminateCarefully phandle) + `finally` + cleanup -{-| - Like `execute3` but with an additional argument consisting in a /feeding/ -function that takes the @stdin@ 'Consumer' and writes to it. +exitCode :: (ExitCode,a) -> Either Int a +exitCode (ec,a) = case ec of + ExitSuccess -> Right a + ExitFailure i -> Left i - Like the consuming function, the feeding function can return a value and -can also fail, terminating the process. +{-| + Like 'execute', but 'IOException's are caught and converted to the error type @e@. - The feeding function is executed /concurrently/ with the consuming -functions, not /before/ them. + Exit codes denoting errors are also converted to @e@ values. + -} +safeExecute :: (IOError -> e) -> (Int -> e) -> PipingPolicy e a -> CreateProcess -> IO (Either e a) +safeExecute exh ech pp cp = collapseEithers <$> (tryIOError $ execute pp cp) + where + collapseEithers = join . join . bimap exh (fmap (bimap ech id . exitCode)) - 'execute3' sets the @std_in@, @std_out@ and @std_err@ fields in the -'CreateProcess' record to 'CreatePipe'. +{-| + A simpler version of 'safeExecute' that assumes the error type @e@ is 'String'. -} -execute3 :: (Show e, Typeable e) - => CreateProcess - -> (IOException -> e) - -> (Consumer ByteString IO () -> IO (Either e a)) - -> (Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e b)) - -> IO (Either e (ExitCode,(a,b))) -execute3 spec ehandler feeder consumefunc = do - executeX handle3 spec' ehandler $ \(hin,hout,herr) -> - (,) (conc (feeder (toHandle hin) `finally` hClose hin) - (consumefunc (fromHandle hout) (fromHandle herr))) - (hClose hin `finally` hClose hout `finally` hClose herr) - where - spec' = spec { std_in = CreatePipe - , std_out = CreatePipe - , std_err = CreatePipe - } +simpleSafeExecute :: PipingPolicy String a -> CreateProcess -> IO (Either String a) +simpleSafeExecute = safeExecute show (mappend "Exit code: " . show) -try' :: (IOException -> e) -> IO (Either e a) -> IO (Either e a) -try' handler action = try action >>= either (return . Left . handler) return -createProcess' :: CreateProcess - -> IO (Either IOException (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle)) -createProcess' = try . createProcess - terminateCarefully :: ProcessHandle -> IO () terminateCarefully pHandle = do mExitCode <- getProcessExitCode pHandle @@ -181,43 +173,67 @@ exitCode <- waitForProcess pHandle return $ Right (exitCode,r) -executeX :: ((forall m. Applicative m => ((t, ProcessHandle) -> m (t, ProcessHandle)) -> (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle))) -> CreateProcess -> (IOException -> e) -> (t -> (IO (Either e a), IO())) -> IO (Either e (ExitCode,a)) -executeX somePrism procSpec exHandler action = mask $ \restore -> runEitherT $ do - maybeHtuple <- bimapEitherT exHandler id $ EitherT $ createProcess' procSpec - EitherT $ try' exHandler $ - case getFirst . getConst . somePrism (Const . First . Just) $ maybeHtuple of - Nothing -> - throwIO (userError "stdin/stdout/stderr handle unexpectedly null") - `finally` - let (_,_,_,phandle) = maybeHtuple - in terminateCarefully phandle - Just (htuple,phandle) -> let (a, cleanup) = action htuple in - -- Handles must be closed *after* terminating the process, because a close - -- operation may block if the external process has unflushed bytes in the stream. - (terminateOnError phandle $ restore a `onException` terminateCarefully phandle) - `finally` - cleanup +{-| + 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. +-} +data PipingPolicy e a = forall t. PipingPolicy (CreateProcess -> CreateProcess) (forall m. Applicative m => (t -> m t) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) (t -> (IO (Either e a), IO ())) + +instance Functor (PipingPolicy e) where + fmap f (PipingPolicy cpf prsm func) = PipingPolicy cpf prsm $ + (fmap (bimap (fmap (fmap f)) id) func) + +instance Bifunctor PipingPolicy where + bimap f g (PipingPolicy cpf prsm func) = PipingPolicy cpf prsm $ + (fmap (bimap (fmap (bimap f g)) id) func) + {-| - Convenience function that merges 'ExitFailure' values into the @e@ value. + Do not pipe any standard stream. +-} +nopiping :: PipingPolicy e () +nopiping = PipingPolicy id nohandles (\() -> (return $ return (), return ())) - The @e@ value is created from the exit code. +{-| + Pipe stderr and stdout. - Usually composed with the @execute@ functions. - -} -exitCode :: Functor c => (Int -> e) -> c (Either e (ExitCode,a)) -> c (Either e a) -exitCode f m = conversion <$> m - where - conversion r = case r of - Left e -> Left e - Right (code,a) -> case code of - ExitSuccess -> Right a - ExitFailure i -> Left $ f i + See also the 'separated' and 'combined' functions. +-} +pipeoe :: (Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)) + -> PipingPolicy e a +pipeoe consumefunc = PipingPolicy changecp handlesoe handler + where handler (hout,herr) = + (,) (consumefunc (fromHandle hout) (fromHandle herr)) + (hClose hout `finally` hClose herr) + changecp cp = cp { std_out = CreatePipe + , std_err = CreatePipe + } {-| + Pipe stdin, 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) +pipeioe feeder consumefunc = PipingPolicy changecp handlesioe handler + where handler (hin,hout,herr) = + (,) (conceit (feeder (toHandle hin) `finally` hClose hin) + (consumefunc (fromHandle hout) (fromHandle herr))) + (hClose hin `finally` hClose hout `finally` hClose herr) + changecp cp = cp { std_in = CreatePipe + , std_out = CreatePipe + , std_err = CreatePipe + } + +{-| 'separate' should be used when we want to consume @stdout@ and @stderr@ concurrently and independently. It constructs a function that can be plugged -into 'execute' or 'execute3'. +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 @@ -226,27 +242,17 @@ However, if any of the consuming functions fails with @e@, the whole computation fails immediately with @e@. -} -separate :: (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 outfunc errfunc outprod errprod = - conc (buffer_ outfunc outprod) - (buffer_ errfunc errprod) - -{-| - Type synonym for a function that takes a method to "tear down" a FreeT-based -list of lines as first parameter, a 'ByteString' source as second parameter, -and returns a (possibly failing) computation. Presumably, the bytes are decoded -into text, the text split into lines, and the "tear down" function applied. +separated :: (Show e, Typeable e) + => (Producer ByteString IO () -> IO (Either e a)) + -> (Producer ByteString IO () -> IO (Either e b)) + -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e (a,b)) +separated outfunc errfunc outprod errprod = + conceit (buffer_ outfunc outprod) (buffer_ errfunc errprod) -See the @pipes-group@ package for utilities on how to manipulate these -FreeT-based lists. They allow you to handle individual lines without forcing -you to have a whole line in memory at any given time. +data LinePolicy e = LinePolicy ((FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> Producer ByteString IO () -> IO (Either e ())) - See also 'linePolicy' and 'combineLines'. --} -type LinePolicy e = (FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> Producer ByteString IO () -> IO (Either e ()) +instance Functor LinePolicy where + fmap f (LinePolicy func) = LinePolicy $ fmap (fmap (fmap (bimap f id))) func {-| Constructs a 'LinePolicy'. @@ -268,30 +274,37 @@ linePolicy :: (forall r. Producer ByteString IO r -> Producer T.Text IO (Producer ByteString IO r)) -> (forall r. Producer T.Text IO r -> Producer T.Text IO r) - -> (LeftoverPolicy (Producer ByteString IO ()) e ()) + -> (LeftoverPolicy () ByteString e) -> LinePolicy e -linePolicy decoder transform lopo teardown producer = do - teardown freeLines >>= lopo () - where - freeLines = transFreeT transform - . viewLines - . decoder - $ producer - viewLines = getConst . T.lines Const +linePolicy decoder transform lopo = LinePolicy $ \teardown producer -> do + let freeLines = transFreeT transform + . viewLines + . decoder + $ producer + viewLines = getConst . T.lines Const + teardown freeLines >>= runLeftoverPolicy lopo () {-| In the Pipes ecosystem, leftovers from decoding operations are often stored -in the result value of 'Producer's (often as 'Producer's themselves). This is a -type synonym for a function that receives a value @a@ and some leftovers @l@, -and may modify the value or fail outright, depending of what the leftovers are. +in the result value of 'Producer's (as 'Producer's themselves). + + A 'LeftoverPolicy' receives a value @a@ and a producer of lefovers of type +@l@. Analyzing the producer, it may modify the value @a@ or fail outright, +depending of what the leftovers are. -} -type LeftoverPolicy l e a = a -> l -> IO (Either e a) +data LeftoverPolicy a l e = LeftoverPolicy { runLeftoverPolicy :: a -> Producer l IO () -> IO (Either e a) } +instance Functor (LeftoverPolicy a l) where + fmap f (LeftoverPolicy x) = LeftoverPolicy $ fmap (fmap (fmap (bimap f id))) x + +instance Profunctor (LeftoverPolicy a) where + dimap ab cd (LeftoverPolicy pf) = LeftoverPolicy $ \a p -> liftM (bimap cd id) $ pf a $ p >-> P.map ab + {-| Never fails for any leftover. -} -ignoreLeftovers :: LeftoverPolicy l e a -ignoreLeftovers a _ = return $ Right a +ignoreLeftovers :: LeftoverPolicy a l e +ignoreLeftovers = LeftoverPolicy $ pure . pure . pure {-| Fails if it encounters any leftover, and constructs the error out of the @@ -305,8 +318,8 @@ the error @a@ and/or the first undecoded values @b@ in your custom error datatype. -} -failOnLeftovers :: (a -> b -> e) -> LeftoverPolicy (Producer b IO ()) e a -failOnLeftovers errh a remainingBytes = do +failOnLeftovers :: (a -> b -> e) -> LeftoverPolicy a b e +failOnLeftovers errh = LeftoverPolicy $ \a remainingBytes -> do r <- next remainingBytes return $ case r of Left () -> Right a @@ -319,36 +332,35 @@ For both @stdout@ and @stderr@, a 'LinePolicy' must be supplied. - Like with 'separate', the streams are drained to completion if no errors + 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. - 'combineLines' returns a function that can be plugged into 'execute' or -'execute3'. + 'combined' returns a function that can be plugged into funtions like 'pipeioe'. - /Beware!/ 'combineLines' avoids situations in which a line emitted + /Beware!/ 'combined' avoids situations in which a line emitted in @stderr@ cuts a long line emitted in @stdout@, see <http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here> for a description of the problem. To avoid this, the combined text stream is locked while writing each individual line. But this means that if the external program stops writing to a handle /while in the middle of a line/, lines coming from the other handles won't get printed, either! -} -combineLines :: (Show e, Typeable e) - => LinePolicy e - -> LinePolicy e - -> (Producer T.Text IO () -> IO (Either e a)) - -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a) -combineLines fun1 fun2 combinedConsumer prod1 prod2 = - combineManyLines [fmap (($prod1).buffer_) fun1, fmap (($prod2).buffer_) fun2] combinedConsumer +combined :: (Show e, Typeable e) + => LinePolicy e + -> LinePolicy e + -> (Producer T.Text IO () -> IO (Either e a)) + -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a) +combined (LinePolicy fun1) (LinePolicy fun2) combinedConsumer prod1 prod2 = + manyCombined [fmap (($prod1).buffer_) fun1, fmap (($prod2).buffer_) fun2] combinedConsumer -combineManyLines :: (Show e, Typeable e) - => [((FreeT (Producer T.Text IO) IO (Producer ByteString IO ())) -> IO (Producer ByteString IO ())) -> IO (Either e ())] - -> (Producer T.Text IO () -> IO (Either e a)) - -> IO (Either e a) -combineManyLines actions consumer = do +manyCombined :: (Show e, Typeable e) + => [((FreeT (Producer T.Text IO) IO (Producer ByteString IO ())) -> IO (Producer ByteString IO ())) -> IO (Either e ())] + -> (Producer T.Text IO () -> IO (Either e a)) + -> IO (Either e a) +manyCombined actions consumer = do (outbox, inbox, seal) <- spawn' Unbounded mVar <- newMVar outbox - r <- conc (mapConc ($ iterTLines mVar) actions `finally` atomically seal) + r <- conceit (mapConceit ($ iterTLines mVar) actions `finally` atomically seal) (consumer (fromInput inbox) `finally` atomically seal) return $ snd <$> r where @@ -359,7 +371,7 @@ {-| Useful for constructing @stdout@ or @stderr@ consuming functions from a -'Consumer', to be plugged into 'separated' or 'combineLines'. +'Consumer', to be plugged into 'separated' or 'combined'. You may need to use 'surely' for the types to fit. -} @@ -367,8 +379,7 @@ useConsumer consumer producer = runEffect $ producer >-> consumer {-| - Useful for constructing @stdin@ feeding functions from a 'Producer', to be -plugged into 'execute3'. + Useful for constructing @stdin@ feeding functions from a 'Producer'. You may need to use 'surely' for the types to fit. -} @@ -382,21 +393,21 @@ > surely = fmap (fmap Right) -} -surely :: (Functor f, Functor f') => f (f' a) -> f (f' (Either e a)) +surely :: (Functor f0, Functor f1) => f0 (f1 a) -> f0 (f1 (Either e a)) surely = fmap (fmap Right) {-| Useful when we want to plug in a handler that does its work in the 'SafeT' transformer. -} -safely :: (MFunctor t, C.MonadCatch m, MonadIO m) +safely :: (MFunctor t, C.MonadMask m, MonadIO m) => (t (SafeT m) l -> (SafeT m) x) - -> t m l -> m x + -> t m l -> m x safely activity = runSafeT . activity . hoist lift fallibly :: (MFunctor t, Monad m, Error e) => (t (ErrorT e m) l -> (ErrorT e m) x) - -> t m l -> m (Either e x) + -> t m l -> m (Either e x) fallibly activity = runErrorT . activity . hoist lift {-| @@ -411,7 +422,7 @@ 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) + -> t m l -> m (Either e w) monoidally errh activity proxy = do (r,w) <- runWriterT . runErrorT . activity . hoist (lift.lift) $ proxy return $ case r of @@ -419,56 +430,57 @@ Right () -> Right $ w {-| - Useful when we want to construct different error values @e@ depending on -what feeding/consuming function throws an exeption, instead of relying in the -catch-all error callback supplied in 'execute' or 'execute3'. - -} -exceptionally :: (IOException -> e) - -> (x -> IO (Either e a)) - -> (x -> IO (Either e a)) -exceptionally handler operation x = try' handler (operation x) - -{-| - Value to plug into a 'separate' or 'combineLines' function when we are not -interested in doing anything with the handle. It returns immediately with @()@. + 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 -'combineLines' drain the streams to completion before returning. +'combined' drain the streams to completion before returning. -} -nop :: (MFunctor t, Monad m) => t m l -> m (Either e ()) -nop = \_ -> return $ Right () +nop :: Applicative m => i -> m (Either e ()) +nop = pure . pure . pure $ () buffer :: (Show e, Typeable e) - => LeftoverPolicy l e a - -> (Producer b IO () -> IO (Either e a)) - -> Producer b IO l -> IO (Either e a) + => LeftoverPolicy a l e + -> (Producer b IO () -> IO (Either e a)) + -> Producer b IO (Producer l IO ()) -> IO (Either e a) buffer policy activity producer = do (outbox,inbox,seal) <- spawn' Unbounded - r <- conc (do feeding <- async $ runEffect $ - producer >-> (toOutput outbox >> P.drain) + r <- conceit + (do feeding <- async $ runEffect $ + producer >-> (toOutput outbox >> P.drain) Right <$> wait feeding `finally` atomically seal ) (activity (fromInput inbox) `finally` atomically seal) case r of Left e -> return $ Left e - Right (lp,r') -> policy r' lp + Right (lp,r') -> runLeftoverPolicy policy r' lp +buffer_ :: (Show e, Typeable e) + => (Producer ByteString IO () -> IO (Either e a)) + -> Producer ByteString IO () -> IO (Either e a) +buffer_ activity producer = do + (outbox,inbox,seal) <- spawn' Unbounded + r <- conceit + (do feeding <- async $ runEffect $ + producer >-> (toOutput outbox >> P.drain) + Right <$> wait feeding `finally` atomically seal + ) + (activity (fromInput inbox) `finally` atomically seal) + return $ fmap snd r + {-| - Adapts a function that works with 'Producer's of decoded values so that it works with 'Producer's of still undecoded values, by supplying a decoding function and a 'LeftoverPolicy'. + Adapts a function that works with 'Producer's of decoded values so that it +works with 'Producer's of still undecoded values, by supplying a decoding +function and a 'LeftoverPolicy'. -} encoding :: (Show e, Typeable e) - => (Producer b IO r -> Producer t IO (Producer b IO r)) - -> LeftoverPolicy (Producer b IO r) e x - -> (Producer t IO () -> IO (Either e x)) - -> Producer b IO r -> IO (Either e x) + => (Producer b IO () -> Producer t IO (Producer b IO ())) + -> LeftoverPolicy a b e + -> (Producer t IO () -> IO (Either e a)) + -> Producer b IO () -> IO (Either e a) encoding decoder policy activity producer = buffer policy activity $ decoder producer -buffer_ :: (Show e, Typeable e) - => (Producer ByteString IO () -> IO (Either e a)) - -> Producer ByteString IO () -> IO (Either e a) -buffer_ = buffer ignoreLeftovers - data WrappedError e = WrappedError e deriving (Show, Typeable) @@ -482,7 +494,7 @@ (\(WrappedError e) -> return . Left $ e) {-| - 'Conc' is very similar to 'Control.Concurrent.Async.Concurrently' from the + 'Conceit' is very similar to 'Control.Concurrent.Async.Concurrently' from the @async@ package, but it has an explicit error type @e@. The 'Applicative' instance is used to run actions concurrently, wait until @@ -491,82 +503,104 @@ 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: 'Conc' behaves like 'Concurrently' for successes and + To put it another way: 'Conceit' behaves like 'Concurrently' for successes and like 'race' for errors. -} -newtype Conc e a = Conc { runConc :: IO (Either e a) } +newtype Conceit e a = Conceit { runConceit :: IO (Either e a) } -instance Functor (Conc e) where - fmap f (Conc x) = Conc $ fmap (fmap f) x +instance Functor (Conceit e) where + fmap f (Conceit x) = Conceit $ fmap (fmap f) x -instance (Show e, Typeable e) => Applicative (Conc e) where - pure = Conc . pure . pure - Conc fs <*> Conc as = - Conc . revealError $ +instance Bifunctor Conceit where + bimap f g (Conceit x) = Conceit $ liftM (bimap f g) x + +instance (Show e, Typeable e) => Applicative (Conceit e) where + pure = Conceit . pure . pure + Conceit fs <*> Conceit as = + Conceit . revealError $ uncurry ($) <$> concurrently (elideError fs) (elideError as) -instance (Show e, Typeable e) => Alternative (Conc e) where - empty = Conc $ forever (threadDelay maxBound) - Conc as <|> Conc bs = - Conc $ either id id <$> race as bs +instance (Show e, Typeable e) => Alternative (Conceit e) where + empty = Conceit $ forever (threadDelay maxBound) + Conceit as <|> Conceit bs = + Conceit $ either id id <$> race as bs -conc :: (Show e, Typeable e) - => IO (Either e a) - -> IO (Either e b) - -> IO (Either e (a,b)) -conc c1 c2 = runConc $ (,) <$> Conc c1 - <*> Conc c2 +instance (Show e, Typeable e, Monoid a) => Monoid (Conceit e a) where + mempty = Conceit . pure . pure $ mempty + mappend c1 c2 = (<>) <$> c1 <*> c2 +conceit :: (Show e, Typeable e) + => IO (Either e a) + -> IO (Either e b) + -> IO (Either e (a,b)) +conceit c1 c2 = runConceit $ (,) <$> Conceit c1 <*> Conceit c2 + {-| Works similarly to 'Control.Concurrent.Async.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@. -} -mapConc :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b)) -mapConc f = revealError . mapConcurrently (elideError . f) +mapConceit :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b)) +mapConceit f = revealError . mapConcurrently (elideError . f) {-| - 'ForkProd' is a newtype around a function that does something with 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 'Conc', if any of the functions fails with @e@ the others are +Like with 'Conceit', if any of the functions fails with @e@ the others are immediately cancelled and the whole computation fails with @e@. - 'ForkProd' and its accompanying functions are useful to run multiple + 'Siphon' and its accompanying functions are useful to run multiple parsers from "Pipes.Parse" in parallel over the same 'Producer'. -} -newtype ForkProd b e a = ForkProd { runForkProd :: Producer b IO () -> IO (Either e a) } +newtype Siphon b e a = Siphon { runSiphon :: Producer b IO () -> IO (Either e a) } -instance Functor (ForkProd b e) where - fmap f (ForkProd x) = ForkProd $ fmap (fmap (fmap f)) x +instance Functor (Siphon b e) where + fmap f (Siphon x) = Siphon $ fmap (fmap (fmap f)) x -instance (Show e, Typeable e) => Applicative (ForkProd b e) where - pure = ForkProd . pure . pure . pure - ForkProd fs <*> ForkProd as = - ForkProd $ \producer -> do +instance Bifunctor (Siphon b) where + bimap f g (Siphon x) = Siphon $ fmap (liftM (bimap f g)) x + +instance (Show e, Typeable e) => Applicative (Siphon b e) where + pure = Siphon . pure . pure . pure + Siphon fs <*> Siphon as = + Siphon $ \producer -> do (outbox1,inbox1,seal1) <- spawn' Unbounded (outbox2,inbox2,seal2) <- spawn' Unbounded - r <- conc (do - feeding <- async $ runEffect $ - producer >-> P.tee (toOutput outbox1 >> P.drain) - >-> (toOutput outbox2 >> P.drain) - sealing <- async $ wait feeding `finally` atomically seal1 - `finally` atomically seal2 - return $ Right () - ) - (fmap (uncurry ($)) <$> conc ((fs $ fromInput inbox1) - `finally` atomically seal1) - ((as $ fromInput inbox2) - `finally` atomically seal2) - ) + r <- conceit (do + feeding <- async $ runEffect $ + producer >-> P.tee (toOutput outbox1 >> P.drain) + >-> (toOutput outbox2 >> P.drain) + sealing <- async $ wait feeding `finally` atomically seal1 + `finally` atomically seal2 + return $ Right () + ) + (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) + `finally` atomically seal1) + ((as $ fromInput inbox2) + `finally` atomically seal2) + ) return $ fmap snd r -forkProd :: (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))) -forkProd c1 c2 = runForkProd $ (,) <$> ForkProd c1 - <*> ForkProd c2 +instance (Show e, Typeable e, Monoid a) => Monoid (Siphon b e a) where + mempty = Siphon . pure . pure . pure $ mempty + mappend s1 s2 = (<>) <$> s1 <*> s2 + +forkSiphon :: (Show e, Typeable e) + => (Producer b IO () -> IO (Either e x)) + -> (Producer b IO () -> IO (Either e y)) + -> Producer b IO () -> IO (Either e (x,y)) +forkSiphon c1 c2 = runSiphon $ (,) <$> Siphon c1 <*> Siphon c2 + +newtype SiphonL a b e = SiphonL { runSiphonL :: Producer b IO () -> IO (Either e a) } + +instance Profunctor (SiphonL e) where + dimap ab cd (SiphonL pf) = SiphonL $ \p -> liftM (bimap cd id) $ pf $ p >-> P.map ab + +newtype SiphonR e b a = SiphonR { runSiphonR :: Producer b IO () -> IO (Either e a) } + +instance Profunctor (SiphonR e) where + dimap ab cd (SiphonR pf) = SiphonR $ \p -> liftM (fmap cd) $ pf $ p >-> P.map ab {- $reexports
src/System/Process/Streaming/Tutorial.hs view
@@ -8,7 +8,7 @@ -- * Introduction -- $introduction - -- * stdin and stderr to different files + -- * Stdin and stderr to different files -- $stdinstderr -- * Missing executable @@ -47,6 +47,7 @@ > import Data.Monoid > import qualified Data.Attoparsec.Text as A > import Control.Applicative +> import Control.Monad > import Control.Lens (view) > import Pipes > import qualified Pipes.ByteString as B @@ -60,6 +61,8 @@ > 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 -} @@ -67,18 +70,17 @@ {- $stdinstderr -Using 'separate' to consume @stdout@ and @stderr@ concurrently, and functions +Using 'separated' to consume @stdout@ and @stderr@ concurrently, and functions from @pipes-safe@ to write the files. > example1 :: IO (Either String ((),())) -> example1 = exitCode show $ -> execute program show $ separate -> (consume "stdout.log") -> (consume "stderr.log") -> where -> consume file = surely . safely . useConsumer $ -> S.withFile file WriteMode B.toHandle -> program = shell "{ echo ooo ; echo eee 1>&2 ; }" +> example1 = simpleSafeExecute +> (pipeoe $ separated (consume "stdout.log") (consume "stderr.log")) +> (shell "{ echo ooo ; echo eee 1>&2 ; }") +> where +> consume file = surely . safely . useConsumer $ +> S.withFile file WriteMode B.toHandle + -} @@ -87,11 +89,8 @@ Missing executables and other 'IOException's are converted to an error type @e@ and returned in the 'Left' of an 'Either': -> example2 :: IO (Either String ((),())) -> example2 = exitCode show $ -> execute (proc "fsdfsdf" []) show $ separate -> nop -> nop +> example2 :: IO (Either String ()) +> example2 = simpleSafeExecute nopiping (proc "fsdfsdf" []) Returns: @@ -102,7 +101,7 @@ {- $combinelines -Here we use 'combineLines' to process 'stdout' and 'stderr' together. +Here we use 'combined' to process 'stdout' and 'stderr' together. Notice that they are consumed together as 'Text'. We have to specify a decoding function for each stream, and a 'LeftoverPolicy' as well. @@ -110,23 +109,23 @@ We also add a prefix to the lines coming from @stderr@. > example3 :: IO (Either String ()) -> example3 = exitCode show $ -> execute program show $ combineLines -> (linePolicy T.decodeIso8859_1 id policy) -> (linePolicy T.decodeIso8859_1 annotate policy) -> (surely . safely . useConsumer $ -> S.withFile "combined.txt" WriteMode T.toHandle) +> example3 = simpleSafeExecute +> (pipeoe $ combined +> (linePolicy T.decodeIso8859_1 id policy) +> (linePolicy T.decodeIso8859_1 annotate policy) +> (surely . safely . useConsumer $ +> S.withFile "combined.txt" WriteMode T.toHandle)) +> (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") > where -> policy = failOnLeftovers $ \_ _->"badbytes" -> annotate x = P.yield "errprefix: " *> x -> program = shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }" +> policy = failOnLeftovers $ \_ _->"badbytes" +> annotate x = P.yield "errprefix: " *> x -} {- $forkProd -Plugging parsers from @pipes-parse@ into 'separate' or 'combineLines' is easy +Plugging parsers from @pipes-parse@ into 'separated' or 'combined' is easy because running 'evalStateT' on a parser returns a function that consumes a 'Producer'. @@ -134,7 +133,7 @@ Pipes parsers using function 'parse' from package @pipes-attoparsec@. Stdout is decoded to Text and parsed by the two parsers in parallel using the -auxiliary 'forkProd' function. The results are aggregated in a tuple. +auxiliary 'forkSiphon' function. The results are aggregated in a tuple. Stderr is ignored using the 'nop' function. @@ -143,18 +142,22 @@ > many (A.notChar c) *> A.many1 (some (A.char c) <* many (A.notChar c)) > > parser1 = parseChars 'o' +> > parser2 = parseChars 'a' > > example4 ::IO (Either String (([Char], [Char]),())) -> example4 = exitCode show $ -> execute program show $ separate -> (encoding T.decodeIso8859_1 (failOnLeftovers $ \_ _->"badbytes") $ -> forkProd (P.evalStateT $ adapt parser1) -> (P.evalStateT $ adapt parser2)) -> nop +> example4 = simpleSafeExecute +> (pipeoe $ separated +> (encoding T.decodeIso8859_1 (failOnLeftovers $ \_ _->"badbytes") $ +> forkSiphon (adapt parser1) (adapt parser2)) +> nop) +> (shell "{ echo ooaaoo ; echo aaooaoa; }") > where -> adapt p = bimap (const "parse error") id <$> P.parse p -> program = shell "{ echo ooaaoo ; echo aaooaoa; }" +> adapt p = P.evalStateT $ do +> r <- P.parse p +> return $ case r of +> Just (Right r') -> Right r' +> _ -> Left "parse error" Returns: @@ -170,10 +173,9 @@ @e@. > example5 ::IO (Either String ((),())) -> example5 = exitCode show $ -> execute (shell "sleep 10s") show $ separate -> (\_ -> return $ Left "fast return!") -> nop +> example5 = simpleSafeExecute +> (pipeoe $ separated (\_ -> return $ Left "fast return!") nop) +> (shell "sleep 10s") Returns: @@ -193,20 +195,19 @@ @pipes-text@. Plugging folds defined in "Pipes.Prelude" (or @pipes-bytestring@ or -@pipes-text@) into 'separate' or 'combineLines' is easy because the folds -return functions that consume 'Producer's. Folds form the @foldl@ package -could also be useful here. +@pipes-text@) into 'separated' or 'combined' is easy because the folds +return functions that consume 'Producer's. Notice that @stdin@ is written concurrently with the reading of @stdout@. It is not the case that @sdtin@ is written first and then @stdout@ is read. -> example6 = exitCode show $ -> execute3 (shell "cat") show -> (surely . useProducer $ yield "aaaaaa\naaaaa") -> (separate -> (encoding T.decodeIso8859_1 ignoreLeftovers $ surely $ T.toLazyM) -> nop -> ) +> example6 = simpleSafeExecute +> (pipeioe +> (surely . useProducer $ yield "aaaaaa\naaaaa") +> (separated +> (encoding T.decodeIso8859_1 ignoreLeftovers $ surely $ T.toLazyM) +> nop)) +> (shell "cat") Returns: @@ -219,10 +220,9 @@ In this example we collect @stdout@ and @stderr@ as lazy bytestrings, using a fold defined in @pipes-bytestring@. -> example7 = exitCode show $ -> execute program show $ separate (surely B.toLazyM) (surely B.toLazyM) -> where -> program = shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }" +> example7 = simpleSafeExecute +> (pipeoe $ separated (surely B.toLazyM) (surely B.toLazyM)) +> (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") Returns: @@ -232,21 +232,19 @@ {- $wordcount - In this example we count words emitted to @stdout@ in a streaming fashing, + In this example we count words emitted to @stdout@ in a streaming fashion, without having to keep whole words in memory. We use a lens from @pipes-text@ to split the text into words, and a trivial fold from @pipes-group@ to create a 'Producer' of 'Int' values. Then we sum the ints using a fold from "Pipes.Prelude". -> example8 = exitCode show $ -> execute program show $ separate -> (encoding T.decodeIso8859_1 ignoreLeftovers $ surely $ -> P.sum . G.folds const () (const 1) . view T.words -> ) -> nop -> where -> program = shell "{ echo aaa ; echo bbb ; echo ccc ; }" +> example8 = simpleSafeExecute +> (pipeoe $ separated +> (encoding T.decodeIso8859_1 ignoreLeftovers $ surely $ +> P.sum . G.folds const () (const 1) . view T.words) +> nop) +> (shell "{ echo aaa ; echo bbb ; echo ccc ; }") -} @@ -255,7 +253,7 @@ Sometimes it's useful to launch external programs during a ghci session, like this: ->>> a <- async $ execute (proc "xeyes" []) show $ separate nop nop +>>> a <- async $ execute nopiping (proc "xeyes" []) Cancelling the async causes the termination of the external program: