diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -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 ------- 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.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 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 @@ -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 diff --git a/src/System/Process/Streaming/Tutorial.hs b/src/System/Process/Streaming/Tutorial.hs new file mode 100644 --- /dev/null +++ b/src/System/Process/Streaming/Tutorial.hs @@ -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" +-} diff --git a/tests/doctests.hs b/tests/doctests.hs new file mode 100644 --- /dev/null +++ b/tests/doctests.hs @@ -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"]