process-streaming 0.6.5.0 → 0.6.6.0
raw patch · 3 files changed
+259/−218 lines, 3 files
Files
- CHANGELOG +5/−0
- process-streaming.cabal +1/−1
- src/System/Process/Streaming.hs +253/−217
CHANGELOG view
@@ -1,3 +1,8 @@+0.6.6.0+-------++- Added fromFoldable, fromEnumerable, prefixLines+ 0.6.5.0 -------
process-streaming.cabal view
@@ -1,5 +1,5 @@ name: process-streaming-version: 0.6.5.0+version: 0.6.6.0 license: BSD3 license-file: LICENSE data-files:
src/System/Process/Streaming.hs view
@@ -45,6 +45,8 @@ , fromProducerM , fromSafeProducer , fromFallibleProducer+ , fromFoldable+ , fromEnumerable -- * Siphoning bytes out of stdout/stderr , Siphon , SiphonOp (..)@@ -69,6 +71,7 @@ , Lines , toLines , tweakLines+ , prefixLines -- * Pipelines , executePipeline , executePipelineFallibly@@ -77,9 +80,6 @@ , stage , pipefail , inbound--- -- * Utilities--- -- $utilities--- , surely -- * Re-exports -- $reexports , module System.Process,@@ -221,25 +221,59 @@ -- 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))+ | 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) 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+ 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. @@ -319,54 +353,6 @@ separated outfunc errfunc outprod errprod = conceit (outfunc outprod) (errfunc errprod) -{-|- A configuration parameter used in functions that combine lines 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---{-|- Specifies a transformation that will be applied to each line of text,- represented as a 'Producer'.-- Line prefixes are easy to add using applicative notation:-- > (\x -> yield "prefix: " *> x)--}-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 '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.- -}-toLines :: DecodingFunction ByteString Text - -> Siphon ByteString e ()- -> Lines e -toLines decoder lopo = Lines- (\tweaker teardown producer -> do- let freeLines = transFreeT tweaker - . viewLines - . decoder- $ producer- viewLines = getConst . T.lines Const- teardown freeLines >>= runSiphon lopo)- id - -- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here combined :: Lines e -> Lines e @@ -391,53 +377,7 @@ join $ withMVar mvar $ \output -> do runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain) -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) --fromFallibleProducer :: Producer b (ExceptT e IO) r -> Pump b e ()-fromFallibleProducer producer = Pump $ \consumer -> runExceptT $ runEffect (mute producer >-> hoist lift consumer) --{-| - Useful when we want to plug in a handler that does its work in the 'SafeT'-transformer.- -}-safely :: (MFunctor t, C.MonadMask m, MonadIO m) - => (t (SafeT m) l -> (SafeT m) x) - -> t m l -> m x -safely activity = runSafeT . activity . hoist lift --{-|- See the section /Non-lens decoding functions/ in the documentation for the-@pipes-text@ package. --}-type DecodingFunction bytes text = forall r. Producer bytes IO r -> Producer text IO (Producer bytes IO r)--{-|- Constructs a 'Siphon' that works on encoded values out of a 'Siphon' that-works on decoded values. - - The two first arguments are a decoding function and a 'Siphon' that-determines how to handle leftovers. Pass @pure id@ to ignore leftovers. Pass-@unwanted id@ to abort the computation if leftovers remain.- -}-encoded :: DecodingFunction bytes text- -> Siphon bytes e (a -> b)- -> Siphon text e a - -> Siphon bytes e b-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 instance Bifunctor (Pump b) where@@ -470,6 +410,26 @@ mempty = Pump . pure . pure . pure $ mempty mappend s1 s2 = (<>) <$> s1 <*> s2 ++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 = fromProducerM (fmap pure . runSafeT)++fromFallibleProducer :: Producer b (ExceptT e IO) r -> Pump b e ()+fromFallibleProducer = fromProducerM runExceptT++fromFoldable :: Foldable f => f b -> Pump b e ()+fromFoldable = fromProducer . each++fromEnumerable :: Enumerable t => t IO b -> Pump b e ()+fromEnumerable = fromProducer . every++ {-| A 'Siphon' represents a computation that completely drains a producer, but may fail early with an error of type @e@. @@ -482,6 +442,72 @@ -} newtype Siphon b e a = Siphon (Lift (Siphon_ b e) a) deriving (Functor,Applicative) +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++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.@@ -542,7 +568,6 @@ Left () -> Right mempty Right (b,_) -> Right (absurd (f b)) - allowLefts :: Monad m => Pipe (Either b a) b m r allowLefts = do e <- await@@ -557,104 +582,7 @@ 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 (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--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 --fromFallibleConsumer :: Consumer b (ExceptT e IO) r -> Siphon b e ()-fromFallibleConsumer consumer = siphon $ \producer -> runExceptT $ runEffect (hoist lift producer >-> mute consumer) - {-| - 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 ---{-| - 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@. @@ -665,7 +593,6 @@ -> Siphon b e a siphon f = Siphon (Other (Nonexhaustive f)) - {-| Builds a 'Siphon' out of a computation that drains a 'Producer' completely, but may fail with an error of type @e@.@@ -712,6 +639,36 @@ 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 ++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 = fromConsumerM (fmap pure . runSafeT)++fromFallibleConsumer :: Consumer b (ExceptT e IO) r -> Siphon b e ()+fromFallibleConsumer = fromConsumerM 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 +++{-| + 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)+ {-| Constructs a 'Siphon' that aborts the computation if the underlying 'Producer' produces anything.@@ -723,6 +680,91 @@ Left r -> Right (a,r) Right (b,_) -> Left b +{-|+ See the section /Non-lens decoding functions/ in the documentation for the+@pipes-text@ package. +-}+type DecodingFunction bytes text = forall r. Producer bytes IO r -> Producer text IO (Producer bytes IO r)++{-|+ Constructs a 'Siphon' that works on encoded values out of a 'Siphon' that+works on decoded values. + + The two first arguments are a decoding function and a 'Siphon' that+determines how to handle leftovers. Pass @pure id@ to ignore leftovers. Pass+@unwanted id@ to abort the computation if leftovers remain.+ -}+encoded :: DecodingFunction bytes text+ -> Siphon bytes e (a -> b)+ -> Siphon text e a + -> Siphon bytes e b+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)+++{-|+ A configuration parameter used in functions that combine lines 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+++{-|+ Specifies a transformation that will be applied to each line of text,+ represented as a 'Producer'.++ Line prefixes are easy to add using applicative notation:++ > (\x -> yield "prefix: " *> x)+-}+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) +++{-|+ Specifies a prefix that will be calculated and appeded for each line of+ text.+-}+prefixLines :: IO T.Text -> Lines e -> Lines e +prefixLines tio = tweakLines (\p -> liftIO tio *> p) +++{-|+ 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.+ -}+toLines :: DecodingFunction ByteString Text + -> Siphon ByteString e ()+ -> Lines e +toLines decoder lopo = Lines+ (\tweaker teardown producer -> do+ let freeLines = transFreeT tweaker + . viewLines + . decoder+ $ producer+ viewLines = getConst . T.lines Const+ teardown freeLines >>= runSiphon lopo)+ id ++ executePipeline :: Piping Void a -> Tree (Stage Void) -> IO a executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline @@ -982,12 +1024,6 @@ 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.