process-streaming 0.6.9.0 → 0.7.0.0
raw patch · 7 files changed
+790/−476 lines, 7 filesdep ~pipes-concurrency
Dependency ranges changed: pipes-concurrency
Files
- CHANGELOG +14/−0
- process-streaming.cabal +7/−5
- src/System/Process/Streaming.hs +208/−442
- src/System/Process/Streaming/Extended.hs +131/−0
- src/System/Process/Streaming/Internal.hs +397/−0
- src/System/Process/Streaming/Tutorial.hs +22/−18
- tests/test.hs +11/−11
CHANGELOG view
@@ -1,3 +1,17 @@+0.7.0.0+-------+- Changed signature of toLines to make it more consistent with that of encoded+ (breaking change).+- Stricter signatures of some fromConsumer... functions (breaking change).+- Pump constructor now hidden (breaking change).+- Added new fromConsumer... functions for Consumers that don't stop.+- Functions to throw exceptions for encoding errors (unwantedX, leftoverX,+ _leftoverX). Also a new LeftoverException.+- Re-exported more decoding functions from pipes-text.+- New module System.Process.Streaming.Extended with some experimental+ functions.+- New module System.Process.Streaming.Internal. + 0.6.9.0 ------- - fromLazyBytes, intoLazyBytes, intoLazyText.
process-streaming.cabal view
@@ -1,5 +1,5 @@ name: process-streaming-version: 0.6.9.0+version: 0.7.0.0 license: BSD3 license-file: LICENSE data-files: @@ -19,9 +19,11 @@ default-language: Haskell2010 hs-source-dirs: src exposed-modules: - System.Process.Lens- System.Process.Streaming+ System.Process.Streaming.Extended+ System.Process.Streaming.Internal System.Process.Streaming.Tutorial+ System.Process.Streaming+ System.Process.Lens other-modules: ghc-options: -Wall -threaded -O2 build-depends: @@ -67,7 +69,7 @@ , pipes-bytestring >= 2.1.0 && < 2.2 , pipes-text >= 0.0.0.10 && < 0.0.2 , text >= 0.11.2 && < 1.2.1- , pipes-concurrency >= 2.0.2 && < 3+ , pipes-concurrency >= 2.0.3 && < 3 , pipes-safe >= 2.2.0 && < 3 , pipes-parse >=3.0.1 && <3.1 , exceptions >= 0.6.0 && < 1.0@@ -101,7 +103,7 @@ , 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-concurrency >= 2.0.3 && < 3 , pipes-safe >= 2.2.0 && < 3 , pipes-parse >=3.0.1 && <3.1 , exceptions >= 0.6.0 && < 1
src/System/Process/Streaming.hs view
@@ -16,6 +16,7 @@ {-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE ViewPatterns #-}@@ -41,7 +42,7 @@ , pipeioec -- * Pumping bytes into stdin- , Pump (..)+ , Pump , fromProducer , fromProducerM , fromSafeProducer@@ -56,26 +57,33 @@ , fromFold , fromFold' , fromFold'_- , fromFoldl- , fromFoldlIO- , fromFoldlM , fromConsumer+ , fromConsumer' , fromConsumerM+ , fromConsumerM' , fromSafeConsumer , fromFallibleConsumer , fromParser , fromParserM - , unwanted+ , fromFoldl+ , fromFoldlIO+ , fromFoldlM , intoLazyBytes , intoLazyText + , unwanted , DecodingFunction , encoded , SiphonOp (..)- -- * Line handling+ -- * Handling lines , Lines , toLines , tweakLines , prefixLines+ -- * Throwing exceptions+ , unwantedX+ , LeftoverException (..)+ , leftoverX+ , _leftoverX -- * Pipelines , executePipelineFallibly , executePipeline@@ -88,38 +96,27 @@ -- $reexports , module System.Process , T.decodeUtf8 - -- * Deprecated- -- $deprecated- , PipingPolicy- , LinePolicy- , linePolicy + , T.decodeAscii + , T.decodeIso8859_1 ) where -import Data.Maybe import qualified Data.ByteString.Lazy as BL-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-import Data.Traversable+import Data.Typeable import Data.Tree-import Data.String import qualified Data.Text.Lazy as TL import Data.Text -import Data.Text.Encoding +import Data.Text.Encoding hiding (decodeUtf8) import Data.Void 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 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 Control.Exception import Control.Concurrent@@ -127,7 +124,6 @@ import Pipes import qualified Pipes as P import qualified Pipes.Prelude as P-import Pipes.Lift import Pipes.ByteString import Pipes.Parse import qualified Pipes.Text as T@@ -135,14 +131,15 @@ 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 +import System.Process.Streaming.Internal+ {-|- A simplified version of 'executeFallibly' for when the error type is `Void`.- Note however that this function may still throw exceptions.+ A simplified version of 'executeFallibly' for when the error type unifies+ with `Void`. Note however that this function may still throw exceptions. -} execute :: Piping Void a -> CreateProcess -> IO (ExitCode,a) execute pp cprocess = either absurd id <$> executeFallibly pp cprocess@@ -216,8 +213,8 @@ -> (t ->(IO (Either e a),IO ())) -> IO (Either e (ExitCode,a)) executeInternal record somePrism allocator = mask $ \restore -> do- (min,mout,merr,phandle) <- createProcess record- case getFirst . getConst . somePrism (Const . First . Just) $ (min,mout,merr) of+ (mi,mout,merr,phandle) <- createProcess record+ case getFirst . getConst . somePrism (Const . First . Just) $ (mi,mout,merr) of Nothing -> throwIO (userError "stdin/stdout/stderr handle unwantedly null") `finally`@@ -228,10 +225,6 @@ -- operation may block if the external process has unflushed bytes in the stream. (restore (terminateOnError phandle action) `onException` terminateCarefully phandle) `finally` cleanup -exitCode :: (ExitCode,a) -> Either Int a-exitCode (ec,a) = case ec of- ExitSuccess -> Right a - ExitFailure i -> Left i terminateCarefully :: ProcessHandle -> IO () terminateCarefully pHandle = do@@ -254,83 +247,6 @@ return $ Right (exitCode,r) {-|- A 'Piping' determines what standard streams will be piped and what to-do with them.-- The user doesn't need to manually set the 'std_in', 'std_out' and 'std_err'-fields of the 'CreateProcess' record to 'CreatePipe', this is done-automatically. -- A 'Piping' is parametrized by the type @e@ of errors that can abort-the processing of the streams.- -}--- Knows that there is a stdin, stdout and a stderr,--- but doesn't know anything about file handlers or CreateProcess.-data Piping e a = - PPNone a- | PPOutput - (Producer ByteString IO () - -> - IO (Either e a))- | PPError - (Producer ByteString IO () - -> - IO (Either e a))- | PPOutputError - ((Producer ByteString IO ()- ,Producer ByteString IO ()) - -> - IO (Either e a))- | PPInput - ((Consumer ByteString IO ()- ,IO ()) - -> - IO (Either e a))- | PPInputOutput - ((Consumer ByteString IO ()- ,IO ()- ,Producer ByteString IO ()) - -> - IO (Either e a))- | PPInputError - ((Consumer ByteString IO ()- ,IO () - ,Producer ByteString IO ()) - -> - IO (Either e a))- | PPInputOutputError - ((Consumer ByteString IO ()- ,IO ()- ,Producer ByteString IO ()- ,Producer ByteString IO ()) - -> - IO (Either e a))- deriving (Functor)---{-| - 'first' is useful to massage errors.--}-instance Bifunctor Piping where- bimap f g pp = case pp of- PPNone a -> PPNone $ - g a - PPOutput action -> PPOutput $ - fmap (fmap (bimap f g)) action- PPError action -> PPError $ - fmap (fmap (bimap f g)) action- PPOutputError action -> PPOutputError $ - fmap (fmap (bimap f g)) action- PPInput action -> PPInput $ - fmap (fmap (bimap f g)) action- PPInputOutput action -> PPInputOutput $ - fmap (fmap (bimap f g)) action- PPInputError action -> PPInputError $ - fmap (fmap (bimap f g)) action- PPInputOutputError action -> PPInputOutputError $ - fmap (fmap (bimap f g)) action--{-| Do not pipe any standard stream. -} nopiping :: Piping e ()@@ -340,27 +256,27 @@ Pipe @stdout@. -} pipeo :: Siphon ByteString e a -> Piping e a-pipeo (runSiphon -> siphonout) = PPOutput $ siphonout+pipeo (runSiphonDumb -> siphonout) = PPOutput $ siphonout {-| Pipe @stderr@. -} pipee :: Siphon ByteString e a -> Piping e a-pipee (runSiphon -> siphonout) = PPError $ siphonout+pipee (runSiphonDumb -> siphonout) = PPError $ siphonout {-| Pipe @stdout@ and @stderr@. -} pipeoe :: Siphon ByteString e a -> Siphon ByteString e b -> Piping e (a,b)-pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = +pipeoe (runSiphonDumb -> siphonout) (runSiphonDumb -> siphonerr) = PPOutputError $ uncurry $ separated siphonout siphonerr {-| Pipe @stdout@ and @stderr@ and consume them combined as 'Text'. -} pipeoec :: Lines e -> Lines e -> Siphon Text e a -> Piping e a-pipeoec policy1 policy2 (runSiphon -> siphon) = - PPOutputError $ uncurry $ combined policy1 policy2 siphon +pipeoec policy1 policy2 (runSiphonDumb -> s) = + PPOutputError $ uncurry $ combined policy1 policy2 s {-| Pipe @stdin@.@@ -372,21 +288,21 @@ Pipe @stdin@ and @stdout@. -} pipeio :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a)-pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) ->+pipeio (Pump feeder) (runSiphonDumb -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) -> (conceit (feeder consumer `finally` cleanup) (siphonout producer)) {-| Pipe @stdin@ and @stderr@. -} pipeie :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a)-pipeie (Pump feeder) (runSiphon -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) ->+pipeie (Pump feeder) (runSiphonDumb -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) -> (conceit (feeder consumer `finally` cleanup) (siphonerr producer)) {-| Pipe @stdin@, @stdout@ and @stderr@. -} pipeioe :: Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> Piping e (i,a,b)-pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $+pipeioe (Pump feeder) (runSiphonDumb -> siphonout) (runSiphonDumb -> siphonerr) = fmap flattenTuple $ PPInputOutputError $ \(consumer,cleanup,outprod,errprod) -> (conceit (feeder consumer `finally` cleanup) (separated siphonout siphonerr outprod errprod))@@ -397,10 +313,10 @@ Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'. -} pipeioec :: Pump ByteString e i -> Lines e -> Lines e -> Siphon Text e a -> Piping e (i,a)-pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $+pipeioec (Pump feeder) policy1 policy2 (runSiphonDumb -> s) = PPInputOutputError $ \(consumer,cleanup,outprod,errprod) -> (conceit (feeder consumer `finally` cleanup) - (combined policy1 policy2 siphon outprod errprod))+ (combined policy1 policy2 s outprod errprod)) separated :: (Producer ByteString IO () -> IO (Either e a)) -> (Producer ByteString IO () -> IO (Either e b))@@ -408,70 +324,6 @@ separated outfunc errfunc outprod errprod = conceit (outfunc outprod) (errfunc errprod) --- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here-combined :: Lines e - -> Lines e - -> (Producer T.Text IO () -> IO (Either e a))- -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)-combined (Lines fun1 twk1) (Lines fun2 twk2) combinedConsumer prod1 prod2 = - manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer - where - manyCombined :: [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())]- -> (Producer T.Text IO () -> IO (Either e a))- -> IO (Either e a) - manyCombined actions consumer = do- (outbox, inbox, seal) <- spawn' Single- mVar <- newMVar outbox- runConceit $ - Conceit (mapConceit ($ iterTLines mVar) actions `finally` atomically seal)- *>- Conceit (consumer (fromInput inbox) `finally` atomically seal)- where - iterTLines mvar = iterT $ \textProducer -> do- -- the P.drain bit was difficult to figure out!!!- join $ withMVar mvar $ \output -> do- runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain)---newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor---{-| - 'first' is useful to massage errors.--}-instance Bifunctor (Pump b) where- bimap f g (Pump x) = Pump $ fmap (liftM (bimap f g)) x--instance Applicative (Pump b e) where- pure = Pump . pure . pure . pure- Pump fs <*> Pump as = - Pump $ \consumer -> do- (outbox1,inbox1,seal1) <- spawn' Single- (outbox2,inbox2,seal2) <- spawn' Single- runConceit $ - Conceit (runExceptT $ do- r1 <- ExceptT $ (fs $ toOutput outbox1) - `finally` atomically seal1- r2 <- ExceptT $ (as $ toOutput outbox2) - `finally` atomically seal2- return $ r1 r2 - )- <* - Conceit (do- (runEffect $- (fromInput inbox1 >> fromInput inbox2) >-> consumer)- `finally` atomically seal1- `finally` atomically seal2- runExceptT $ pure ()- )--instance (Monoid a) => Monoid (Pump b e a) where- mempty = Pump . pure . pure . pure $ mempty- mappend s1 s2 = (<>) <$> s1 <*> s2--instance IsString b => IsString (Pump b e ()) where - fromString = fromProducer . P.yield . fromString - fromProducer :: Producer b IO r -> Pump b e () fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) @@ -493,170 +345,7 @@ fromLazyBytes :: BL.ByteString -> Pump ByteString e () fromLazyBytes = fromProducer . fromLazy -{-| - A 'Siphon' represents a computation that completely drains a producer, but-may fail early with an error of type @e@. -- -}-newtype Siphon b e a = Siphon (Lift (Siphon_ b e) a) deriving (Functor)---{-| - 'pure' creates a 'Siphon' that does nothing besides draining the-'Producer'. -- '<*>' executes its arguments concurrently. The 'Producer' is forked so- that each argument receives its own copy of the data.--}-instance Applicative (Siphon b e) where- pure a = Siphon (pure a)- (Siphon fa) <*> (Siphon a) = Siphon (fa <*> a)--data Siphon_ b e a = - Exhaustive (forall r. Producer b IO r -> IO (Either e (a,r)))- | Nonexhaustive (Producer b IO () -> IO (Either e a))- deriving (Functor)---instance Applicative (Siphon_ b e) where- pure a = Exhaustive $ \producer -> do- r <- runEffect (producer >-> P.drain)- pure (pure (a,r))- s1 <*> s2 = bifurcate (nonexhaustive s1) (nonexhaustive s2) - where - bifurcate fs as = Exhaustive $ \producer -> do- (outbox1,inbox1,seal1) <- spawn' 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 (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--{-| - 'first' is useful to massage errors.--}-instance Bifunctor (Siphon b) where- bimap f g (Siphon s) = Siphon $ case s of- Pure a -> Pure (g a)- Other o -> Other (bimap f g o)--instance (Monoid a) => Monoid (Siphon b e a) where- mempty = pure mempty- mappend s1 s2 = (<>) <$> s1 <*> s2--newtype SiphonOp e a b = SiphonOp { getSiphonOp :: Siphon b e a } ---- | 'contramap' carn turn a 'SiphonOp' for bytes into a 'SiphonOp' for text.-instance Contravariant (SiphonOp e a) where- contramap f (SiphonOp (Siphon s)) = SiphonOp . Siphon $ case s of- Pure p -> Pure p- Other o -> Other $ case o of- Exhaustive e -> Exhaustive $ \producer ->- e $ producer >-> P.map f- Nonexhaustive ne -> Nonexhaustive $ \producer ->- ne $ producer >-> P.map f---- | 'divide' builds a 'SiphonOp' for a composite out of the 'SiphonOp's--- for the parts.-instance Monoid a => Divisible (SiphonOp e a) where- divide divider siphonOp1 siphonOp2 = contramap divider . SiphonOp $ - (getSiphonOp (contramap fst siphonOp1)) - `mappend`- (getSiphonOp (contramap snd siphonOp2))- conquer = SiphonOp (pure mempty)---- | 'choose' builds a 'SiphonOp' for a sum out of the 'SiphonOp's--- for the branches.-instance Monoid a => Decidable (SiphonOp e a) where- choose chooser (SiphonOp s1) (SiphonOp s2) = - contramap chooser . SiphonOp $ - (contraPipeMapL s1) - `mappend`- (contraPipeMapR s2)- where- contraPipeMapL (Siphon s) = Siphon $ case s of- Pure p -> Pure p- Other o -> Other $ case o of- Exhaustive e -> Exhaustive $ \producer ->- e $ producer >-> allowLefts- Nonexhaustive ne -> Nonexhaustive $ \producer ->- ne $ producer >-> allowLefts- contraPipeMapR (Siphon s) = Siphon $ case s of- Pure p -> Pure p- Other o -> Other $ case o of- Exhaustive e -> Exhaustive $ \producer ->- e $ producer >-> allowRights- Nonexhaustive ne -> Nonexhaustive $ \producer ->- ne $ producer >-> allowRights- allowLefts = do- e <- await- case e of - Left l -> Pipes.yield l >> allowLefts- Right _ -> allowLefts- allowRights = do- e <- await- case e of - Right r -> Pipes.yield r >> allowRights- Left _ -> allowRights- lose f = SiphonOp . Siphon . Other . Nonexhaustive $ \producer -> do- n <- next producer - return $ case n of - Left () -> Right mempty- Right (b,_) -> Right (absurd (f b))--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- intoLazyBytes :: Siphon ByteString e BL.ByteString intoLazyBytes = fromFold toLazyM @@ -724,27 +413,42 @@ fromFoldlM whittle aFoldM = siphon' $ \producer -> whittle $ L.impurely P.foldM' aFoldM (hoist liftIO producer) -fromConsumer :: Consumer b IO r -> Siphon b e ()-fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer +fromConsumer :: Consumer b IO () -> Siphon b e ()+fromConsumer consumer = fromFold $ \producer -> runEffect $ producer >-> consumer +fromConsumer' :: Consumer b IO Void -> Siphon b e ()+fromConsumer' consumer = fromFold'_$ \producer -> runEffect $ producer >-> fmap absurd consumer + fromConsumerM :: MonadIO m => (m () -> IO (Either e a)) - -> Consumer b m r + -> Consumer b m () -> Siphon b e a-fromConsumerM whittle consumer = siphon $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> mute consumer +fromConsumerM whittle consumer = siphon $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> consumer -fromSafeConsumer :: Consumer b (SafeT IO) r -> Siphon b e ()-fromSafeConsumer = fromConsumerM (fmap pure . runSafeT)+fromConsumerM' :: MonadIO m + => (forall r. m r -> IO (Either e (a,r))) + -> Consumer b m Void+ -> Siphon b e a+fromConsumerM' whittle consumer = siphon' $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> fmap absurd consumer -fromFallibleConsumer :: Consumer b (ExceptT e IO) r -> Siphon b e ()-fromFallibleConsumer = fromConsumerM runExceptT+fromSafeConsumer :: Consumer b (SafeT IO) Void -> Siphon b e ()+fromSafeConsumer = fromConsumerM' (fmap (\r -> Right ((),r)) . runSafeT) +fromFallibleConsumer :: Consumer b (ExceptT e IO) Void -> Siphon b e ()+fromFallibleConsumer = fromConsumerM' (fmap (fmap (\r -> ((), r))) . runExceptT)+ {-| Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'. -} fromParser :: Parser b IO (Either e a) -> Siphon b e a -fromParser parser = siphon $ Pipes.Parse.evalStateT parser -+fromParser parser = siphon' $ \producer -> drainage $ Pipes.Parse.runStateT parser producer+ where+ drainage m = do + (a,leftovers) <- m+ r <- runEffect (leftovers >-> P.drain)+ case a of+ Left e -> return (Left e)+ Right a' -> return (Right (a',r)) {-| Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'.@@ -760,8 +464,8 @@ return (a,r) {-|- Constructs a 'Siphon' that aborts the computation if the underlying-'Producer' produces anything.+ Constructs a 'Siphon' that aborts the computation with an explicit error+ if the underlying 'Producer' produces anything. -} unwanted :: a -> Siphon b b a unwanted a = siphon' $ \producer -> do@@ -771,6 +475,54 @@ Right (b,_) -> Left b {-|+ Like 'unwanted', but throws an exception instead of using the explicit+ error type.+-}+unwantedX :: Exception ex => (b -> ex) -> a -> Siphon b e a+unwantedX f a = siphon' $ \producer -> do+ n <- next producer + case n of + Left r -> return $ Right (a,r)+ Right (b,_) -> throwIO (f b)++{-|+ Exception that carries a message and a sample of the leftover data. +-}+data LeftoverException b = LeftoverException String b deriving (Typeable)++instance (Typeable b) => Exception (LeftoverException b)++instance (Typeable b) => Show (LeftoverException b) where+ show (LeftoverException msg _) = + "[Leftovers of type " ++ typeName (Proxy::Data.Typeable.Proxy b) ++ "]" ++ msg'+ where+ typeName p = showsTypeRep (typeRep p) []+ msg' = case msg of+ [] -> []+ _ -> " " ++ msg++{-|+ Throws 'LeftoverException' if any data comes out of the underlying+ producer, and returns 'id' otherwise.+-}+leftoverX :: String + -- ^ Error message+ -> Siphon ByteString e (a -> a)+leftoverX msg = unwantedX (LeftoverException msg') id+ where + msg' = "leftoverX." ++ case msg of+ "" -> ""+ _ -> " " ++ msg++{-|+ Like 'leftoverX', but doesn't take an error message.+-}+_leftoverX :: Siphon ByteString e (a -> a)+_leftoverX = unwantedX (LeftoverException msg) id+ where + msg = "_leftoverX."++{-| See the section /Non-lens decoding functions/ in the documentation for the @pipes-text@ package. -}@@ -783,9 +535,10 @@ encoded :: DecodingFunction bytes text -- ^ A decoding function. -> Siphon bytes e (a -> b)- -- ^ A 'Siphon' that determines how to handle leftovers. - -- Pass @pure id@ to ignore leftovers. Pass- -- @unwanted id@ to abort the computation if leftovers remain.+ -- ^ A 'Siphon' that determines how to handle decoding leftovers.+ -- Pass @pure id@ to ignore leftovers. Pass @unwanted id@ to abort+ -- the computation with an explicit error if leftovers remain. Pass+ -- '_leftoverX' to throw a 'LeftoverException' if leftovers remain. -> Siphon text e a -> Siphon bytes e b encoded decoder (Siphon (unLift -> policy)) (Siphon (unLift -> activity)) = @@ -797,23 +550,65 @@ pure (f a,r) -{-|- A configuration parameter used in functions that combine lines of text from- multiple streams.- -}+newtype SiphonOp e a b = SiphonOp { getSiphonOp :: Siphon b e a } -data Lines e = Lines - {- teardown :: (forall r. Producer T.Text IO r -> Producer T.Text IO r)- -> (FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) - -> Producer ByteString IO () -> IO (Either e ())- , lineTweaker :: forall r. Producer T.Text IO r -> Producer T.Text IO r- } +-- | '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 --- | '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+-- | 'divide' builds a 'SiphonOp' for a composite out of the 'SiphonOp's+-- for the parts.+instance Monoid a => Divisible (SiphonOp e a) where+ divide divider siphonOp1 siphonOp2 = contramap divider . SiphonOp $ + (getSiphonOp (contramap fst siphonOp1)) + `mappend`+ (getSiphonOp (contramap snd siphonOp2))+ conquer = SiphonOp (pure mempty) +-- | 'choose' builds a 'SiphonOp' for a sum out of the 'SiphonOp's+-- for the branches.+instance Monoid a => Decidable (SiphonOp e a) where+ choose chooser (SiphonOp s1) (SiphonOp s2) = + contramap chooser . SiphonOp $ + (contraPipeMapL s1) + `mappend`+ (contraPipeMapR s2)+ where+ contraPipeMapL (Siphon s) = Siphon $ case s of+ Pure p -> Pure p+ Other o -> Other $ case o of+ Exhaustive e -> Exhaustive $ \producer ->+ e $ producer >-> allowLefts+ Nonexhaustive ne -> Nonexhaustive $ \producer ->+ ne $ producer >-> allowLefts+ contraPipeMapR (Siphon s) = Siphon $ case s of+ Pure p -> Pure p+ Other o -> Other $ case o of+ Exhaustive e -> Exhaustive $ \producer ->+ e $ producer >-> allowRights+ Nonexhaustive ne -> Nonexhaustive $ \producer ->+ ne $ producer >-> allowRights+ allowLefts = do+ e <- await+ case e of + Left l -> Pipes.yield l >> allowLefts+ Right _ -> allowLefts+ allowRights = do+ e <- await+ case e of + Right r -> Pipes.yield r >> allowRights+ Left _ -> allowRights+ lose f = SiphonOp . Siphon . Other . Nonexhaustive $ \producer -> do+ n <- next producer + return $ case n of + Left () -> Right mempty+ Right (b,_) -> Right (absurd (f b)) {-| Specifies a transformation that will be applied to each line of text,@@ -836,26 +631,27 @@ -} toLines :: DecodingFunction ByteString Text -- ^ A decoding function for lines of text.- -> Siphon ByteString e ()- -- ^ 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.+ -> Siphon ByteString e (() -> ())+ -- ^ A 'Siphon' that determines how to handle decoding leftovers.+ -- Pass @pure id@ to ignore leftovers. Pass @unwanted id@ to abort+ -- the computation with an explicit error if leftovers remain. Pass+ -- '_leftoverX' to throw a 'LeftoverException' if leftovers remain. -> Lines e toLines decoder lopo = Lines- (\tweaker teardown producer -> do+ (\tweaker tear producer -> do let freeLines = transFreeT tweaker . viewLines . decoder $ producer viewLines = getConst . T.lines Const- teardown freeLines >>= runSiphon lopo)+ tear freeLines >>= runSiphonDumb (fmap ($()) lopo)) id {-|- A simplified version of 'executePipelineFallibly' for when the error type is `Void`.- Note however that this function may still throw exceptions.+ A simplified version of 'executePipelineFallibly' for when the error type+ unified with `Void`. Note however that this function may still throw+ exceptions. -} executePipeline :: Piping Void a -> Tree (Stage Void) -> IO a executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline@@ -883,18 +679,18 @@ -- its parent in the tree. -> 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 + PPNone _ -> blende ecpol <$> executeFallibly policy cp + PPOutput _ -> blende ecpol <$> executeFallibly policy cp PPError action -> do- (eoutbox, einbox, eseal) <- spawn' Single+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action $ fromInput einbox) <* (Conceit $ blende ecpol <$> executeFallibly (pipee (errf lpol)) cp `finally` atomically eseal) PPOutputError action -> do - (outbox, inbox, seal) <- spawn' Single- (eoutbox, einbox, eseal) <- spawn' Single+ (outbox, inbox, seal) <- spawn' (bounded 1)+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action $ (fromInput inbox,fromInput einbox))@@ -903,11 +699,11 @@ (pipeoe (fromConsumer.toOutput $ outbox) (errf lpol)) cp `finally` atomically seal `finally` atomically eseal )- PPInput action -> blende ecpol <$> executeFallibly policy cp- PPInputOutput action -> blende ecpol <$> executeFallibly policy cp+ PPInput _ -> blende ecpol <$> executeFallibly policy cp+ PPInputOutput _ -> blende ecpol <$> executeFallibly policy cp PPInputError action -> do- (outbox, inbox, seal) <- spawn' Single- (eoutbox, einbox, eseal) <- spawn' Single+ (outbox, inbox, seal) <- spawn' (bounded 1)+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))@@ -917,9 +713,9 @@ `finally` atomically seal `finally` atomically eseal ) PPInputOutputError action -> do- (ioutbox, iinbox, iseal) <- spawn' Single- (ooutbox, oinbox, oseal) <- spawn' Single- (eoutbox, einbox, eseal) <- spawn' Single+ (ioutbox, iinbox, iseal) <- spawn' (bounded 1)+ (ooutbox, oinbox, oseal) <- spawn' (bounded 1)+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))@@ -943,7 +739,7 @@ (\i _ -> mute $ pipei i) pipeline PPOutput action -> do- (outbox, inbox, seal) <- spawn' Single+ (outbox, inbox, seal) <- spawn' (bounded 1) runConceit $ (Conceit $ action $ fromInput inbox) <* @@ -956,7 +752,7 @@ `finally` atomically seal ) PPError action -> do- (eoutbox, einbox, eseal) <- spawn' Single+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action $ fromInput einbox)@@ -969,8 +765,8 @@ pipeline `finally` atomically eseal) PPOutputError action -> do- (outbox, inbox, seal) <- spawn' Single- (eoutbox, einbox, eseal) <- spawn' Single+ (outbox, inbox, seal) <- spawn' (bounded 1)+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action $ (fromInput inbox,fromInput einbox))@@ -984,7 +780,7 @@ `finally` atomically seal `finally` atomically eseal ) PPInput action -> do- (outbox, inbox, seal) <- spawn' Single+ (outbox, inbox, seal) <- spawn' (bounded 1) runConceit $ (Conceit $ action (toOutput outbox,atomically seal)) <* @@ -997,8 +793,8 @@ `finally` atomically seal ) PPInputOutput action -> do- (ioutbox, iinbox, iseal) <- spawn' Single- (ooutbox, oinbox, oseal) <- spawn' Single+ (ioutbox, iinbox, iseal) <- spawn' (bounded 1)+ (ooutbox, oinbox, oseal) <- spawn' (bounded 1) runConceit $ (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox)) <* @@ -1011,8 +807,8 @@ `finally` atomically iseal `finally` atomically oseal ) PPInputError action -> do- (outbox, inbox, seal) <- spawn' Single- (eoutbox, einbox, eseal) <- spawn' Single+ (outbox, inbox, seal) <- spawn' (bounded 1)+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))@@ -1026,9 +822,9 @@ `finally` atomically seal `finally` atomically eseal ) PPInputOutputError action -> do- (ioutbox, iinbox, iseal) <- spawn' Single- (ooutbox, oinbox, oseal) <- spawn' Single- (eoutbox, einbox, eseal) <- spawn' Single+ (ioutbox, iinbox, iseal) <- spawn' (bounded 1)+ (ooutbox, oinbox, oseal) <- spawn' (bounded 1)+ (eoutbox, einbox, eseal) <- spawn' (bounded 1) errf <- errorSiphonUTF8 <$> newMVar eoutbox runConceit $ (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))@@ -1055,21 +851,7 @@ mute :: Functor f => f a -> f () mute = fmap (const ()) -{-|- An individual stage in a process pipeline. - -}-data Stage e = Stage - {- processDefinition' :: CreateProcess - , stderrLines' :: Lines e- , exitCodePolicy' :: ExitCode -> Either e ()- , inbound' :: forall r. Producer ByteString IO r - -> Producer ByteString (ExceptT e IO) r - } -instance Functor (Stage) where- fmap f (Stage a b c d) = Stage a (fmap f b) (bimap f id . c) (hoist (mapExceptT $ liftM (bimap f id)) . d)- {-| Builds a 'Stage'. -}@@ -1103,14 +885,14 @@ 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+ runTree _ppend _ppend' (Node (Stage _cp _lpol _ecpol pipe) forest) = case forest of [] -> siphon $ \producer ->- blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ pipe producer) lpol) cp+ blende _ecpol <$> executeFallibly (_ppend (fromFallibleProducer $ pipe producer) _lpol) _cp c1 : cs -> siphon $ \producer ->- blende ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ pipe producer) (runNonEmpty ppend ppend' (c1 :| cs)) lpol) cp+ blende _ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ pipe producer) (runNonEmpty _ppend _ppend' (c1 :| cs)) _lpol) _cp - runNonEmpty ppend ppend' (b :| bs) = - runTree ppend ppend' b <* Prelude.foldr (<*) (pure ()) (runTree ppend' ppend' <$> bs) + runNonEmpty _ppend _ppend' (b :| bs) = + runTree _ppend _ppend' b <* Prelude.foldr (<*) (pure ()) (runTree _ppend' _ppend' <$> bs) blende :: (ExitCode -> Either e ()) -> Either e (ExitCode,a) -> Either e a blende f r = r >>= \(ec,a) -> f ec *> pure a@@ -1129,19 +911,3 @@ "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/Extended.hs view
@@ -0,0 +1,131 @@++{-|+-}++{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module System.Process.Streaming.Extended ( + Piap+ , piapi+ , piapo+ , piape+ , piapoe+ , samei+ , sameo+ , samee+ , sameioe+ , toPiping+ , pumpFromHandle + , siphonToHandle + , module System.Process.Streaming+ ) where++import Data.Text +import Control.Applicative+import Control.Exception+import Control.Concurrent.Conceit+import Pipes.ByteString+import System.IO++import System.Process.Streaming+import System.Process.Streaming.Internal++toPiping :: Piap e a -> Piping e a +toPiping (Piap f) = PPInputOutputError f++{-|+ Do stuff with @stdout@.+-}+piapo :: Siphon ByteString e a -> Piap e a+piapo s = Piap $ \(consumer, cleanup, producer1, producer2) -> do+ let nullInput = runPump (pure ()) consumer `finally` cleanup+ drainOutput = runSiphonDumb s producer1 + drainError = runSiphonDumb (pure ()) producer2+ runConceit $ + (\_ r _ -> r)+ <$>+ Conceit nullInput+ <*>+ Conceit drainOutput+ <*>+ Conceit drainError++{-|+ Do stuff with @stderr@.+-}+piape :: Siphon ByteString e a -> Piap e a+piape s = Piap $ \(consumer, cleanup, producer1, producer2) -> do+ let nullInput = runPump (pure ()) consumer `finally` cleanup+ drainOutput = runSiphonDumb (pure ()) producer1 + drainError = runSiphonDumb s producer2+ runConceit $ + (\_ _ r -> r)+ <$>+ Conceit nullInput+ <*>+ Conceit drainOutput+ <*>+ Conceit drainError++{-|+ Do stuff with @stdin@.+-}+piapi :: Pump ByteString e a -> Piap e a+piapi p = Piap $ \(consumer, cleanup, producer1, producer2) -> do+ let nullInput = runPump p consumer `finally` cleanup+ drainOutput = runSiphonDumb (pure ()) producer1 + drainError = runSiphonDumb (pure ()) producer2+ runConceit $ + (\r _ _ -> r)+ <$>+ Conceit nullInput+ <*>+ Conceit drainOutput+ <*>+ Conceit drainError++{-|+ Do stuff with @stdout@ and @stderr@ combined as 'Text'. +-}+piapoe :: Lines e -> Lines e -> Siphon Text e a -> Piap e a+piapoe policy1 policy2 s = Piap $ \(consumer, cleanup, producer1, producer2) -> do+ let nullInput = runPump (pure ()) consumer `finally` cleanup+ combination = combined policy1 policy2 (runSiphonDumb s) producer1 producer2 + runConceit $ + (\_ r -> r)+ <$>+ Conceit nullInput+ <*>+ Conceit combination++{-|+ Pipe @stdin@ to the created process' @stdin@.+-}+samei :: Piap e ()+samei = piapi $ pumpFromHandle System.IO.stdin++{-|+ Pipe the created process' @stdout@ to @stdout@.+-}+sameo :: Piap e ()+sameo = piapo $ siphonToHandle System.IO.stdout++{-|+ Pipe the created process' @stderr@ to @stderr@.+-}+samee :: Piap e ()+samee = piape $ siphonToHandle System.IO.stderr++sameioe :: Piap e ()+sameioe = samei *> sameo *> samee++pumpFromHandle :: Handle -> Pump ByteString e ()+pumpFromHandle = fromProducer . fromHandle++siphonToHandle :: Handle -> Siphon ByteString e ()+siphonToHandle = fromConsumer . toHandle
+ src/System/Process/Streaming/Internal.hs view
@@ -0,0 +1,397 @@+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ViewPatterns #-}++module System.Process.Streaming.Internal ( + Piping(..), + Piap(..), + Pump(..),+ Siphon(..),+ runSiphon,+ runSiphonDumb,+ Siphon_(..),+ exhaustive,+ Lines(..),+ combined,+ manyCombined,+ Stage(..)+ ) where++import Data.Bifunctor+import Data.Monoid+import Data.Text +import Control.Applicative+import Control.Applicative.Lift+import Control.Monad+import Control.Monad.Trans.Free hiding (Pure)+import Control.Monad.Trans.Except+import Control.Exception+import Control.Concurrent+import Control.Concurrent.Conceit+import Pipes+import qualified Pipes as P+import qualified Pipes.Prelude as P+import Pipes.ByteString+import qualified Pipes.Text as T+import Pipes.Concurrent+import System.Process+import System.Exit++{-|+ A 'Piping' determines what standard streams will be piped and what to+do with them.++ The user doesn't need to manually set the 'std_in', 'std_out' and 'std_err'+fields of the 'CreateProcess' record to 'CreatePipe', this is done+automatically. ++ A 'Piping' is parametrized by the type @e@ of errors that can abort+the processing of the streams.+ -}+-- Knows that there is a stdin, stdout and a stderr,+-- but doesn't know anything about file handlers or CreateProcess.+data Piping e a = + PPNone a+ | PPOutput + (Producer ByteString IO () + -> + IO (Either e a))+ | PPError + (Producer ByteString IO () + -> + IO (Either e a))+ | PPOutputError + ((Producer ByteString IO ()+ ,Producer ByteString IO ()) + -> + IO (Either e a))+ | PPInput + ((Consumer ByteString IO ()+ ,IO ()) + -> + IO (Either e a))+ | PPInputOutput + ((Consumer ByteString IO ()+ ,IO ()+ ,Producer ByteString IO ()) + -> + IO (Either e a))+ | PPInputError + ((Consumer ByteString IO ()+ ,IO () + ,Producer ByteString IO ()) + -> + IO (Either e a))+ | PPInputOutputError + ((Consumer ByteString IO ()+ ,IO ()+ ,Producer ByteString IO ()+ ,Producer ByteString IO ()) + -> + IO (Either e a))+ deriving (Functor)+++{-| + 'first' is useful to massage errors.+-}+instance Bifunctor Piping where+ bimap f g pp = case pp of+ PPNone a -> PPNone $ + g a + PPOutput action -> PPOutput $ + fmap (fmap (bimap f g)) action+ PPError action -> PPError $ + fmap (fmap (bimap f g)) action+ PPOutputError action -> PPOutputError $ + fmap (fmap (bimap f g)) action+ PPInput action -> PPInput $ + fmap (fmap (bimap f g)) action+ PPInputOutput action -> PPInputOutput $ + fmap (fmap (bimap f g)) action+ PPInputError action -> PPInputError $ + fmap (fmap (bimap f g)) action+ PPInputOutputError action -> PPInputOutputError $ + fmap (fmap (bimap f g)) action+++newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor++{-| + 'first' is useful to massage errors.+-}+instance Bifunctor (Pump b) where+ bimap f g (Pump x) = Pump $ fmap (liftM (bimap f g)) x++instance Applicative (Pump b e) where+ pure = Pump . pure . pure . pure+ Pump fs <*> Pump as = + Pump $ \consumer -> do+ (outbox1,inbox1,seal1) <- spawn' (bounded 1)+ (outbox2,inbox2,seal2) <- spawn' (bounded 1)+ runConceit $ + Conceit (runExceptT $ do+ r1 <- ExceptT $ (fs $ toOutput outbox1) + `finally` atomically seal1+ r2 <- ExceptT $ (as $ toOutput outbox2) + `finally` atomically seal2+ return $ r1 r2 + )+ <* + Conceit (do+ (runEffect $+ (fromInput inbox1 >> fromInput inbox2) >-> consumer)+ `finally` atomically seal1+ `finally` atomically seal2+ runExceptT $ pure ()+ )++instance (Monoid a) => Monoid (Pump b e a) where+ mempty = Pump . pure . pure . pure $ mempty+ mappend s1 s2 = (<>) <$> s1 <*> s2++-- instance IsString b => IsString (Pump b e ()) where +-- fromString = fromProducer . P.yield . fromString ++{-|+ An alternative to `Piping` for defining what to do with the+ standard streams. 'Piap' is an instance of 'Applicative', unlike+ 'Piping'. ++ With 'Piap', the standard streams are always piped. The values of+ @std_in@, @std_out@ and @std_err@ in the 'CreateProcess' record are+ ignored.+ -}+newtype Piap e a = Piap { runPiap :: (Consumer ByteString IO (), IO (), Producer ByteString IO (), Producer ByteString IO ()) -> IO (Either e a) } deriving (Functor)++instance Bifunctor Piap where+ bimap f g (Piap action) = Piap $ fmap (fmap (bimap f g)) action+++{-| + 'pure' creates a 'Piap' that writes nothing to @stdin@ and drains+ @stdout@ and @stderr@, discarding the data.++ '<*>' schedules the writes to @stdin@ sequentially, and the reads from+ @stdout@ and @stderr@ concurrently.+-}+instance Applicative (Piap e) where+ pure a = Piap $ \(consumer, cleanup, producer1, producer2) -> do+ let nullInput = runPump (pure ()) consumer `finally` cleanup+ drainOutput = runSiphonDumb (pure ()) producer1+ drainError = runSiphonDumb (pure ()) producer2+ runConceit $ + (\_ _ _ -> a)+ <$>+ Conceit nullInput+ <*>+ Conceit drainOutput+ <*>+ Conceit drainError++ (Piap fa) <*> (Piap fb) = Piap $ \(consumer, cleanup, producer1, producer2) -> do+ latch <- newEmptyMVar :: IO (MVar ())+ (ioutbox, iinbox, iseal) <- spawn' (bounded 1)+ (ooutbox, oinbox, oseal) <- spawn' (bounded 1)+ (eoutbox, einbox, eseal) <- spawn' (bounded 1)+ (ioutbox2, iinbox2, iseal2) <- spawn' (bounded 1)+ (ooutbox2, oinbox2, oseal2) <- spawn' (bounded 1)+ (eoutbox2, einbox2, eseal2) <- spawn' (bounded 1)+ let reroutei = runEffect (fromInput (iinbox <> iinbox2) >-> consumer)+ `finally` atomically iseal + `finally` atomically iseal2+ `finally` cleanup+ rerouteo = runEffect (producer1 >-> toOutput (ooutbox <> ooutbox2))+ `finally` atomically oseal + `finally` atomically oseal2+ reroutee = runEffect (producer2 >-> toOutput (eoutbox <> eoutbox2))+ `finally` atomically eseal + `finally` atomically eseal2+ deceivedf = fa + (toOutput ioutbox, + atomically iseal `finally` putMVar latch (), + fromInput oinbox, + fromInput einbox)+ `finally` atomically iseal + `finally` atomically oseal + `finally` atomically eseal + deceivedx = fb + (liftIO (takeMVar latch) *> toOutput ioutbox2, + atomically iseal2, + fromInput oinbox2, + fromInput einbox2)+ `finally` atomically iseal2 + `finally` atomically oseal2 + `finally` atomically eseal2 + runConceit $+ _Conceit reroutei + *>+ _Conceit rerouteo + *> + _Conceit reroutee + *> + (Conceit deceivedf <*> Conceit deceivedx)+++{-| + A 'Siphon' represents a computation that completely drains a producer, but+may fail early with an error of type @e@. ++ -}+newtype Siphon b e a = Siphon (Lift (Siphon_ b e) a) deriving (Functor)+++{-| + 'pure' creates a 'Siphon' that does nothing besides draining the+'Producer'. ++ '<*>' executes its arguments concurrently. The 'Producer' is forked so+ that each argument receives its own copy of the data.+-}+instance Applicative (Siphon b e) where+ pure a = Siphon (pure a)+ (Siphon fa) <*> (Siphon a) = Siphon (fa <*> a)++data Siphon_ b e a = + Exhaustive (forall r. Producer b IO r -> IO (Either e (a,r)))+ | Nonexhaustive (Producer b IO () -> IO (Either e a))+ deriving (Functor)+++instance Applicative (Siphon_ b e) where+ pure a = Exhaustive $ \producer -> do+ r <- runEffect (producer >-> P.drain)+ pure (pure (a,r))+ s1 <*> s2 = bifurcate (nonexhaustive s1) (nonexhaustive s2) + where + bifurcate fs as = Exhaustive $ \producer -> do+ (outbox1,inbox1,seal1) <- spawn' (bounded 1)+ (outbox2,inbox2,seal2) <- spawn' (bounded 1)+ runConceit $+ (,)+ <$>+ Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) + `finally` atomically seal1) + ((as $ fromInput inbox2) + `finally` atomically seal2) + )+ <*>+ _Conceit ((runEffect $ + producer >-> P.tee (toOutput outbox1 >> P.drain) + >-> (toOutput outbox2 >> P.drain)) + `finally` atomically seal1 `finally` atomically seal2+ ) ++nonexhaustive :: Siphon_ b e a -> Producer b IO () -> IO (Either e a)+nonexhaustive (Exhaustive e) = \producer -> liftM (fmap fst) (e producer)+nonexhaustive (Nonexhaustive u) = u++exhaustive :: Siphon_ b e a -> Producer b IO r -> IO (Either e (a,r))+exhaustive s = case s of + Exhaustive e -> e+ Nonexhaustive activity -> \producer -> do + (outbox,inbox,seal) <- spawn' (bounded 1)+ runConceit $ + (,) + <$>+ Conceit (activity (fromInput inbox) `finally` atomically seal)+ <*>+ _Conceit (runEffect (producer >-> (toOutput outbox >> P.drain)) + `finally` atomically seal+ )++runSiphon :: Siphon b e a -> Producer b IO r -> IO (Either e (a,r))+runSiphon (Siphon (unLift -> s)) = exhaustive s++runSiphonDumb :: Siphon b e a -> Producer b IO () -> IO (Either e a)+runSiphonDumb (Siphon (unLift -> s)) = nonexhaustive $ case s of + Exhaustive _ -> s+ Nonexhaustive _ -> Exhaustive (exhaustive s)+++instance Bifunctor (Siphon_ b) where+ bimap f g s = case s of+ Exhaustive u -> Exhaustive $ fmap (liftM (bimap f (bimap g id))) u+ Nonexhaustive h -> Nonexhaustive $ fmap (liftM (bimap f g)) h++{-| + 'first' is useful to massage errors.+-}+instance Bifunctor (Siphon b) where+ bimap f g (Siphon s) = Siphon $ case s of+ Pure a -> Pure (g a)+ Other o -> Other (bimap f g o)++instance (Monoid a) => Monoid (Siphon b e a) where+ mempty = pure mempty+ mappend s1 s2 = (<>) <$> s1 <*> s2++++-- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here+combined :: Lines e + -> Lines e + -> (Producer T.Text IO () -> IO (Either e a))+ -> Producer ByteString IO () + -> Producer ByteString IO () + -> IO (Either e a)+combined (Lines fun1 twk1) (Lines fun2 twk2) combinedConsumer prod1 prod2 = + manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer ++manyCombined :: [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())]+ -> (Producer T.Text IO () -> IO (Either e a))+ -> IO (Either e a) +manyCombined actions consumer = do+ (outbox, inbox, seal) <- spawn' (bounded 1)+ mVar <- newMVar outbox+ runConceit $ + Conceit (mapConceit ($ iterTLines mVar) actions `finally` atomically seal)+ *>+ Conceit (consumer (fromInput inbox) `finally` atomically seal)+ where + iterTLines mvar = iterT $ \textProducer -> do+ -- the P.drain bit was difficult to figure out!!!+ join $ withMVar mvar $ \output -> do+ runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain)++{-|+ A configuration parameter used in functions that combine lines of text from+ multiple streams.+ -}++data Lines e = Lines + {+ teardown :: (forall r. Producer T.Text IO r -> Producer T.Text IO r)+ -> (FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) + -> Producer ByteString IO () -> IO (Either e ())+ , lineTweaker :: forall r. Producer T.Text IO r -> Producer T.Text IO r+ } ++-- | 'fmap' maps over the encoding error. +instance Functor Lines where+ fmap f (Lines func lt) = Lines (\x y z -> fmap (bimap f id) $ func x y z) lt++++{-|+ An individual stage in a process pipeline. + -}+data Stage e = Stage + {+ _processDefinition :: CreateProcess + , _stderrLines :: Lines e+ , _exitCodePolicy :: ExitCode -> Either e ()+ , _inbound :: forall r. Producer ByteString IO r + -> Producer ByteString (ExceptT e IO) r + } ++instance Functor (Stage) where+ fmap f (Stage a b c d) = Stage a (fmap f b) (bimap f id . c) (hoist (mapExceptT $ liftM (bimap f id)) . d)++++
src/System/Process/Streaming/Tutorial.hs view
@@ -85,7 +85,6 @@ >>> 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@@ -99,15 +98,20 @@ {- $collstdout -This example uses the 'toLazyM' fold from @pipes-bytestring@.+We can do this with the 'toLazyM' fold from @pipes-bytestring@: ->>> execute (pipeo (fromFold B.toLazyM)) (shell "echo ooo")+>>> execute (pipeo intoLazyBytes) (shell "echo ooo") (ExitSuccess,"ooo\n") +But the 'intoLazyBytes' function is easier to use:++>>> execute (pipeo intoLazyBytes) (shell "echo ooo")+(ExitSuccess,"ooo\n")+ 'Siphon's are functors, so if we wanted to collect the output as a strict 'ByteString', we could do ->>> execute (pipeo (BL.toStrict <$> fromFold B.toLazyM)) (shell "echo ooo")+>>> execute (pipeo (BL.toStrict <$> intoLazyBytes)) (shell "echo ooo") (ExitSuccess,"ooo\n") Of course, collecting the output in this way breaks streaming. But this is OK@@ -120,7 +124,7 @@ We can use 'pipeoe' collect @stdout@ and @stderr@ concurrently: ->>> execute (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) (shell "{ echo ooo ; echo eee 1>&2 ; }")+>>> execute (pipeoe intoLazyBytes intoLazyBytes) (shell "{ echo ooo ; echo eee 1>&2 ; }") (ExitSuccess,("ooo\n","eee\n")) -}@@ -129,17 +133,17 @@ {- $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.+function. 'encoded' takes as parameters a decoding function (here+'decodeUtf8') and a 'Siphon' that specifies how to handle the leftovers. It+returns a function that converts a 'Siphon' for text into a 'Siphon' for+bytes. In the example we pass @pure id@ as the leftover-handling 'Siphon'. This means "drain all the undecoded data remaining in the stream and return-unchanged the result of @(fromFold T.toLazyM)@". In other words: ignore any+unchanged the result of 'intoLazyText'". In other words: ignore any leftovers. ->>> execute (pipeo (encoded T.decodeUtf8 (pure id) (fromFold T.toLazyM))) (shell "echo ooo")+>>> execute (pipeo (encoded decodeUtf8 (pure id) intoLazyText)) (shell "echo ooo") (ExitSuccess,"ooo\n") But suppose we want to interrupt the execution of the program when we@@ -150,12 +154,12 @@ 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")+>>> executeFallibly (pipeo (encoded decodeUtf8 (unwanted id) intoLazyText)) (shell "echo ooo") Right (ExitSuccess,"ooo\n") Notice also that we had to switch from 'execute' to 'executeFallibly'. This is because, for the first time in the tutorial, we actually have a need for-the error type. 'execute' only works when the error type is 'Void'.+the error type. 'execute' only works when the error type unified with 'Void'. Beware: even if the error type is 'Void', exceptions can still be thrown. -}@@ -172,9 +176,9 @@ >>> :{ let - lin = toLines T.decodeUtf8 (pure ()) + lin = toLines decodeUtf8 (pure id) program = shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }"- in execute (pipeoec lin lin (fromFold T.toLazyM)) program+ in execute (pipeoec lin lin intoLazyText) program :} (ExitSuccess,"ooo\neee\n") @@ -182,11 +186,11 @@ >>> :{ let - lin = toLines T.decodeUtf8 (pure ()) + lin = toLines decodeUtf8 (pure id) lin_stdout = tweakLines (\p -> P.yield "O" *> p) lin lin_stderr = tweakLines (\p -> P.yield "E" *> p) lin program = shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }"- in execute (pipeoec lin_stdout lin_stderr (fromFold T.toLazyM)) program+ in execute (pipeoec lin_stdout lin_stderr intoLazyText) program :} (ExitSuccess,"Oooo\nEeee\n") @@ -197,7 +201,7 @@ 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")+>>> execute (pipeio (fromLazyBytes "iii") intoLazyBytes) (shell "cat") (ExitSuccess,((),"iii")) -}
tests/test.hs view
@@ -102,8 +102,8 @@ combinedStdoutStderr :: IO (ExitCode,TL.Text) combinedStdoutStderr = execute- (pipeoec (linePolicy T.decodeIso8859_1 (pure ()))- (tweakLines annotate $ linePolicy T.decodeIso8859_1 (pure ())) + (pipeoec (toLines T.decodeIso8859_1 (pure id))+ (tweakLines annotate $ toLines T.decodeIso8859_1 (pure id)) (fromFold T.toLazyM)) (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") where@@ -195,7 +195,7 @@ singletonPipeline = executePipelineFallibly (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) - (pure $ stage (linePolicy T.decodeUtf8 (pure ())) pipefail $ + (pure $ stage (toLines T.decodeUtf8 (pure id)) pipefail $ shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }" ) @@ -211,7 +211,7 @@ basicPipeline = executePipelineFallibly (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") (fromFold B.toLazyM)) - (fmap (stage (linePolicy T.decodeUtf8 (pure ())) pipefail) $ + (fmap (stage (toLines T.decodeUtf8 (pure id)) pipefail) $ Node (shell "grep aaa") [Node (shell "grep ccc") []] ) -------------------------------------------------------------------------------@@ -243,28 +243,28 @@ succStage = P.map (Data.ByteString.map succ) rootStage :: Stage Int - rootStage = stage (linePolicy T.decodeIso8859_1 (pure ())) + rootStage = stage (toLines T.decodeIso8859_1 (pure id)) pipefail (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ; echo ffff 1>&2 ; }") branch1 :: Stage Int - branch1 = stage (linePolicy T.decodeIso8859_1 (pure ())) + branch1 = stage (toLines T.decodeIso8859_1 (pure id)) pipefail (shell "grep ooo") branch2 :: Stage Int - branch2 = stage (linePolicy T.decodeIso8859_1 (pure ())) + branch2 = stage (toLines T.decodeIso8859_1 (pure id)) pipefail (shell "grep xxx") terminalStage1 :: Stage Int terminalStage1 = inbound (\p -> p >-> succStage) $- stage (linePolicy T.decodeIso8859_1 (pure ())) + stage (toLines T.decodeIso8859_1 (pure id)) pipefail (shell "tr -d b") terminalStage2 :: Stage Int terminalStage2 = inbound (\p -> p >-> succStage) $- stage (linePolicy T.decodeIso8859_1 (pure ())) + stage (toLines T.decodeIso8859_1 (pure id)) pipefail (shell $ "cat > " ++ branchingPipelineFile) @@ -307,7 +307,7 @@ (pipeoec lp lp countLines) (proc "tests/alternating.sh" []) where- lp = linePolicy T.decodeIso8859_1 (pure ()) + lp = toLines T.decodeIso8859_1 (pure id) countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines @@ -316,7 +316,7 @@ (pipeoec lp lp $ (,) <$> countLines <*> countLines) (proc "tests/alternating.sh" []) where- lp = linePolicy T.decodeIso8859_1 (pure ()) + lp = toLines T.decodeIso8859_1 (pure id) countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines