process-streaming 0.6.2.2 → 0.6.5.0
raw patch · 5 files changed
+549/−119 lines, 5 filesdep +contravariantdep +doctestdep +filepathdep ~pipesdep ~text
Dependencies added: contravariant, doctest, filepath, foldl
Dependency ranges changed: pipes, text
Files
- CHANGELOG +10/−0
- process-streaming.cabal +45/−4
- src/System/Process/Streaming.hs +270/−115
- src/System/Process/Streaming/Tutorial.hs +213/−0
- tests/doctests.hs +11/−0
CHANGELOG view
@@ -1,3 +1,13 @@+0.6.5.0+-------++- Brought back the tutorial.+- Added newtype SiphonOp with Contravariant, Divisible and Decidable instances.+- PipingPolicy now deprecated, use Piping instead.+- LinePolicy now deprecated, use Lines instead.+- linePolicy now deprecated, use toLines instead.+- Siphon constructors for folds from the "foldl" package.+ 0.6.2.2 -------
process-streaming.cabal view
@@ -1,5 +1,5 @@ name: process-streaming-version: 0.6.2.2+version: 0.6.5.0 license: BSD3 license-file: LICENSE data-files: @@ -19,9 +19,11 @@ default-language: Haskell2010 hs-source-dirs: src exposed-modules: - System.Process.Streaming System.Process.Lens+ System.Process.Streaming+ System.Process.Streaming.Tutorial other-modules: + ghc-options: -Wall -threaded -O2 build-depends: base >= 4.4 && < 5, transformers >= 0.2 && < 0.5,@@ -29,7 +31,7 @@ free >= 4.2 && < 5, bifunctors >= 4.1 && < 5, process >= 1.2.0 && < 1.3,- pipes >= 4.1.2 && < 4.2,+ pipes >= 4.1.3 && < 4.2, pipes-bytestring >= 2.1.0 && < 2.2, pipes-text >= 0.0.0.10 && < 0.0.2, text >= 0.11.2 && < 1.2.1,@@ -40,7 +42,9 @@ void >= 0.6 && < 0.7, containers >= 0.4, semigroups >= 0.15 && < 0.16,- conceit >= 0.2.1.0 && < 0.3.0.0+ conceit >= 0.2.1.0 && < 0.3.0.0,+ contravariant >= 1.2,+ foldl >= 1.0.7 Test-suite test default-language:@@ -51,6 +55,7 @@ tests main-is: test.hs+ ghc-options: -Wall -threaded build-depends: base >= 4.4 && < 5 , transformers >= 0.2 && < 0.5@@ -78,6 +83,42 @@ , bytestring >= 0.10 , lens >= 4 , directory >= 1.2++test-suite doctests+ type: exitcode-stdio-1.0+ ghc-options: -Wall -threaded+ hs-source-dirs: tests+ main-is: doctests.hs++ build-depends:+ base >= 4.4 && < 5+ , transformers >= 0.2 && < 0.5+ , transformers-compat == 0.3.*+ , free >= 4.2 && < 5+ , bifunctors >= 4.1 && < 5+ , process >= 1.2.0 && < 1.3+ , pipes >= 4.1.2 && < 4.2+ , pipes-bytestring >= 2.1.0 && < 2.2+ , pipes-text >= 0.0.0.10 && < 0.0.2+ , text >= 0.11.2 && < 1.2+ , pipes-concurrency >= 2.0.2 && < 3+ , pipes-safe >= 2.2.0 && < 3+ , pipes-parse >=3.0.1 && <3.1+ , exceptions >= 0.6.0 && < 0.7+ , void >= 0.6 && < 0.7+ , semigroups >= 0.15 && < 0.16+ , containers >= 0.4+ , process-streaming+ , tasty >= 0.9+ , tasty-hunit >= 0.9+ , attoparsec >= 0.11+ , pipes-attoparsec >= 0.5+ , pipes-group >= 1.0.1+ , bytestring >= 0.10+ , lens >= 4+ , directory >= 1.2+ , filepath+ , doctest >= 0.9.1 Source-repository head type: git
src/System/Process/Streaming.hs view
@@ -19,13 +19,15 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} + module System.Process.Streaming ( -- * Execution execute , executeFallibly -- * Piping Policies- , PipingPolicy+ , Piping , nopiping , pipeo , pipee@@ -40,25 +42,32 @@ -- * Pumping bytes into stdin , Pump (..) , fromProducer+ , fromProducerM , fromSafeProducer , fromFallibleProducer -- * Siphoning bytes out of stdout/stderr , Siphon+ , SiphonOp (..) , siphon , siphon' , fromFold , fromFold' , fromFold'_+ , fromFoldl+ , fromFoldlIO+ , fromFoldlM , fromConsumer+ , fromConsumerM , fromSafeConsumer , fromFallibleConsumer , fromParser+ , fromParserM , unwanted , DecodingFunction , encoded -- * Line handling- , LinePolicy- , linePolicy+ , Lines+ , toLines , tweakLines -- * Pipelines , executePipeline@@ -68,14 +77,24 @@ , stage , pipefail , inbound+-- -- * Utilities+-- -- $utilities+-- , surely -- * Re-exports -- $reexports- , module System.Process+ , module System.Process,+ -- * Deprecated+ -- $deprecated+ PipingPolicy,+ LinePolicy,+ linePolicy ) where import Data.Maybe import Data.Bifunctor import Data.Functor.Identity+import Data.Functor.Contravariant+import Data.Functor.Contravariant.Divisible import Data.Either import Data.Monoid import Data.Foldable@@ -87,11 +106,13 @@ import Data.List.NonEmpty import qualified Data.List.NonEmpty as N import Control.Applicative+import Control.Applicative.Lift import Control.Monad-import Control.Monad.Trans.Free+import Control.Monad.Trans.Free hiding (Pure) import Control.Monad.Trans.Except import Control.Monad.Trans.State import Control.Monad.Trans.Writer.Strict+import qualified Control.Foldl as L import qualified Control.Monad.Catch as C import Control.Exception import Control.Concurrent@@ -111,12 +132,12 @@ import System.Process.Lens import System.Exit -execute :: PipingPolicy Void a -> CreateProcess -> IO (ExitCode,a)+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 'PipingPolicy' argument. +a way defined by the 'Piping' argument. This function re-throws any 'IOException's it encounters. @@ -127,7 +148,7 @@ If an error @e@ or an exception happens, the external process is terminated. -}-executeFallibly :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a))+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 ())@@ -186,19 +207,19 @@ return $ Right (exitCode,r) {-|- A 'PipingPolicy' determines what standard streams will be piped and what to+ 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 'PipingPolicy' is parametrized by the type @e@ of errors that can abort+ 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 PipingPolicy e a = +data Piping e a = PPNone a | PPOutput (Producer ByteString IO () -> IO (Either e a)) | PPError (Producer ByteString IO () -> IO (Either e a))@@ -209,7 +230,7 @@ | PPInputOutputError ((Consumer ByteString IO (),IO (),Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a)) deriving (Functor) -instance Bifunctor PipingPolicy where+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@@ -223,59 +244,59 @@ {-| Do not pipe any standard stream. -}-nopiping :: PipingPolicy e ()+nopiping :: Piping e () nopiping = PPNone () {-| Pipe @stdout@. -}-pipeo :: Siphon ByteString e a -> PipingPolicy e a+pipeo :: Siphon ByteString e a -> Piping e a pipeo (runSiphon -> siphonout) = PPOutput $ siphonout {-| Pipe @stderr@. -}-pipee :: Siphon ByteString e a -> PipingPolicy e a+pipee :: Siphon ByteString e a -> Piping e a pipee (runSiphon -> siphonout) = PPError $ siphonout {-| Pipe @stdout@ and @stderr@. -}-pipeoe :: Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (a,b)+pipeoe :: Siphon ByteString e a -> Siphon ByteString e b -> Piping e (a,b) pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = PPOutputError $ uncurry $ separated siphonout siphonerr {-| Pipe @stdout@ and @stderr@ and consume them combined as 'Text'. -}-pipeoec :: LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e a+pipeoec :: Lines e -> Lines e -> Siphon Text e a -> Piping e a pipeoec policy1 policy2 (runSiphon -> siphon) = PPOutputError $ uncurry $ combined policy1 policy2 siphon {-| Pipe @stdin@. -}-pipei :: Pump ByteString e i -> PipingPolicy e i+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 -> PipingPolicy e (i,a)+pipeio :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a) pipeio (Pump feeder) (runSiphon -> 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 -> PipingPolicy e (i,a)+pipeie :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a) pipeie (Pump feeder) (runSiphon -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) -> (conceit (feeder consumer `finally` cleanup) (siphonerr producer)) {-| Pipe @stdin@, @stdout@ and @stderr@. -}-pipeioe :: Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (i,a,b)+pipeioe :: Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> Piping e (i,a,b) pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $ \(consumer,cleanup,outprod,errprod) -> (conceit (feeder consumer `finally` cleanup) @@ -286,7 +307,7 @@ {-| Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'. -}-pipeioec :: Pump ByteString e i -> LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e (i,a)+pipeioec :: Pump ByteString e i -> Lines e -> Lines e -> Siphon Text e a -> Piping e (i,a) pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $ \(consumer,cleanup,outprod,errprod) -> (conceit (feeder consumer `finally` cleanup) @@ -303,7 +324,7 @@ multiple streams. -} -data LinePolicy e = LinePolicy +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 ())) @@ -311,8 +332,9 @@ , lineTweaker :: forall r. Producer T.Text IO r -> Producer T.Text IO r } -instance Functor LinePolicy where- fmap f (LinePolicy func lt) = LinePolicy (\x y z -> fmap (bimap f id) $ func x y z) lt+-- | '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 {-|@@ -323,19 +345,19 @@ > (\x -> yield "prefix: " *> x) -}-tweakLines :: (forall r. Producer T.Text IO r -> Producer T.Text IO r) -> LinePolicy e -> LinePolicy e -tweakLines lt' (LinePolicy tear lt) = LinePolicy tear (lt' . lt) +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) {-|- Constructs a 'LinePolicy' out of a 'DecodingFunction' and a 'Siphon'+ Constructs a 'Lines' out of a 'DecodingFunction' and a 'Siphon' that specifies how to handle decoding failures. Passing @pure ()@ as the 'Siphon' will ignore any leftovers. Passing @unwanted ()@ will abort the computation if leftovers remain. -}-linePolicy :: DecodingFunction ByteString Text +toLines :: DecodingFunction ByteString Text -> Siphon ByteString e ()- -> LinePolicy e -linePolicy decoder lopo = LinePolicy+ -> Lines e +toLines decoder lopo = Lines (\tweaker teardown producer -> do let freeLines = transFreeT tweaker . viewLines @@ -346,11 +368,11 @@ id -- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here-combined :: LinePolicy e - -> LinePolicy e +combined :: Lines e + -> Lines e -> (Producer T.Text IO () -> IO (Either e a)) -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)-combined (LinePolicy fun1 twk1) (LinePolicy fun2 twk2) combinedConsumer prod1 prod2 = +combined (Lines fun1 twk1) (Lines fun2 twk2) combinedConsumer prod1 prod2 = manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer where manyCombined :: [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())]@@ -372,6 +394,9 @@ 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 producer = Pump $ safely $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) @@ -405,12 +430,13 @@ -> Siphon bytes e (a -> b) -> Siphon text e a -> Siphon bytes e b-encoded decoder policy activity = - Unhalting $ \producer ->- runExceptT $ do- (a,leftovers) <- ExceptT $ unhalting activity $ decoder producer - (f,r) <- ExceptT $ unhalting policy leftovers - pure (f a,r)+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) newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor @@ -454,76 +480,146 @@ '<*>' executes its arguments concurrently. The 'Producer' is forked so that each argument receives its own copy of the data. -}-data Siphon b e a = - Trivial a - | Unhalting (forall r. Producer b IO r -> IO (Either e (a,r)))- | Halting (Producer b IO () -> IO (Either e a))- deriving (Functor)+newtype Siphon b e a = Siphon (Lift (Siphon_ b e) a) deriving (Functor,Applicative) -instance Bifunctor (Siphon b) where- bimap f g s = case s of- Trivial a -> Trivial $ g a- Unhalting u -> Unhalting $ fmap (liftM (bimap f (bimap g id))) u- Halting h -> Halting $ fmap (liftM (bimap f g)) h+newtype SiphonOp e a b = SiphonOp { getSiphonOp :: Siphon b e a } -instance Applicative (Siphon b e) where- pure = Trivial- - s1 <*> s2 = case (s1,s2) of- (Trivial f,_) -> fmap f s2- (_,Trivial a) -> fmap ($ a) s1- (_,_) -> bifurcate (halting s1) (halting s2) - where - bifurcate fs as =- Unhalting $ \producer -> do- (outbox1,inbox1,seal1) <- spawn' Single- (outbox2,inbox2,seal2) <- spawn' Single- runConceit $- (,)- <$>- Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) - `finally` atomically seal1) - ((as $ fromInput inbox2) - `finally` atomically seal2) - )- <*>- Conceit ((fmap pure $ runEffect $ - producer >-> P.tee (toOutput outbox1 >> P.drain) - >-> (toOutput outbox2 >> P.drain)) - `finally` atomically seal1 `finally` atomically seal2- ) +-- | '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 -runSiphon :: Siphon b e a -> Producer b IO () -> IO (Either e a)-runSiphon s = case s of - h@(Halting _) -> halting $ Unhalting $ unhalting h - _ -> halting s+-- | '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) --- This might return a computation that *doesn't* completely drain the--- Producer.-halting :: Siphon b e a -> Producer b IO () -> IO (Either e a)-halting s = case s of - a@(Trivial _) -> halting $ Unhalting $ unhalting a- Unhalting u -> \producer -> liftM (fmap fst) $ u producer- Halting h -> h +-- | '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)) -unhalting :: Siphon b e a -> Producer b IO r -> IO (Either e (a,r))-unhalting s = case s of - Trivial a -> \producer -> do- r <- (runEffect $ producer >-> P.drain)- pure . pure $ (a,r)- Unhalting u -> u- Halting activity -> \producer -> do ++allowLefts :: Monad m => Pipe (Either b a) b m r+allowLefts = do+ e <- await+ case e of + Left l -> Pipes.yield l >> allowLefts+ Right _ -> allowLefts+ +allowRights :: Monad m => Pipe (Either b a) a m r+allowRights = do+ e <- await+ case e of + Right r -> Pipes.yield r >> allowRights+ Left _ -> allowRights+ ++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' Single+ (outbox2,inbox2,seal2) <- spawn' Single+ 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' Single runConceit $ (,) <$> Conceit (activity (fromInput inbox) `finally` atomically seal) <*>- Conceit ((fmap pure $ runEffect $ - producer >-> (toOutput outbox >> P.drain))- `finally` atomically seal- )+ _Conceit (runEffect (producer >-> (toOutput outbox >> P.drain)) + `finally` atomically seal+ ) +runSiphon :: Siphon b e a -> Producer b IO () -> IO (Either e a)+runSiphon (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++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@@ -531,6 +627,9 @@ fromConsumer :: Consumer b IO r -> Siphon b e () fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer +fromConsumerM :: MonadIO m => (m () -> IO (Either e a)) -> Consumer b m r -> Siphon b e a+fromConsumerM whittle consumer = siphon $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> mute consumer + fromSafeConsumer :: Consumer b (SafeT IO) r -> Siphon b e () fromSafeConsumer consumer = siphon $ safely $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer @@ -543,7 +642,19 @@ fromParser :: Parser b IO (Either e a) -> Siphon b e a fromParser parser = siphon $ Pipes.Parse.evalStateT parser + {-| + Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'.+ -}+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)++{-| Builds a 'Siphon' out of a computation that does something with a 'Producer', but may fail with an error of type @e@. @@ -552,7 +663,7 @@ -} siphon :: (Producer b IO () -> IO (Either e a)) -> Siphon b e a -siphon = Halting+siphon f = Siphon (Other (Nonexhaustive f)) {-| @@ -560,7 +671,7 @@ but may fail with an error of type @e@. -} siphon' :: (forall r. Producer b IO r -> IO (Either e (a,r))) -> Siphon b e a -siphon' = Unhalting+siphon' f = Siphon (Other (Exhaustive f)) {-| Useful in combination with 'Pipes.Text.toLazyM' from @pipes-text@ and@@ -579,18 +690,40 @@ 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)+ {-| Constructs a 'Siphon' that aborts the computation if the underlying 'Producer' produces anything. -} unwanted :: a -> Siphon b b a-unwanted a = Unhalting $ \producer -> do+unwanted a = siphon' $ \producer -> do n <- next producer return $ case n of Left r -> Right (a,r) Right (b,_) -> Left b -executePipeline :: PipingPolicy Void a -> Tree (Stage Void) -> IO a +executePipeline :: Piping Void a -> Tree (Stage Void) -> IO a executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline @@ -598,7 +731,7 @@ Similar to 'executeFallibly', but instead of a single process it executes a (possibly branching) pipeline of external processes. - The 'PipingPolicy' argument views the pipeline as a synthetic process+ The 'Piping' argument views the pipeline as a synthetic process for which @stdin@ is the @stdin@ of the first stage, @stdout@ is the @stdout@ of the leftmost terminal stage among those closer to the root, and @stderr@ is a combination of the @stderr@ streams of all the@@ -611,7 +744,7 @@ processes are not notified and keep going. There is no SIGPIPE-like functionality, in other words. -}-executePipelineFallibly :: PipingPolicy e a -> Tree (Stage e) -> IO (Either e a)+executePipelineFallibly :: Piping e a -> Tree (Stage e) -> IO (Either e a) executePipelineFallibly policy (Node (Stage cp lpol ecpol _) []) = case policy of PPNone a -> blende ecpol <$> executeFallibly policy cp PPOutput action -> blende ecpol <$> executeFallibly policy cp @@ -772,8 +905,8 @@ `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal ) -errorSiphonUTF8 :: MVar (Output ByteString) -> LinePolicy e -> Siphon ByteString e ()-errorSiphonUTF8 mvar (LinePolicy fun twk) = Halting $ fun twk iterTLines +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!!!@@ -791,7 +924,7 @@ data Stage e = Stage { processDefinition' :: CreateProcess - , stderrLinePolicy' :: LinePolicy e+ , stderrLines' :: Lines e , exitCodePolicy' :: ExitCode -> Either e () , inbound' :: forall r. Producer ByteString IO r -> Producer ByteString (ExceptT e IO) r } @@ -800,12 +933,12 @@ fmap f (Stage a b c d) = Stage a (fmap f b) (bimap f id . c) (hoist (mapExceptT $ liftM (bimap f id)) . d) {-|- Builds a 'Stage' out of a 'LinePolicy' that specifies how to handle+ Builds a 'Stage' out of a 'Lines' that specifies how to handle @stderr@ when piped, a function that determines whether an 'ExitCode' represents an error (some programs return non-standard exit codes) and a process definition. -}-stage :: LinePolicy e -> (ExitCode -> Either e ()) -> CreateProcess -> Stage e +stage :: Lines e -> (ExitCode -> Either e ()) -> CreateProcess -> Stage e stage lp ec cp = Stage cp lp ec (hoist lift) {-|@@ -819,19 +952,19 @@ data CreatePipeline e = CreatePipeline (Stage e) (NonEmpty (Tree (Stage e))) deriving (Functor) -executePipelineInternal :: (Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ())- -> (Pump ByteString e () -> Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ())- -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ())- -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ())+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- [] -> Halting $ \producer ->+ [] -> siphon $ \producer -> blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ pipe producer) lpol) cp- c1 : cs -> Halting $ \producer ->+ c1 : cs -> siphon $ \producer -> blende ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ pipe producer) (runNonEmpty ppend ppend' (c1 :| cs)) lpol) cp runNonEmpty ppend ppend' (b :| bs) = @@ -848,9 +981,31 @@ ExitSuccess -> Right () ExitFailure i -> Left i ++-- {- $utilities+-- +-- -} +-- surely :: ((forall r. m (a,r) -> IO (Either e (c,r))) -> x) -> (forall r. m (a,r) -> IO (c,r)) -> x+-- surely f f' = f (fmap Right . f')+ {- $reexports "System.Process" is re-exported for convenience. -} +{- $deprecated+ ++-} +{-# DEPRECATED PipingPolicy "Use Piping instead" #-} +type PipingPolicy e a = Piping e a ++{-# DEPRECATED LinePolicy "Use Lines instead" #-} +type LinePolicy e = Lines e ++{-# DEPRECATED linePolicy "Use toLines instead" #-} +linePolicy :: DecodingFunction ByteString Text + -> Siphon ByteString e ()+ -> Lines e +linePolicy = toLines
+ src/System/Process/Streaming/Tutorial.hs view
@@ -0,0 +1,213 @@++{-|+ @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 many 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+ ) 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.Monad.Except+>>> import Control.Lens (view)+>>> import Pipes+>>> import qualified Pipes.ByteString as B+>>> import qualified Pipes.Prelude as P+>>> import qualified Pipes.Parse as P+>>> import qualified Pipes.Attoparsec as P+>>> import qualified Pipes.Text as T+>>> import qualified Pipes.Text.Encoding as T+>>> import qualified Pipes.Text.IO as T+>>> import qualified Pipes.Group as G+>>> import qualified Pipes.Safe as S+>>> import qualified Pipes.Safe.Prelude as S+>>> import System.IO+>>> import System.IO.Error+>>> import System.Exit+>>> import System.Process.Streaming++-}++{- $collstdout ++This example uses the 'toLazyM' fold from @pipes-bytestring@.++>>> execute (pipeo (fromFold B.toLazyM)) (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 <$> fromFold B.toLazyM)) (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 (fromFold B.toLazyM) (fromFold B.toLazyM)) (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 (the example+uses one from @pipes-text@) 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 @(fromFold T.toLazyM)@". In other words: ignore any+leftovers.++>>> execute (pipeo (encoded T.decodeUtf8 (pure id) (fromFold T.toLazyM))) (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 T.decodeUtf8 (unwanted id) (fromFold T.toLazyM))) (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 is '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 T.decodeUtf8 (pure ()) + program = shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }"+ in execute (pipeoec lin lin (fromFold T.toLazyM)) 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 T.decodeUtf8 (pure ()) + 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 (fromFold T.toLazyM)) 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 (fromProducer (yield "iii")) (fromFold B.toLazyM)) (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"+-}
+ tests/doctests.hs view
@@ -0,0 +1,11 @@+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"]