process-streaming 0.5.0.2 → 0.6.0.0
raw patch · 8 files changed
+1647/−1516 lines, 8 filesdep +transformers-compatdep −mtldep ~transformerssetup-changed
Dependencies added: transformers-compat
Dependencies removed: mtl
Dependency ranges changed: transformers
Files
- CHANGELOG +45/−38
- LICENSE +27/−27
- README.md +19/−19
- Setup.hs +2/−2
- process-streaming.cabal +86/−86
- src/System/Process/Lens.hs +229/−229
- src/System/Process/Streaming.hs +916/−860
- tests/test.hs +323/−255
CHANGELOG view
@@ -1,38 +1,45 @@-0.5.0.2 -------- - -- A bug slipped by :( - -0.5.0.1 -------- - -- Changes in the internals to reduce the number of spawned threads. - -0.5.0.0 -------- - -- Now the constructors for "PipingPolicy" take "Siphons" directly, instead of - continuation functions. -- Removed "separated" and "combined" functions, added new "PipingPolicy" - constructors in their place. -- Removed "LeftoverPolicy", its function is now performed by Siphons. -- Removed "surely", "safely", "monoidally" functions with confusing signatures. -- Removed all occurrences of unbounded buffers in the code. -- Implemented support for branching pipelines of processes. - -0.3.0.0 -------- - -- Renamed many functions (the number-based naming schema is gone). -- Decisions about how to consume the standard streams were moved to the new -"PipingPolicy" type. -- "ForkProd" is now "Siphon". -- "execute" can now throw IOExceptions (but see "safeExecute"). -- Some complicated types were hidden behind data declarations. -- Functor, Bifunctor and Profunctor instances for some types. - -0.0.1.1 -------- - -- Added some examples. -- Re-exported System.Process from System.Process.Streaming. +0.6.0.0+-------++- Now compatible with earlier versions of transformers.+- Simplified LinePolicy.+- Simplified pipeline functions and datatypes.++0.5.0.2+-------++- A bug slipped by :(++0.5.0.1+-------++- Changes in the internals to reduce the number of spawned threads.++0.5.0.0+-------++- Now the constructors for "PipingPolicy" take "Siphons" directly, instead of+ continuation functions.+- Removed "separated" and "combined" functions, added new "PipingPolicy"+ constructors in their place.+- Removed "LeftoverPolicy", its function is now performed by Siphons.+- Removed "surely", "safely", "monoidally" functions with confusing signatures.+- Removed all occurrences of unbounded buffers in the code.+- Implemented support for branching pipelines of processes.++0.3.0.0+-------++- Renamed many functions (the number-based naming schema is gone).+- Decisions about how to consume the standard streams were moved to the new+"PipingPolicy" type.+- "ForkProd" is now "Siphon". +- "execute" can now throw IOExceptions (but see "safeExecute").+- Some complicated types were hidden behind data declarations.+- Functor, Bifunctor and Profunctor instances for some types.++0.0.1.1+-------++- Added some examples.+- Re-exported System.Process from System.Process.Streaming.
LICENSE view
@@ -1,27 +1,27 @@-Copyright (c) 2014, Daniel Díaz Carrete -All rights reserved. - -Redistribution and use in source and binary forms, with or without modification, -are permitted provided that the following conditions are met: - -* Redistributions of source code must retain the above copyright notice, this - list of conditions and the following disclaimer. - -* Redistributions in binary form must reproduce the above copyright notice, this - list of conditions and the following disclaimer in the documentation and/or - other materials provided with the distribution. - -* Neither the name of the {organization} nor the names of its - contributors may be used to endorse or promote products derived from - this software without specific prior written permission. - -THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND -ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED -WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE -DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR -ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES -(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; -LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON -ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS -SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +Copyright (c) 2014, Daniel Díaz Carrete+All rights reserved.++Redistribution and use in source and binary forms, with or without modification,+are permitted provided that the following conditions are met:++* Redistributions of source code must retain the above copyright notice, this+ list of conditions and the following disclaimer.++* Redistributions in binary form must reproduce the above copyright notice, this+ list of conditions and the following disclaimer in the documentation and/or+ other materials provided with the distribution.++* Neither the name of the {organization} nor the names of its+ contributors may be used to endorse or promote products derived from+ this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
README.md view
@@ -1,19 +1,19 @@-process-streaming -================= - -Exploring how to interact with system processes using a streaming library -(pipes). - -The basic goals: - -- Concurrent, streaming access to stdin, stdout and stderr. - -- Easy integration with regular consumers, parsers from pipes-parse and various - folds. - -- Avoid launching exceptions: use Either or similar solution to signal non-IO - related error conditions. - -Relevant thread in the Haskell Pipes Google Group: - -https://groups.google.com/forum/#!searchin/haskell-pipes/pipes$20process/haskell-pipes/JFfyquj5HAg/Lxz7p50JOh4J +process-streaming+=================++Exploring how to interact with system processes using a streaming library+(pipes).++The basic goals:++- Concurrent, streaming access to stdin, stdout and stderr.++- Easy integration with regular consumers, parsers from pipes-parse and various+ folds.++- Avoid launching exceptions: use Either or similar solution to signal non-IO+ related error conditions.++Relevant thread in the Haskell Pipes Google Group:++https://groups.google.com/forum/#!searchin/haskell-pipes/pipes$20process/haskell-pipes/JFfyquj5HAg/Lxz7p50JOh4J
Setup.hs view
@@ -1,2 +1,2 @@-import Distribution.Simple -main = defaultMain +import Distribution.Simple+main = defaultMain
process-streaming.cabal view
@@ -1,86 +1,86 @@-name: process-streaming -version: 0.5.0.2 -license: BSD3 -license-file: LICENSE -data-files: -author: Daniel Díaz Carrete -maintainer: diaz_carrete@yahoo.com -category: System -build-type: Simple -cabal-version: >= 1.10 -Synopsis: Streaming interface to system processes. -Description: Concurrent, streaming access to the input and outputs of system processes. - -Extra-Source-Files: - README.md - CHANGELOG - -Library - default-language: Haskell2010 - hs-source-dirs: src - exposed-modules: - System.Process.Streaming - System.Process.Lens - other-modules: - build-depends: - base >= 4.4 && < 5, - transformers >= 0.4 && < 0.5, - mtl >= 2.2 && < 2.3, - free >= 4.2 && < 5, - bifunctors >= 4.1 && < 5, - async >= 2.0.1 && < 2.1, - 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, - containers >= 0.4, - semigroups >= 0.15 && < 0.16 - -Test-suite test - default-language: - Haskell2010 - type: - exitcode-stdio-1.0 - hs-source-dirs: - tests - main-is: - test.hs - build-depends: - base >= 4.4 && < 5 - , transformers >= 0.4 && < 0.5 - , mtl >= 2.2 && < 2.3 - , free >= 4.2 && < 5 - , bifunctors >= 4.1 && < 5 - , async >= 2.0.1 && < 2.1 - , 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 - -Source-repository head - type: git - location: https://github.com/danidiaz/process-streaming.git - +name: process-streaming+version: 0.6.0.0+license: BSD3+license-file: LICENSE+data-files: +author: Daniel Díaz Carrete+maintainer: diaz_carrete@yahoo.com+category: System+build-type: Simple+cabal-version: >= 1.10+Synopsis: Streaming interface to system processes. +Description: Concurrent, streaming access to the input and outputs of system processes.++Extra-Source-Files:+ README.md+ CHANGELOG++Library+ default-language: Haskell2010+ hs-source-dirs: src+ exposed-modules: + System.Process.Streaming+ System.Process.Lens+ other-modules: + build-depends: + base >= 4.4 && < 5,+ transformers >= 0.2 && < 0.5,+ transformers-compat == 0.3.*,+ free >= 4.2 && < 5,+ bifunctors >= 4.1 && < 5,+ async >= 2.0.1 && < 2.1,+ 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,+ containers >= 0.4,+ semigroups >= 0.15 && < 0.16++Test-suite test+ default-language:+ Haskell2010+ type:+ exitcode-stdio-1.0+ hs-source-dirs:+ tests+ main-is:+ test.hs+ build-depends:+ base >= 4.4 && < 5+ , transformers >= 0.2 && < 0.5+ , transformers-compat == 0.3.*+ , free >= 4.2 && < 5+ , bifunctors >= 4.1 && < 5+ , async >= 2.0.1 && < 2.1+ , 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++Source-repository head+ type: git+ location: https://github.com/danidiaz/process-streaming.git+
src/System/Process/Lens.hs view
@@ -1,229 +1,229 @@- --- | --- Lenses and traversals for 'CreateProcess' and related types. --- ------------------------------------------------------------------------------ - -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE RankNTypes #-} - -module System.Process.Lens ( - _cmdspec - , _ShellCommand - , _RawCommand - , _cwd - , _env - , streams - , _close_fds - , _create_group - , _delegate_ctlc - , handles - , nohandles - , handleso - , handlese - , handlesoe - , handlesi - , handlesio - , handlesie - , handlesioe - ) where - -import Data.Maybe -import Data.Functor.Identity -import Data.Monoid -import Data.Traversable -import Control.Applicative -import System.IO -import System.Process - -{-| - > _cmdspec :: Lens' CreateProcess CmdSpec --} -_cmdspec :: forall f. Functor f => (CmdSpec -> f CmdSpec) -> CreateProcess -> f CreateProcess -_cmdspec f c = setCmdSpec c <$> f (cmdspec c) - where - setCmdSpec c cmdspec' = c { cmdspec = cmdspec' } - -{-| - > _ShellCommand :: Prism' CmdSpec String --} -_ShellCommand :: forall m. Applicative m => (String -> m String) -> CmdSpec -> m CmdSpec -_ShellCommand f quad = case impure quad of - Left l -> pure l - Right r -> fmap ShellCommand (f r) - where - impure (ShellCommand str) = Right str - impure x = Left x - -{-| - > _RawCommand :: Prism' CmdSpec (FilePath,[String]) --} -_RawCommand :: forall m. Applicative m => ((FilePath,[String]) -> m (FilePath,[String])) -> CmdSpec -> m CmdSpec -_RawCommand f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (RawCommand fpath strs) = Right (fpath,strs) - impure x = Left x - justify (fpath,strs) = RawCommand fpath strs - -{-| - > _cwd :: Lens' CreateProcess (Maybe FilePath) --} -_cwd :: forall f. Functor f => (Maybe FilePath -> f (Maybe FilePath)) -> CreateProcess -> f CreateProcess -_cwd f c = setCwd c <$> f (cwd c) - where - setCwd c cwd' = c { cwd = cwd' } - -{-| - > _env :: Lens' CreateProcess (Maybe [(String,String)]) --} -_env :: forall f. Functor f => (Maybe [(String, String)] -> f (Maybe [(String, String)])) -> CreateProcess -> f CreateProcess -_env f c = setEnv c <$> f (env c) - where - setEnv c env' = c { env = env' } - -{-| - A lens for the @(std_in,std_out,std_err)@ triplet. - - > streams :: Lens' CreateProcess (StdStream,StdStream,StdStream) --} -streams :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess -streams f c = setStreams c <$> f (getStreams c) - where - getStreams c = (std_in c,std_out c, std_err c) - setStreams c (s1,s2,s3) = c { std_in = s1 - , std_out = s2 - , std_err = s3 - } - -_close_fds :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess -_close_fds f c = set_close_fds c <$> f (close_fds c) - where - set_close_fds c cwd' = c { close_fds = cwd' } - - -_create_group :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess -_create_group f c = set_create_group c <$> f (create_group c) - where - set_create_group c cwd' = c { create_group = cwd' } - -_delegate_ctlc :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess -_delegate_ctlc f c = set_delegate_ctlc c <$> f (delegate_ctlc c) - where - set_delegate_ctlc c cwd' = c { delegate_ctlc = cwd' } - -{-| - A 'Lens' for the return value of 'createProcess' that focuses on the handles. - - > handles :: Lens' (Maybe Handle, Maybe Handle, Maybe Handle,ProcessHandle) (Maybe Handle, Maybe Handle, Maybe Handle) - -} -handles :: forall m. Functor m => ((Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) -> m (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) -handles f quad = setHandles quad <$> f (getHandles quad) - where - setHandles (c1'',c2'',c3'',c4'') (c1',c2',c3') = (c1',c2',c3',c4'') - getHandles (c1'',c2'',c3'',c4'') = (c1'',c2'',c3'') - - -{-| - A 'Prism' that matches when none of the standard streams have been piped. - - > nohandles :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) () - -} -nohandles :: forall m. Applicative m => (() -> m ()) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -nohandles f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Nothing, Nothing, Nothing) = Right () - impure x = Left x - justify () = (Nothing, Nothing, Nothing) - - -{-| - A 'Prism' that matches when only @stdin@ has been piped. - - > handlesi :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle) - -} -handlesi :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -handlesi f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Just h1, Nothing, Nothing) = Right h1 - impure x = Left x - justify h1 = (Just h1, Nothing, Nothing) - -handlesio :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -handlesio f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Just h1, Just h2, Nothing) = Right (h1,h2) - impure x = Left x - justify (h1,h2) = (Just h1, Just h2, Nothing) - -handlesie :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -handlesie f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Just h1, Nothing, Just h2) = Right (h1,h2) - impure x = Left x - justify (h1,h2) = (Just h1, Nothing, Just h2) - -{-| - A 'Prism' that matches when all three @stdin@, @stdout@ and @stderr@ have been piped. - - > handlesioe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle, Handle) - -} -handlesioe :: forall m. Applicative m => ((Handle, Handle, Handle) -> m (Handle, Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -handlesioe f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Just h1, Just h2, Just h3) = Right (h1, h2, h3) - impure x = Left x - justify (h1, h2, h3) = (Just h1, Just h2, Just h3) - -{-| - A 'Prism' that matches when only @stdout@ and @stderr@ have been piped. - - > handlesoe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle) - -} -handlesoe :: forall m. Applicative m => ((Handle, Handle) -> m (Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -handlesoe f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Nothing, Just h2, Just h3) = Right (h2, h3) - impure x = Left x - justify (h2, h3) = (Nothing, Just h2, Just h3) - -{-| - A 'Prism' that matches when only @stdout@ has been piped. - - > handleso :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle) - -} -handleso :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -handleso f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Nothing, Just h2, Nothing) = Right h2 - impure x = Left x - justify h2 = (Nothing, Just h2, Nothing) - -{-| - A 'Prism' that matches when only @stderr@ has been piped. - - > handlese :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle) - -} -handlese :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle) -handlese f quad = case impure quad of - Left l -> pure l - Right r -> fmap justify (f r) - where - impure (Nothing, Nothing, Just h2) = Right h2 - impure x = Left x - justify h2 = (Nothing, Nothing, Just h2) ++-- |+-- Lenses and traversals for 'CreateProcess' and related types.+--+-----------------------------------------------------------------------------++{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE RankNTypes #-}++module System.Process.Lens ( + _cmdspec+ , _ShellCommand+ , _RawCommand+ , _cwd+ , _env+ , streams+ , _close_fds+ , _create_group+ , _delegate_ctlc + , handles+ , nohandles+ , handleso+ , handlese+ , handlesoe+ , handlesi+ , handlesio+ , handlesie+ , handlesioe+ ) where++import Data.Maybe+import Data.Functor.Identity+import Data.Monoid+import Data.Traversable+import Control.Applicative+import System.IO+import System.Process++{-|+ > _cmdspec :: Lens' CreateProcess CmdSpec +-}+_cmdspec :: forall f. Functor f => (CmdSpec -> f CmdSpec) -> CreateProcess -> f CreateProcess +_cmdspec f c = setCmdSpec c <$> f (cmdspec c)+ where+ setCmdSpec c cmdspec' = c { cmdspec = cmdspec' } ++{-|+ > _ShellCommand :: Prism' CmdSpec String+-}+_ShellCommand :: forall m. Applicative m => (String -> m String) -> CmdSpec -> m CmdSpec +_ShellCommand f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap ShellCommand (f r)+ where + impure (ShellCommand str) = Right str+ impure x = Left x++{-|+ > _RawCommand :: Prism' CmdSpec (FilePath,[String])+-}+_RawCommand :: forall m. Applicative m => ((FilePath,[String]) -> m (FilePath,[String])) -> CmdSpec -> m CmdSpec +_RawCommand f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (RawCommand fpath strs) = Right (fpath,strs)+ impure x = Left x+ justify (fpath,strs) = RawCommand fpath strs++{-|+ > _cwd :: Lens' CreateProcess (Maybe FilePath)+-}+_cwd :: forall f. Functor f => (Maybe FilePath -> f (Maybe FilePath)) -> CreateProcess -> f CreateProcess +_cwd f c = setCwd c <$> f (cwd c)+ where+ setCwd c cwd' = c { cwd = cwd' } ++{-|+ > _env :: Lens' CreateProcess (Maybe [(String,String)])+-}+_env :: forall f. Functor f => (Maybe [(String, String)] -> f (Maybe [(String, String)])) -> CreateProcess -> f CreateProcess +_env f c = setEnv c <$> f (env c)+ where+ setEnv c env' = c { env = env' } ++{-| + A lens for the @(std_in,std_out,std_err)@ triplet. ++ > streams :: Lens' CreateProcess (StdStream,StdStream,StdStream)+-}+streams :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess +streams f c = setStreams c <$> f (getStreams c)+ where + getStreams c = (std_in c,std_out c, std_err c)+ setStreams c (s1,s2,s3) = c { std_in = s1 + , std_out = s2 + , std_err = s3 + } ++_close_fds :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_close_fds f c = set_close_fds c <$> f (close_fds c)+ where+ set_close_fds c cwd' = c { close_fds = cwd' } +++_create_group :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_create_group f c = set_create_group c <$> f (create_group c)+ where+ set_create_group c cwd' = c { create_group = cwd' } ++_delegate_ctlc :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess +_delegate_ctlc f c = set_delegate_ctlc c <$> f (delegate_ctlc c)+ where+ set_delegate_ctlc c cwd' = c { delegate_ctlc = cwd' } ++{-|+ A 'Lens' for the return value of 'createProcess' that focuses on the handles.++ > handles :: Lens' (Maybe Handle, Maybe Handle, Maybe Handle,ProcessHandle) (Maybe Handle, Maybe Handle, Maybe Handle)+ -}+handles :: forall m. Functor m => ((Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) -> m (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) +handles f quad = setHandles quad <$> f (getHandles quad) + where+ setHandles (c1'',c2'',c3'',c4'') (c1',c2',c3') = (c1',c2',c3',c4'')+ getHandles (c1'',c2'',c3'',c4'') = (c1'',c2'',c3'')+ ++{-|+ A 'Prism' that matches when none of the standard streams have been piped.++ > nohandles :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) ()+ -}+nohandles :: forall m. Applicative m => (() -> m ()) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+nohandles f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Nothing, Nothing, Nothing) = Right () + impure x = Left x+ justify () = (Nothing, Nothing, Nothing) +++{-|+ A 'Prism' that matches when only @stdin@ has been piped.++ > handlesi :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)+ -}+handlesi :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+handlesi f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Just h1, Nothing, Nothing) = Right h1+ impure x = Left x+ justify h1 = (Just h1, Nothing, Nothing) ++handlesio :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+handlesio f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Just h1, Just h2, Nothing) = Right (h1,h2)+ impure x = Left x+ justify (h1,h2) = (Just h1, Just h2, Nothing) ++handlesie :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+handlesie f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Just h1, Nothing, Just h2) = Right (h1,h2)+ impure x = Left x+ justify (h1,h2) = (Just h1, Nothing, Just h2) ++{-|+ A 'Prism' that matches when all three @stdin@, @stdout@ and @stderr@ have been piped.++ > handlesioe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle, Handle)+ -}+handlesioe :: forall m. Applicative m => ((Handle, Handle, Handle) -> m (Handle, Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+handlesioe f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Just h1, Just h2, Just h3) = Right (h1, h2, h3) + impure x = Left x+ justify (h1, h2, h3) = (Just h1, Just h2, Just h3) ++{-|+ A 'Prism' that matches when only @stdout@ and @stderr@ have been piped.++ > handlesoe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle)+ -}+handlesoe :: forall m. Applicative m => ((Handle, Handle) -> m (Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+handlesoe f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Nothing, Just h2, Just h3) = Right (h2, h3) + impure x = Left x+ justify (h2, h3) = (Nothing, Just h2, Just h3) ++{-|+ A 'Prism' that matches when only @stdout@ has been piped.++ > handleso :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)+ -}+handleso :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+handleso f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Nothing, Just h2, Nothing) = Right h2+ impure x = Left x+ justify h2 = (Nothing, Just h2, Nothing) ++{-|+ A 'Prism' that matches when only @stderr@ has been piped.++ > handlese :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)+ -}+handlese :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)+handlese f quad = case impure quad of+ Left l -> pure l+ Right r -> fmap justify (f r)+ where + impure (Nothing, Nothing, Just h2) = Right h2+ impure x = Left x+ justify h2 = (Nothing, Nothing, Just h2)
src/System/Process/Streaming.hs view
@@ -1,860 +1,916 @@- --- | --- This module contains helper functions and types built on top of --- "System.Process" and "Pipes". --- --- They provide concurrent, streaming access to the inputs and outputs of --- system processes. --- --- Error conditions that are not directly related to IO are made explicit --- in the types. --- --- Regular 'Consumer's, 'Parser's from @pipes-parse@ and folds from --- "Pipes.Prelude" (also folds from @pipes-bytestring@ and @pipes-text@) --- can be used to consume the output streams of the external processes. --- ------------------------------------------------------------------------------ - -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE ViewPatterns #-} - -module System.Process.Streaming ( - -- * Execution - execute - , executeFallibly - -- * Piping Policies - , PipingPolicy - , nopiping - , pipeo - , pipee - , pipeoe - , pipeoec - , pipei - , pipeio - , pipeie - , pipeioe - , pipeioec - - -- * Pumping bytes into stdin - , Pump (..) - , fromProducer - , fromSafeProducer - , fromFallibleProducer - -- * Siphoning bytes out of stdout/stderr - , Siphon - , siphon - , siphon' - , fromFold - , fromFold' - , fromFold'_ - , fromConsumer - , fromSafeConsumer - , fromFallibleConsumer - , fromParser - , unwanted - , DecodingFunction - , encoded - -- * Line handling - , LinePolicy - , linePolicy - -- * Pipelines - , executePipeline - , executePipelineFallibly - , CreatePipeline (..) - , simplePipeline - , Stage (..) - , SubsequentStage (..) - -- * Re-exports - -- $reexports - , module System.Process - ) where - -import Data.Maybe -import Data.Bifunctor -import Data.Functor.Identity -import Data.Either -import Data.Monoid -import Data.Foldable -import Data.Traversable -import Data.Typeable -import Data.Tree -import Data.Text -import Data.Text.Encoding -import Data.Void -import Data.List.NonEmpty -import qualified Data.List.NonEmpty as N -import Control.Applicative -import Control.Monad -import Control.Monad.Trans.Free -import Control.Monad.Except -import Control.Monad.State -import Control.Monad.Writer.Strict -import qualified Control.Monad.Catch as C -import Control.Exception -import Control.Concurrent -import Control.Concurrent.Async -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 -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 - -execute :: PipingPolicy 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. - - This fuction re-throws any 'IOException's it encounters. - - If the consumption of the standard streams fails with @e@, the whole -computation is immediately aborted and @e@ is returned. (An exception is not -thrown in this case.). - - If an error @e@ or an exception happens, the external process is -terminated. - -} -executeFallibly :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a)) -executeFallibly pp record = case pp of - PPNone a -> executeInternal record nohandles $ - \() -> (return . Right $ a,return ()) - PPOutput action -> executeInternal (record{std_out = CreatePipe}) handleso $ - \h->(action (fromHandle h),hClose h) - PPError action -> executeInternal (record{std_err = CreatePipe}) handlese $ - \h->(action (fromHandle h),hClose h) - PPOutputError action -> executeInternal (record{std_out = CreatePipe, std_err = CreatePipe}) handlesoe $ - \(hout,herr)->(action (fromHandle hout,fromHandle herr),hClose hout `finally` hClose herr) - PPInput action -> executeInternal (record{std_in = CreatePipe}) handlesi $ - \h -> (action (toHandle h, hClose h), return ()) - PPInputOutput action -> executeInternal (record{std_in = CreatePipe,std_out = CreatePipe}) handlesio $ - \(hin,hout) -> (action (toHandle hin,hClose hin,fromHandle hout), hClose hout) - PPInputError action -> executeInternal (record{std_in = CreatePipe,std_err = CreatePipe}) handlesie $ - \(hin,herr) -> (action (toHandle hin,hClose hin,fromHandle herr), hClose herr) - PPInputOutputError action -> executeInternal (record{std_in = CreatePipe, std_out = CreatePipe, std_err = CreatePipe}) handlesioe $ - \(hin,hout,herr) -> (action (toHandle hin,hClose hin,fromHandle hout,fromHandle herr), hClose hout `finally` hClose herr) - -executeInternal :: CreateProcess -> (forall m. Applicative m => (t -> m t) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (t ->(IO (Either e a),IO ())) -> 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 - Nothing -> - throwIO (userError "stdin/stdout/stderr handle unwantedly null") - `finally` - terminateCarefully phandle - Just t -> - let (action,cleanup) = allocator t in - -- Handles must be closed *after* terminating the process, because a close - -- operation may block if the external process has unflushed bytes in the stream. - (restore (terminateOnError phandle 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 - mExitCode <- getProcessExitCode pHandle - case mExitCode of - Nothing -> terminateProcess pHandle - Just _ -> return () - -terminateOnError :: ProcessHandle - -> IO (Either e a) - -> IO (Either e (ExitCode,a)) -terminateOnError pHandle action = do - result <- action - case result of - Left e -> do - terminateCarefully pHandle - return $ Left e - Right r -> do - exitCode <- waitForProcess pHandle - return $ Right (exitCode,r) - -{-| - A 'PipingPolicy' 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 -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 = - 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) - -instance Bifunctor PipingPolicy 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 :: PipingPolicy e () -nopiping = PPNone () - -{-| - Pipe @stdout@. --} -pipeo :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a -pipeo (runSiphon -> siphonout) = PPOutput $ siphonout - -{-| - Pipe @stderr@. --} -pipee :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a -pipee (runSiphon -> siphonout) = PPError $ siphonout - -{-| - Pipe @stdout@ and @stderr@. --} -pipeoe :: (Show e,Typeable e) => Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (a,b) -pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = - PPOutputError $ uncurry $ separated siphonout siphonerr - -{-| - Pipe @stdout@ and @stderr@ and consume them combined as 'Text'. --} -pipeoec :: (Show e,Typeable e) => LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e a -pipeoec policy1 policy2 (runSiphon -> siphon) = - PPOutputError $ uncurry $ combined policy1 policy2 siphon - -{-| - Pipe @stdin@. --} -pipei :: (Show e, Typeable e) => Pump ByteString e i -> PipingPolicy e i -pipei (Pump feeder) = PPInput $ \(consumer,cleanup) -> feeder consumer `finally` cleanup - -{-| - Pipe @stdin@ and @stdout@. --} -pipeio :: (Show e, Typeable e) - => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a) -pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) -> - (conceit (feeder consumer `finally` cleanup) (siphonout producer)) - -{-| - Pipe @stdin@ and @stderr@. --} -pipeie :: (Show e, Typeable e) - => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy 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 :: (Show e, Typeable e) - => Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (i,a,b) -pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $ - \(consumer,cleanup,outprod,errprod) -> - (conceit (feeder consumer `finally` cleanup) - (separated siphonout siphonerr outprod errprod)) - where - flattenTuple (i, (a, b)) = (i,a,b) - -{-| - Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'. --} -pipeioec :: (Show e, Typeable e) - => Pump ByteString e i -> LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e (i,a) -pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $ - \(consumer,cleanup,outprod,errprod) -> - (conceit (feeder consumer `finally` cleanup) - (combined policy1 policy2 siphon outprod errprod)) - -separated :: (Show e, Typeable e) - => (Producer ByteString IO () -> IO (Either e a)) - -> (Producer ByteString IO () -> IO (Either e b)) - -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e (a,b)) -separated outfunc errfunc outprod errprod = - conceit (outfunc outprod) (errfunc errprod) - -{-| - Defines how to decode a stream of bytes into text, including what to do - in presence of leftovers. Also defines how to manipulate each individual - line of text. - -} -data LinePolicy e = LinePolicy ((FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> Producer ByteString IO () -> IO (Either e ())) - -instance Functor LinePolicy where - fmap f (LinePolicy func) = LinePolicy $ fmap (fmap (fmap (bimap f id))) func - -{-| - Constructs a 'LinePolicy'. - - The second argument is a 'Siphon' value that specifies how to handle -decoding failures. Passing @pure ()@ will ignore any leftovers. Passing -@unwanted ()@ will abort the computation if leftovers remain. - - The third argument is a function that modifies each individual line. - The line is represented as a 'Producer' to avoid having to keep it - wholly in memory. If you want the lines unmodified, just pass @id@. - Line prefixes are easy to add using applicative notation: - - > (\x -> yield "prefix: " *> x) - - -} - -linePolicy :: (Show e, Typeable e) - => DecodingFunction ByteString Text - -> Siphon ByteString e () - -> (forall r. Producer T.Text IO r -> Producer T.Text IO r) - -> LinePolicy e -linePolicy decoder lopo transform = LinePolicy $ \teardown producer -> do - let freeLines = transFreeT transform - . viewLines - . decoder - $ producer - viewLines = getConst . T.lines Const - teardown freeLines >>= runSiphon lopo - --- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here -combined :: (Show e, Typeable e) - => LinePolicy e - -> LinePolicy e - -> (Producer T.Text IO () -> IO (Either e a)) - -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a) -combined (LinePolicy fun1) (LinePolicy fun2) combinedConsumer prod1 prod2 = - manyCombined [fmap ($prod1) fun1, fmap ($prod2) fun2] combinedConsumer - where - manyCombined :: (Show e, Typeable e) - => [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())] - -> (Producer T.Text IO () -> IO (Either e a)) - -> IO (Either e a) - manyCombined actions consumer = do - (outbox, inbox, seal) <- spawn' 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) - -fromProducer :: Producer b IO r -> Pump b e () -fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> 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 :: (Show e, Typeable e) - => DecodingFunction bytes text - -> 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) - -data WrappedError e = WrappedError e - deriving (Show, Typeable) - -instance (Show e, Typeable e) => Exception (WrappedError e) - -elideError :: (Show e, Typeable e) => IO (Either e a) -> IO a -elideError action = action >>= either (throwIO . WrappedError) return - -revealError :: (Show e, Typeable e) => IO a -> IO (Either e a) -revealError action = catch (action >>= return . Right) - (\(WrappedError e) -> return . Left $ e) - -newtype Conceit e a = Conceit { runConceit :: IO (Either e a) } - -instance Functor (Conceit e) where - fmap f (Conceit x) = Conceit $ fmap (fmap f) x - -instance Bifunctor Conceit where - bimap f g (Conceit x) = Conceit $ liftM (bimap f g) x - -instance (Show e, Typeable e) => Applicative (Conceit e) where - pure = Conceit . pure . pure - Conceit fs <*> Conceit as = - Conceit . revealError $ - uncurry ($) <$> concurrently (elideError fs) (elideError as) - -instance (Show e, Typeable e) => Alternative (Conceit e) where - empty = Conceit $ forever (threadDelay maxBound) - Conceit as <|> Conceit bs = - Conceit $ either id id <$> race as bs - -instance (Show e, Typeable e, Monoid a) => Monoid (Conceit e a) where - mempty = Conceit . pure . pure $ mempty - mappend c1 c2 = (<>) <$> c1 <*> c2 - -conceit :: (Show e, Typeable e) - => IO (Either e a) - -> IO (Either e b) - -> IO (Either e (a,b)) -conceit c1 c2 = runConceit $ (,) <$> Conceit c1 <*> Conceit c2 - -{-| - Works similarly to 'Control.Concurrent.Async.mapConcurrently' from the -@async@ package, but if any of the computations fails with @e@, the others are -immediately cancelled and the whole computation fails with @e@. - -} -mapConceit :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b)) -mapConceit f = revealError . mapConcurrently (elideError . f) - -newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } - -instance Functor (Pump b e) where - fmap f (Pump x) = Pump $ fmap (fmap (fmap f)) x - -instance Bifunctor (Pump b) where - bimap f g (Pump x) = Pump $ fmap (liftM (bimap f g)) x - -instance (Show e, Typeable e) => 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 (Show e, Typeable e, Monoid a) => Monoid (Pump b e a) where - mempty = Pump . pure . pure . pure $ mempty - mappend s1 s2 = (<>) <$> s1 <*> s2 - -{-| - A 'Siphon' represents a computation that completely drains a producer, but -may fail early with an error of type @e@. - - '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. - -} -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) - -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 - -instance (Show e, Typeable e) => 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 - ) - -runSiphon :: (Show e, Typeable e) => Siphon b e a -> Producer b IO () -> IO (Either e a) -runSiphon s = case s of - h@(Halting _) -> halting $ Unhalting $ unhalting h - _ -> halting s - --- This might return a computation that *doesn't* completely drain the --- Producer. -halting :: (Show e, Typeable e) => 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 - -unhalting :: (Show e, Typeable e) => 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 - (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 - ) - -instance (Show e, Typeable e, 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 - -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 - -{-| - Builds a 'Siphon' out of a computation that does something with - a 'Producer', but may fail with an error of type @e@. - - Even if the original computation doesn't completely drain the 'Producer', - the constructed 'Siphon' will. --} -siphon :: (Producer b IO () -> IO (Either e a)) - -> Siphon b e a -siphon = Halting - - -{-| - Builds a 'Siphon' out of a computation that drains a 'Producer' completely, -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 - -fromFold :: (Producer b IO () -> IO a) -> Siphon b e a -fromFold aFold = siphon $ fmap (fmap pure) $ aFold - -{-| - Builds a 'Siphon' out of a computation that folds a 'Producer' and drains it completely. --} -fromFold' :: (forall r. Producer b IO r -> IO (a,r)) -> Siphon b e a -fromFold' aFold = siphon' $ fmap (fmap pure) aFold - -fromFold'_ :: (forall r. Producer b IO r -> IO r) -> Siphon b e () -fromFold'_ aFold = fromFold' $ fmap (fmap ((,) ())) aFold - -{-| - Constructs a 'Siphon' that aborts the computation if the underlying -'Producer' produces anything. - -} -unwanted :: a -> Siphon b b a -unwanted a = Unhalting $ \producer -> do - n <- next producer - return $ case n of - Left r -> Right (a,r) - Right (b,_) -> Left b - -executePipeline :: PipingPolicy Void a -> CreatePipeline Void -> IO a -executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline - - -{-| - 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 - 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 - stages. - - The combined @stderr@ stream always has UTF-8 encoding. - - This function has a limitation compared to the standard UNIX pipelines. - If a downstream process terminates early without error, the upstream - processes are not notified and keep going. There is no SIGPIPE-like - functionality, in other words. - -} -executePipelineFallibly :: (Show e,Typeable e) => PipingPolicy e a -> CreatePipeline e -> IO (Either e a) -executePipelineFallibly policy pipeline = case policy of - PPNone a -> fmap (fmap (const a)) $ - executePipelineInternal - (\o _ -> mute $ pipeo o) - (\i o _ -> mute $ pipeio i o) - (\i _ -> mute $ pipei i) - (\i _ -> mute $ pipei i) - pipeline - PPOutput action -> do - (outbox, inbox, seal) <- spawn' Single - runConceit $ - (Conceit $ action $ fromInput inbox) - <* - (Conceit $ executePipelineInternal - (\o _ -> pipeo o) - (\i o _ -> mute $ pipeio i o) - (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ outbox)) - (\i _ -> mute $ pipei i) - pipeline - `finally` atomically seal - ) - PPError action -> do - (eoutbox, einbox, eseal) <- spawn' Single - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action $ fromInput einbox) - <* - (Conceit $ executePipelineInternal - (\o l -> mute $ pipeoe o (errf l)) - (\i o l -> mute $ pipeioe i o (errf l)) - (\i l -> mute $ pipeie i (errf l)) - (\i l -> mute $ pipeie i (errf l)) - pipeline - `finally` atomically eseal) - PPOutputError action -> do - (outbox, inbox, seal) <- spawn' Single - (eoutbox, einbox, eseal) <- spawn' Single - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action $ (fromInput inbox,fromInput einbox)) - <* - (Conceit $ executePipelineInternal - (\o l -> mute $ pipeoe o (errf l)) - (\i o l -> mute $ pipeioe i o (errf l)) - (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ outbox) (errf l)) - (\i l -> mute $ pipeie i (errf l)) - pipeline - `finally` atomically seal `finally` atomically eseal - ) - PPInput action -> do - (outbox, inbox, seal) <- spawn' Single - runConceit $ - (Conceit $ action (toOutput outbox,atomically seal)) - <* - (Conceit $ executePipelineInternal - (\o _ -> mute $ pipeio (fromProducer . fromInput $ inbox) o) - (\i o _ -> mute $ pipeio i o) - (\i _ -> mute $ pipei i) - (\i _ -> mute $ pipei i) - pipeline - `finally` atomically seal - ) - PPInputOutput action -> do - (ioutbox, iinbox, iseal) <- spawn' Single - (ooutbox, oinbox, oseal) <- spawn' Single - runConceit $ - (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox)) - <* - (Conceit $ executePipelineInternal - (\o _ -> mute $ pipeio (fromProducer . fromInput $ iinbox) o) - (\i o _ -> mute $ pipeio i o) - (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ ooutbox)) - (\i _ -> mute $ pipei i) - pipeline - `finally` atomically iseal `finally` atomically oseal - ) - PPInputError action -> do - (outbox, inbox, seal) <- spawn' Single - (eoutbox, einbox, eseal) <- spawn' Single - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox)) - <* - (Conceit $ executePipelineInternal - (\o l -> mute $ pipeioe (fromProducer . fromInput $ inbox) o (errf l)) - (\i o l -> mute $ pipeioe i o (errf l)) - (\i l -> mute $ pipeie i (errf l)) - (\i l -> mute $ pipeie i (errf l)) - pipeline - `finally` atomically seal `finally` atomically eseal - ) - PPInputOutputError action -> do - (ioutbox, iinbox, iseal) <- spawn' Single - (ooutbox, oinbox, oseal) <- spawn' Single - (eoutbox, einbox, eseal) <- spawn' Single - errf <- errorSiphonUTF8 <$> newMVar eoutbox - runConceit $ - (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox)) - <* - (Conceit $ executePipelineInternal - (\o l -> mute $ pipeioe (fromProducer . fromInput $ iinbox) o (errf l)) - (\i o l -> mute $ pipeioe i o (errf l)) - (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ ooutbox) (errf l)) - (\i l -> mute $ pipeie i (errf l)) - pipeline - `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal - ) - where - errorSiphonUTF8 :: MVar (Output ByteString) -> LinePolicy e -> Siphon ByteString e () - errorSiphonUTF8 mvar (LinePolicy fun) = Halting $ fun iterTLines - where - iterTLines = iterT $ \textProducer -> do - -- the P.drain bit was difficult to figure out!!! - join $ withMVar mvar $ \output -> do - runEffect $ (textProducer <* P.yield (singleton '\n')) - >-> P.map Data.Text.Encoding.encodeUtf8 - >-> (toOutput output >> P.drain) - -mute :: Functor f => f a -> f () -mute = fmap (const ()) - -{-| - An individual stage in a process pipeline. - - The 'LinePolicy' field defines how to handle @stderr@ when @stderr@ is - piped. - - Also required is a function that determines if the returned exit code - represents an error or not. This is necessary because some programs use - non-standard exit codes. - -} -data Stage e = Stage - { - processDefinition :: CreateProcess - , stderrLinePolicy :: LinePolicy e - , exitCodePolicy :: Int -> Maybe e - } deriving (Functor) - -{-| - Any stage beyond the first in a process pipeline. - - Incoming data is passed through the 'Pipe' before being fed to the process. - - Use 'cat' (the identity 'Pipe' from 'Pipes') if no pre-processing is required. - -} -data SubsequentStage e = SubsequentStage (forall a.Pipe ByteString ByteString (ExceptT e IO) a) (Stage e) - -instance Functor (SubsequentStage) where - fmap f (SubsequentStage bs s) = SubsequentStage (hoist (mapExceptT $ liftM (bimap f id)) bs) (fmap f s) - -data CreatePipeline e = CreatePipeline (Stage e) (NonEmpty (Tree (SubsequentStage e))) deriving (Functor) - -{-| - Builds a (possibly branching) pipeline assuming that @stderr@ has the same -encoding in all the stages, that no computation is perfored between the stages, -and that any exit code besides 'ExitSuccess' in a stage actually represents an -error. - -} -simplePipeline :: DecodingFunction ByteString Text -> CreateProcess -> NonEmpty (Tree (CreateProcess)) -> CreatePipeline String -simplePipeline decoder initial forest = CreatePipeline (simpleStage initial) (fmap (fmap simpleSubsequentStage) forest) - where - simpleStage cp = Stage cp simpleLinePolicy simpleErrorPolicy - simpleSubsequentStage = SubsequentStage P.cat . simpleStage - simpleLinePolicy = linePolicy decoder (pure ()) id - simpleErrorPolicy = Just . ("Exit failure: " ++) . show - -executePipelineInternal :: (Show e,Typeable e) - => (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 ()) - -> 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 (SubsequentStage pipe (Stage cp lpol ecpol)) forest) = case forest of - [] -> Halting $ \producer -> - blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ hoist lift producer >-> pipe) lpol) cp - c1 : cs -> Halting $ \producer -> - blende ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ hoist lift producer >-> pipe) (runNonEmpty ppend ppend' (c1 :| cs)) lpol) cp - - runNonEmpty ppend ppend' (b :| bs) = - runTree ppend ppend' b <* Prelude.foldr (<*) (pure ()) (runTree ppend' ppend' <$> bs) - - blende :: (Int -> Maybe e) -> Either e (ExitCode,()) -> Either e () - blende f (Right (ExitFailure i,())) = case f i of - Nothing -> Right () - Just e -> Left e - blende _ (Right (ExitSuccess,())) = Right () - blende _ (Left e) = Left e - -{- $reexports - -"System.Process" is re-exported for convenience. - --} - ++-- |+-- This module contains helper functions and types built on top of+-- "System.Process" and "Pipes".+--+-- They provide concurrent, streaming access to the inputs and outputs of+-- system processes.+--+-- Error conditions not directly related to IO are made explicit+-- in the types.+--+-- Regular 'Consumer's, 'Parser's from @pipes-parse@ and various folds can+-- be used to consume the output streams of the external processes.+--+-----------------------------------------------------------------------------++{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE ViewPatterns #-}++module System.Process.Streaming ( + -- * Execution+ execute+ , executeFallibly+ -- * Piping Policies+ , PipingPolicy+ , nopiping+ , pipeo+ , pipee+ , pipeoe+ , pipeoec+ , pipei+ , pipeio+ , pipeie+ , pipeioe+ , pipeioec++ -- * Pumping bytes into stdin+ , Pump (..)+ , fromProducer+ , fromSafeProducer+ , fromFallibleProducer+ -- * Siphoning bytes out of stdout/stderr+ , Siphon+ , siphon+ , siphon'+ , fromFold+ , fromFold'+ , fromFold'_+ , fromConsumer+ , fromSafeConsumer+ , fromFallibleConsumer+ , fromParser+ , unwanted+ , DecodingFunction+ , encoded+ -- * Line handling+ , LinePolicy+ , linePolicy+ , tweakLines+ -- * Pipelines+ , executePipeline+ , executePipelineFallibly+-- , CreatePipeline (..)+ --, simplePipeline+ , Stage+ , stage+ , pipefail+ , inbound+ -- , SubsequentStage (..)+ -- * Re-exports+ -- $reexports+ , module System.Process+ ) where++import Data.Maybe+import Data.Bifunctor+import Data.Functor.Identity+import Data.Either+import Data.Monoid+import Data.Foldable+import Data.Traversable+import Data.Typeable+import Data.Tree+import Data.Text +import Data.Text.Encoding +import Data.Void+import Data.List.NonEmpty+import qualified Data.List.NonEmpty as N+import Control.Applicative+import Control.Monad+import Control.Monad.Trans.Free+import Control.Monad.Trans.Except+import Control.Monad.Trans.State+import Control.Monad.Trans.Writer.Strict+import qualified Control.Monad.Catch as C+import Control.Exception+import Control.Concurrent+import Control.Concurrent.Async+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+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++execute :: PipingPolicy 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. ++ This function re-throws any 'IOException's it encounters.++ If the consumption of the standard streams fails with @e@, the whole+computation is immediately aborted and @e@ is returned. (An exception is not+thrown in this case.). ++ If an error @e@ or an exception happens, the external process is+terminated.+ -}+executeFallibly :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a))+executeFallibly pp record = case pp of+ PPNone a -> executeInternal record nohandles $ + \() -> (return . Right $ a,return ())+ PPOutput action -> executeInternal (record{std_out = CreatePipe}) handleso $+ \h->(action (fromHandle h),hClose h) + PPError action -> executeInternal (record{std_err = CreatePipe}) handlese $+ \h->(action (fromHandle h),hClose h)+ PPOutputError action -> executeInternal (record{std_out = CreatePipe, std_err = CreatePipe}) handlesoe $+ \(hout,herr)->(action (fromHandle hout,fromHandle herr),hClose hout `finally` hClose herr)+ PPInput action -> executeInternal (record{std_in = CreatePipe}) handlesi $+ \h -> (action (toHandle h, hClose h), return ())+ PPInputOutput action -> executeInternal (record{std_in = CreatePipe,std_out = CreatePipe}) handlesio $+ \(hin,hout) -> (action (toHandle hin,hClose hin,fromHandle hout), hClose hout)+ PPInputError action -> executeInternal (record{std_in = CreatePipe,std_err = CreatePipe}) handlesie $+ \(hin,herr) -> (action (toHandle hin,hClose hin,fromHandle herr), hClose herr)+ PPInputOutputError action -> executeInternal (record{std_in = CreatePipe, std_out = CreatePipe, std_err = CreatePipe}) handlesioe $+ \(hin,hout,herr) -> (action (toHandle hin,hClose hin,fromHandle hout,fromHandle herr), hClose hout `finally` hClose herr)++executeInternal :: CreateProcess -> (forall m. Applicative m => (t -> m t) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (t ->(IO (Either e a),IO ())) -> 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+ Nothing -> + throwIO (userError "stdin/stdout/stderr handle unwantedly null")+ `finally`+ terminateCarefully phandle + Just t -> + let (action,cleanup) = allocator t in+ -- Handles must be closed *after* terminating the process, because a close+ -- operation may block if the external process has unflushed bytes in the stream.+ (restore (terminateOnError phandle 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+ mExitCode <- getProcessExitCode pHandle + case mExitCode of + Nothing -> terminateProcess pHandle + Just _ -> return ()++terminateOnError :: ProcessHandle + -> IO (Either e a)+ -> IO (Either e (ExitCode,a))+terminateOnError pHandle action = do+ result <- action+ case result of+ Left e -> do + terminateCarefully pHandle+ return $ Left e+ Right r -> do + exitCode <- waitForProcess pHandle + return $ Right (exitCode,r) ++{-|+ A 'PipingPolicy' 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+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 = + 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)++instance Bifunctor PipingPolicy 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 :: PipingPolicy e ()+nopiping = PPNone ()++{-|+ Pipe @stdout@.+-}+pipeo :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a+pipeo (runSiphon -> siphonout) = PPOutput $ siphonout++{-|+ Pipe @stderr@.+-}+pipee :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a+pipee (runSiphon -> siphonout) = PPError $ siphonout++{-|+ Pipe @stdout@ and @stderr@.+-}+pipeoe :: (Show e,Typeable e) => Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (a,b)+pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = + PPOutputError $ uncurry $ separated siphonout siphonerr ++{-|+ Pipe @stdout@ and @stderr@ and consume them combined as 'Text'. +-}+pipeoec :: (Show e,Typeable e) => LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e a+pipeoec policy1 policy2 (runSiphon -> siphon) = + PPOutputError $ uncurry $ combined policy1 policy2 siphon ++{-|+ Pipe @stdin@.+-}+pipei :: (Show e, Typeable e) => Pump ByteString e i -> PipingPolicy e i+pipei (Pump feeder) = PPInput $ \(consumer,cleanup) -> feeder consumer `finally` cleanup++{-|+ Pipe @stdin@ and @stdout@.+-}+pipeio :: (Show e, Typeable e)+ => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a)+pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) ->+ (conceit (feeder consumer `finally` cleanup) (siphonout producer))++{-|+ Pipe @stdin@ and @stderr@.+-}+pipeie :: (Show e, Typeable e)+ => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy 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 :: (Show e, Typeable e)+ => Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (i,a,b)+pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $+ \(consumer,cleanup,outprod,errprod) -> + (conceit (feeder consumer `finally` cleanup) + (separated siphonout siphonerr outprod errprod))+ where+ flattenTuple (i, (a, b)) = (i,a,b)++{-|+ Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'.+-}+pipeioec :: (Show e, Typeable e)+ => Pump ByteString e i -> LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e (i,a)+pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $+ \(consumer,cleanup,outprod,errprod) -> + (conceit (feeder consumer `finally` cleanup) + (combined policy1 policy2 siphon outprod errprod))++separated :: (Show e, Typeable e)+ => (Producer ByteString IO () -> IO (Either e a))+ -> (Producer ByteString IO () -> IO (Either e b))+ -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e (a,b))+separated outfunc errfunc outprod errprod = + conceit (outfunc outprod) (errfunc errprod)++{-|+ A configuration parameter used in functions that combine lines from+ multiple streams.+ -}++data LinePolicy e = LinePolicy + {+ 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+ } ++instance Functor LinePolicy where+ fmap f (LinePolicy func lt) = LinePolicy (\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) -> LinePolicy e -> LinePolicy e +tweakLines lt' (LinePolicy tear lt) = LinePolicy tear (lt' . lt) ++{-|+ Constructs a 'LinePolicy' 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 :: (Show e, Typeable e)+ => DecodingFunction ByteString Text + -> Siphon ByteString e ()+ -> LinePolicy e +linePolicy decoder lopo = LinePolicy+ (\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 :: (Show e, Typeable e) + => LinePolicy e + -> LinePolicy e + -> (Producer T.Text IO () -> IO (Either e a))+ -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)+combined (LinePolicy fun1 twk1) (LinePolicy fun2 twk2) combinedConsumer prod1 prod2 = + manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer + where + manyCombined :: (Show e, Typeable e) + => [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())]+ -> (Producer T.Text IO () -> IO (Either e a))+ -> IO (Either e a) + manyCombined actions consumer = do+ (outbox, inbox, seal) <- spawn' 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)++fromProducer :: Producer b IO r -> Pump b e ()+fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> 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 :: (Show e, Typeable e) + => DecodingFunction bytes text+ -> 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)++data WrappedError e = WrappedError e+ deriving (Show, Typeable)++instance (Show e, Typeable e) => Exception (WrappedError e)++elideError :: (Show e, Typeable e) => IO (Either e a) -> IO a+elideError action = action >>= either (throwIO . WrappedError) return++revealError :: (Show e, Typeable e) => IO a -> IO (Either e a) +revealError action = catch (action >>= return . Right)+ (\(WrappedError e) -> return . Left $ e) ++newtype Conceit e a = Conceit { runConceit :: IO (Either e a) } deriving Functor++instance Bifunctor Conceit where+ bimap f g (Conceit x) = Conceit $ liftM (bimap f g) x++instance (Show e, Typeable e) => Applicative (Conceit e) where+ pure = Conceit . pure . pure+ Conceit fs <*> Conceit as =+ Conceit . revealError $ + uncurry ($) <$> concurrently (elideError fs) (elideError as)++instance (Show e, Typeable e) => Alternative (Conceit e) where+ empty = Conceit $ forever (threadDelay maxBound)+ Conceit as <|> Conceit bs =+ Conceit $ either id id <$> race as bs++instance (Show e, Typeable e, Monoid a) => Monoid (Conceit e a) where+ mempty = Conceit . pure . pure $ mempty+ mappend c1 c2 = (<>) <$> c1 <*> c2++conceit :: (Show e, Typeable e) + => IO (Either e a)+ -> IO (Either e b)+ -> IO (Either e (a,b))+conceit c1 c2 = runConceit $ (,) <$> Conceit c1 <*> Conceit c2++{-| + Works similarly to 'Control.Concurrent.Async.mapConcurrently' from the+@async@ package, but if any of the computations fails with @e@, the others are+immediately cancelled and the whole computation fails with @e@. + -}+mapConceit :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b))+mapConceit f = revealError . mapConcurrently (elideError . f)++newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor++instance Bifunctor (Pump b) where+ bimap f g (Pump x) = Pump $ fmap (liftM (bimap f g)) x++instance (Show e, Typeable e) => 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 (Show e, Typeable e, Monoid a) => Monoid (Pump b e a) where+ mempty = Pump . pure . pure . pure $ mempty+ mappend s1 s2 = (<>) <$> s1 <*> s2++{-| + A 'Siphon' represents a computation that completely drains a producer, but+may fail early with an error of type @e@. ++ '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.+ -}+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)++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++instance (Show e, Typeable e) => 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+ ) ++runSiphon :: (Show e, Typeable e) => Siphon b e a -> Producer b IO () -> IO (Either e a)+runSiphon s = case s of + h@(Halting _) -> halting $ Unhalting $ unhalting h + _ -> halting s++-- This might return a computation that *doesn't* completely drain the+-- Producer.+halting :: (Show e, Typeable e) => 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 ++unhalting :: (Show e, Typeable e) => 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 + (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+ )++instance (Show e, Typeable e, 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 ++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 ++{-| + Builds a 'Siphon' out of a computation that does something with+ a 'Producer', but may fail with an error of type @e@.+ + Even if the original computation doesn't completely drain the 'Producer',+ the constructed 'Siphon' will.+-}+siphon :: (Producer b IO () -> IO (Either e a))+ -> Siphon b e a +siphon = Halting+++{-| + Builds a 'Siphon' out of a computation that drains a 'Producer' completely,+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++{-| + Useful in combination with 'Pipes.Text.toLazyM' from @pipes-text@ and+ 'Pipes.ByteString.toLazyM' from @pipes-bytestring@, when the user+ wants to collect all the output. +-}+fromFold :: (Producer b IO () -> IO a) -> Siphon b e a +fromFold aFold = siphon $ fmap (fmap pure) $ aFold ++{-| + Builds a 'Siphon' out of a computation that folds a 'Producer' and drains it completely.+-}+fromFold' :: (forall r. Producer b IO r -> IO (a,r)) -> Siphon b e a +fromFold' aFold = siphon' $ fmap (fmap pure) aFold++fromFold'_ :: (forall r. Producer b IO r -> IO r) -> Siphon b e () +fromFold'_ aFold = fromFold' $ fmap (fmap ((,) ())) aFold++{-|+ Constructs a 'Siphon' that aborts the computation if the underlying+'Producer' produces anything.+ -}+unwanted :: a -> Siphon b b a+unwanted a = Unhalting $ \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 pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline+++{-|+ 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+ 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+ stages.++ The combined @stderr@ stream always has UTF-8 encoding.++ This function has a limitation compared to the standard UNIX pipelines.+ If a downstream process terminates early without error, the upstream+ processes are not notified and keep going. There is no SIGPIPE-like+ functionality, in other words. + -}+executePipelineFallibly :: (Show e,Typeable e) => PipingPolicy 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 + PPError action -> do+ (eoutbox, einbox, eseal) <- spawn' Single+ 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+ errf <- errorSiphonUTF8 <$> newMVar eoutbox+ runConceit $ + (Conceit $ action $ (fromInput inbox,fromInput einbox))+ <* + (Conceit $ blende ecpol <$> executeFallibly+ (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+ PPInputError action -> do+ (outbox, inbox, seal) <- spawn' Single+ (eoutbox, einbox, eseal) <- spawn' Single+ errf <- errorSiphonUTF8 <$> newMVar eoutbox+ runConceit $ + (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))+ <* + (Conceit $ blende ecpol <$> executeFallibly+ (pipeie (fromProducer . fromInput $ inbox) (errf lpol)) cp+ `finally` atomically seal `finally` atomically eseal+ )+ PPInputOutputError action -> do+ (ioutbox, iinbox, iseal) <- spawn' Single+ (ooutbox, oinbox, oseal) <- spawn' Single+ (eoutbox, einbox, eseal) <- spawn' Single+ errf <- errorSiphonUTF8 <$> newMVar eoutbox+ runConceit $ + (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))+ <* + (Conceit $ blende ecpol <$> executeFallibly+ (pipeioe (fromProducer . fromInput $ iinbox) + (fromConsumer . toOutput $ ooutbox) + (errf lpol) + )+ cp+ `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal+ )+executePipelineFallibly policy (Node s (s':ss)) = + let pipeline = CreatePipeline s $ s' :| ss + in case policy of + PPNone a -> fmap (fmap (const a)) $+ executePipelineInternal + (\o _ -> mute $ pipeo o) + (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipei i) + (\i _ -> mute $ pipei i) + pipeline+ PPOutput action -> do+ (outbox, inbox, seal) <- spawn' Single+ runConceit $ + (Conceit $ action $ fromInput inbox)+ <* + (Conceit $ executePipelineInternal + (\o _ -> pipeo o)+ (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ outbox)) + (\i _ -> mute $ pipei i)+ pipeline+ `finally` atomically seal+ ) + PPError action -> do+ (eoutbox, einbox, eseal) <- spawn' Single+ errf <- errorSiphonUTF8 <$> newMVar eoutbox+ runConceit $ + (Conceit $ action $ fromInput einbox)+ <*+ (Conceit $ executePipelineInternal + (\o l -> mute $ pipeoe o (errf l)) + (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeie i (errf l)) + (\i l -> mute $ pipeie i (errf l))+ pipeline+ `finally` atomically eseal)+ PPOutputError action -> do+ (outbox, inbox, seal) <- spawn' Single+ (eoutbox, einbox, eseal) <- spawn' Single+ errf <- errorSiphonUTF8 <$> newMVar eoutbox+ runConceit $ + (Conceit $ action $ (fromInput inbox,fromInput einbox))+ <* + (Conceit $ executePipelineInternal + (\o l -> mute $ pipeoe o (errf l))+ (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ outbox) (errf l)) + (\i l -> mute $ pipeie i (errf l))+ pipeline+ `finally` atomically seal `finally` atomically eseal+ )+ PPInput action -> do+ (outbox, inbox, seal) <- spawn' Single+ runConceit $ + (Conceit $ action (toOutput outbox,atomically seal))+ <* + (Conceit $ executePipelineInternal + (\o _ -> mute $ pipeio (fromProducer . fromInput $ inbox) o)+ (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipei i) + (\i _ -> mute $ pipei i) + pipeline+ `finally` atomically seal+ )+ PPInputOutput action -> do+ (ioutbox, iinbox, iseal) <- spawn' Single+ (ooutbox, oinbox, oseal) <- spawn' Single+ runConceit $ + (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox))+ <* + (Conceit $ executePipelineInternal + (\o _ -> mute $ pipeio (fromProducer . fromInput $ iinbox) o)+ (\i o _ -> mute $ pipeio i o) + (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ ooutbox)) + (\i _ -> mute $ pipei i) + pipeline+ `finally` atomically iseal `finally` atomically oseal+ )+ PPInputError action -> do+ (outbox, inbox, seal) <- spawn' Single+ (eoutbox, einbox, eseal) <- spawn' Single+ errf <- errorSiphonUTF8 <$> newMVar eoutbox+ runConceit $ + (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))+ <* + (Conceit $ executePipelineInternal + (\o l -> mute $ pipeioe (fromProducer . fromInput $ inbox) o (errf l))+ (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeie i (errf l)) + (\i l -> mute $ pipeie i (errf l)) + pipeline+ `finally` atomically seal `finally` atomically eseal+ )+ PPInputOutputError action -> do+ (ioutbox, iinbox, iseal) <- spawn' Single+ (ooutbox, oinbox, oseal) <- spawn' Single+ (eoutbox, einbox, eseal) <- spawn' Single+ errf <- errorSiphonUTF8 <$> newMVar eoutbox+ runConceit $ + (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))+ <* + (Conceit $ executePipelineInternal + (\o l -> mute $ pipeioe (fromProducer . fromInput $ iinbox) o (errf l))+ (\i o l -> mute $ pipeioe i o (errf l)) + (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ ooutbox) (errf l)) + (\i l -> mute $ pipeie i (errf l)) + pipeline+ `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 + where + iterTLines = iterT $ \textProducer -> do+ -- the P.drain bit was difficult to figure out!!!+ join $ withMVar mvar $ \output -> do+ runEffect $ (textProducer <* P.yield (singleton '\n')) + >-> P.map Data.Text.Encoding.encodeUtf8 + >-> (toOutput output >> P.drain)++mute :: Functor f => f a -> f ()+mute = fmap (const ())++{-|+ An individual stage in a process pipeline. + -}+data Stage e = Stage + {+ processDefinition' :: CreateProcess + , stderrLinePolicy' :: LinePolicy 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' out of a 'LinePolicy' 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 lp ec cp = Stage cp lp ec (hoist lift) ++{-|+ Applies a transformation to the stream of bytes flowing into a stage from previous stages.++ This function is ignored for first stages.+-}+inbound :: (forall r. Producer ByteString (ExceptT e IO) r -> Producer ByteString (ExceptT e IO) r)+ -> Stage e -> Stage e +inbound f (Stage a b c d) = Stage a b c (f . d)++data CreatePipeline e = CreatePipeline (Stage e) (NonEmpty (Tree (Stage e))) deriving (Functor)++executePipelineInternal :: (Show e,Typeable e) + => (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 ())+ -> 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 ->+ blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ pipe producer) lpol) cp+ c1 : cs -> Halting $ \producer ->+ 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) + +blende :: (ExitCode -> Either e ()) -> Either e (ExitCode,a) -> Either e a+blende f r = r >>= \(ec,a) -> f ec *> pure a++{-|+ Converts any 'ExitFailure' to the left side of an 'Either'. +-}+pipefail :: ExitCode -> Either Int ()+pipefail ec = case ec of+ ExitSuccess -> Right ()+ ExitFailure i -> Left i++{- $reexports+ +"System.Process" is re-exported for convenience.++-} +
tests/test.hs view
@@ -1,255 +1,323 @@-{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE ViewPatterns #-} - -module Main where - -import Test.Tasty -import Test.Tasty.HUnit - -import Data.Bifunctor -import Data.Monoid -import Data.Foldable -import Data.List.NonEmpty -import Data.ByteString -import Data.ByteString.Lazy as BL -import Data.Text.Lazy as TL -import Data.Typeable -import Data.Tree -import qualified Data.Attoparsec.Text as A -import Control.Applicative -import Control.Monad -import Control.Monad.Trans.Except -import Control.Lens (view) -import Control.Concurrent.Async -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.Directory -import System.Process.Streaming - -main = defaultMain tests - -tests :: TestTree -tests = testGroup "Tests" - [ testCollectStdoutStderrAsByteString - , testFeedStdinCollectStdoutAsText - , testCombinedStdoutStderr - , testInterruptExecution - , testFailIfAnythingShowsInStderr - , testTwoTextParsersInParallel - , testCountWords - , testBasicPipeline - , testBranchingPipeline - ] - -------------------------------------------------------------------------------- -testCollectStdoutStderrAsByteString :: TestTree -testCollectStdoutStderrAsByteString = testCase "collectStdoutStderrAsByteString" $ do - r <- collectStdoutStderrAsByteString - case r of - (ExitSuccess,("ooo\nppp\n","eee\nffff\n")) -> return () - _ -> assertFailure "oops" - -collectStdoutStderrAsByteString :: IO (ExitCode,(BL.ByteString,BL.ByteString)) -collectStdoutStderrAsByteString = execute - (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) - (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") - - -------------------------------------------------------------------------------- -testFeedStdinCollectStdoutAsText :: TestTree -testFeedStdinCollectStdoutAsText = testCase "feedStdinCollectStdoutAsText" $ do - r <- feedStdinCollectStdoutAsText - case r of - (ExitSuccess,((),"aaaaaa\naaaaa")) -> return () - _ -> assertFailure "oops" - -feedStdinCollectStdoutAsText :: IO (ExitCode, ((), Text)) -feedStdinCollectStdoutAsText = execute - (pipeio (fromProducer $ yield "aaaaaa\naaaaa") - (encoded T.decodeIso8859_1 (pure id) $ fromFold T.toLazyM)) - (shell "cat") - -------------------------------------------------------------------------------- - -testCombinedStdoutStderr :: TestTree -testCombinedStdoutStderr = testCase "testCombinedStdoutStderr" $ do - r <- combinedStdoutStderr - case r of - (ExitSuccess,TL.lines -> ls) -> do - assertEqual "line count" (Prelude.length ls) 4 - assertBool "expected lines" $ - getAll $ foldMap (All . flip Prelude.elem ls) $ - [ "ooo" - , "ppp" - , "errprefix: eee" - , "errprefix: ffff" - ] - _ -> assertFailure "oops" - -combinedStdoutStderr :: IO (ExitCode,TL.Text) -combinedStdoutStderr = execute - (pipeoec (linePolicy T.decodeIso8859_1 (pure ()) id) - (linePolicy T.decodeIso8859_1 (pure ()) annotate) - (fromFold T.toLazyM)) - (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }") - where - annotate x = P.yield "errprefix: " *> x - -------------------------------------------------------------------------------- - -testInterruptExecution :: TestTree -testInterruptExecution = localOption (mkTimeout $ 5*(10^6)) $ - testCase "interruptExecution" $ do - r <- interruptExecution - case r of - Left "interrupted" -> return () - _ -> assertFailure "oops" - -interruptExecution :: IO (Either String (ExitCode,())) -interruptExecution = executeFallibly - (pipeo . siphon $ \_ -> runExceptT . throwE $ "interrupted") - (shell "sleep 100s") - -------------------------------------------------------------------------------- - -testFailIfAnythingShowsInStderr :: TestTree -testFailIfAnythingShowsInStderr = localOption (mkTimeout $ 5*(10^6)) $ - testCase "failIfAnythingShowsInStderr" $ do - r <- failIfAnythingShowsInStderr - case r of - Left "morestuff\n" -> return () - _ -> assertFailure "oops" - -failIfAnythingShowsInStderr :: IO (Either T.ByteString (ExitCode,())) -failIfAnythingShowsInStderr = executeFallibly - (pipee (unwanted ())) - (shell "{ echo morestuff 1>&2 ; sleep 100s ; }") - -------------------------------------------------------------------------------- - -testTwoTextParsersInParallel :: TestTree -testTwoTextParsersInParallel = testCase "twoTextParsersInParallel" $ do - r <- twoTextParsersInParallel - case r of - Right (ExitSuccess,("ooooooo","aaaaaa")) -> return () - _ -> assertFailure "oops" - -parseChars :: Char -> A.Parser [Char] -parseChars c = fmap mconcat $ - many (A.notChar c) *> A.many1 (some (A.char c) <* many (A.notChar c)) - -parser1 = parseChars 'o' - -parser2 = parseChars 'a' - -twoTextParsersInParallel :: IO (Either String (ExitCode,([Char], [Char]))) -twoTextParsersInParallel = executeFallibly - (pipeo (encoded T.decodeIso8859_1 (pure id) $ - (,) <$> adapt parser1 <*> adapt parser2)) - (shell "{ echo ooaaoo ; echo aaooaoa; }") - where - adapt p = fromParser $ do - r <- P.parse p - return $ case r of - Just (Right r') -> Right r' - _ -> Left "parse error" - -------------------------------------------------------------------------------- - -testCountWords :: TestTree -testCountWords = testCase "testCountWords" $ do - r <- countWords - case r of - (ExitSuccess,3) -> return () - _ -> assertFailure "oops" - -countWords :: IO (ExitCode,Int) -countWords = execute - (pipeo (encoded T.decodeIso8859_1 (pure id) $ - fromFold $ P.sum . G.folds const () (const 1) . view T.words)) - (shell "{ echo aaa ; echo bbb ; echo ccc ; }") - -------------------------------------------------------------------------------- -testBasicPipeline :: TestTree -testBasicPipeline = testCase "basicPipeline" $ do - r <- basicPipeline - case r of - Right ((),"aaaccc\n") -> return () - _ -> assertFailure "oops" - -basicPipeline :: IO (Either String ((),BL.ByteString)) -basicPipeline = executePipelineFallibly - (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") - (fromFold B.toLazyM)) - (simplePipeline T.decodeUtf8 (shell "grep aaa") (pure . pure $ shell "grep ccc")) - -------------------------------------------------------------------------------- - -testBranchingPipeline :: TestTree -testBranchingPipeline = testCase "branchingPipeline" $ do - exists <- doesFileExist branchingPipelineFile - when exists $ removeFile branchingPipelineFile - r <- branchingPipeline - case r of - ("ppp\v","eee\nffff\n") -> return () - _ -> assertFailure "oops" - fileContents <- withFile branchingPipelineFile ReadMode $ \hIn -> do - B.toLazyM $ B.fromHandle hIn - assertBool "file contexts" $ BL.isPrefixOf "yyy" fileContents - -branchingPipelineFile :: String -branchingPipelineFile = "dist/test/process-streaming-pipeline-text.txt" - -branchingPipeline :: IO (BL.ByteString, BL.ByteString) -branchingPipeline = executePipeline - (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) - (CreatePipeline rootStage . fromList $ - [ Node branch1 [pure terminalStage1] , Node branch2 [pure terminalStage2] ] ) - where - succStage = SubsequentStage (P.map (Data.ByteString.map succ)) - - rootStage :: (Show e, Typeable e) => Stage e - rootStage = Stage (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ; echo ffff 1>&2 ; }") - (linePolicy T.decodeIso8859_1 (pure ()) id) - (\_ -> Nothing) - - branch1 :: (Show e, Typeable e) => SubsequentStage e - branch1 = SubsequentStage cat $ - Stage (shell "grep ooo") - (linePolicy T.decodeIso8859_1 (pure ()) id) - (\_ -> Nothing) - - branch2 :: (Show e, Typeable e) => SubsequentStage e - branch2 = SubsequentStage cat $ - Stage (shell "grep xxx") - (linePolicy T.decodeIso8859_1 (pure ()) id) - (\_ -> Nothing) - - terminalStage1 :: (Show e, Typeable e) => SubsequentStage e - terminalStage1 = succStage $ - Stage (shell "tr -d b") - (linePolicy T.decodeIso8859_1 (pure ()) id) - (\_ -> Nothing) - - terminalStage2 :: (Show e, Typeable e) => SubsequentStage e - terminalStage2 = succStage $ - Stage (shell $ "cat > " ++ branchingPipelineFile) - (linePolicy T.decodeIso8859_1 (pure ()) id) - (\_ -> Nothing) - - - - - +{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ViewPatterns #-}++module Main where ++import Test.Tasty+import Test.Tasty.HUnit++import Data.Bifunctor+import Data.Monoid+import Data.Foldable+import Data.List.NonEmpty+import Data.ByteString+import Data.ByteString.Lazy as BL+import Data.Text.Lazy as TL+import Data.Typeable+import Data.Tree+import qualified Data.Attoparsec.Text as A+import Control.Applicative+import Control.Monad+import Control.Monad.Trans.Except+import Control.Lens (view)+import Control.Concurrent.Async+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.Directory+import System.Process.Streaming++main = defaultMain tests++tests :: TestTree+tests = testGroup "Tests" + [ testCollectStdoutStderrAsByteString+ , testFeedStdinCollectStdoutAsText + , testCombinedStdoutStderr+ , testInterruptExecution + , testFailIfAnythingShowsInStderr + , testTwoTextParsersInParallel + , testCountWords + , testSingletonPipeline + , testBasicPipeline+ , testBranchingPipeline + , testDrainageDeadlock+ , testAlternatingWithCombined + ]++-------------------------------------------------------------------------------+testCollectStdoutStderrAsByteString :: TestTree+testCollectStdoutStderrAsByteString = testCase "collectStdoutStderrAsByteString" $ do+ r <- collectStdoutStderrAsByteString+ case r of+ (ExitSuccess,("ooo\nppp\n","eee\nffff\n")) -> return ()+ _ -> assertFailure "oops"++collectStdoutStderrAsByteString :: IO (ExitCode,(BL.ByteString,BL.ByteString))+collectStdoutStderrAsByteString = execute+ (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM))+ (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }")+++-------------------------------------------------------------------------------+testFeedStdinCollectStdoutAsText :: TestTree+testFeedStdinCollectStdoutAsText = testCase "feedStdinCollectStdoutAsText" $ do+ r <- feedStdinCollectStdoutAsText+ case r of+ (ExitSuccess,((),"aaaaaa\naaaaa")) -> return ()+ _ -> assertFailure "oops"++feedStdinCollectStdoutAsText :: IO (ExitCode, ((), Text))+feedStdinCollectStdoutAsText = execute+ (pipeio (fromProducer $ yield "aaaaaa\naaaaa")+ (encoded T.decodeIso8859_1 (pure id) $ fromFold T.toLazyM))+ (shell "cat")++-------------------------------------------------------------------------------++testCombinedStdoutStderr :: TestTree+testCombinedStdoutStderr = testCase "testCombinedStdoutStderr" $ do+ r <- combinedStdoutStderr + case r of + (ExitSuccess,TL.lines -> ls) -> do+ assertEqual "line count" (Prelude.length ls) 4+ assertBool "expected lines" $ + getAll $ foldMap (All . flip Prelude.elem ls) $+ [ "ooo"+ , "ppp"+ , "errprefix: eee"+ , "errprefix: ffff"+ ]+ _ -> assertFailure "oops"++combinedStdoutStderr :: IO (ExitCode,TL.Text)+combinedStdoutStderr = execute+ (pipeoec (linePolicy T.decodeIso8859_1 (pure ()))+ (tweakLines annotate $ linePolicy T.decodeIso8859_1 (pure ())) + (fromFold T.toLazyM))+ (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }")+ where+ annotate x = P.yield "errprefix: " *> x ++-------------------------------------------------------------------------------++testInterruptExecution :: TestTree+testInterruptExecution = localOption (mkTimeout $ 5*(10^6)) $+ testCase "interruptExecution" $ do+ r <- interruptExecution+ case r of+ Left "interrupted" -> return ()+ _ -> assertFailure "oops"++interruptExecution :: IO (Either String (ExitCode,()))+interruptExecution = executeFallibly+ (pipeo . siphon $ \_ -> runExceptT . throwE $ "interrupted")+ (shell "sleep 100s")++-------------------------------------------------------------------------------++testFailIfAnythingShowsInStderr :: TestTree+testFailIfAnythingShowsInStderr = localOption (mkTimeout $ 5*(10^6)) $+ testCase "failIfAnythingShowsInStderr" $ do+ r <- failIfAnythingShowsInStderr + case r of+ Left "morestuff\n" -> return ()+ _ -> assertFailure "oops"++failIfAnythingShowsInStderr :: IO (Either T.ByteString (ExitCode,()))+failIfAnythingShowsInStderr = executeFallibly+ (pipee (unwanted ()))+ (shell "{ echo morestuff 1>&2 ; sleep 100s ; }")++-------------------------------------------------------------------------------++testTwoTextParsersInParallel :: TestTree+testTwoTextParsersInParallel = testCase "twoTextParsersInParallel" $ do+ r <- twoTextParsersInParallel+ case r of + Right (ExitSuccess,("ooooooo","aaaaaa")) -> return ()+ _ -> assertFailure "oops"++parseChars :: Char -> A.Parser [Char] +parseChars c = fmap mconcat $ + many (A.notChar c) *> A.many1 (some (A.char c) <* many (A.notChar c))+ +parser1 = parseChars 'o'++parser2 = parseChars 'a'++twoTextParsersInParallel :: IO (Either String (ExitCode,([Char], [Char])))+twoTextParsersInParallel = executeFallibly+ (pipeo (encoded T.decodeIso8859_1 (pure id) $ + (,) <$> adapt parser1 <*> adapt parser2))+ (shell "{ echo ooaaoo ; echo aaooaoa; }")+ where+ adapt p = fromParser $ do+ r <- P.parse p+ return $ case r of+ Just (Right r') -> Right r'+ _ -> Left "parse error"++-------------------------------------------------------------------------------++testCountWords :: TestTree+testCountWords = testCase "testCountWords" $ do+ r <- countWords + case r of + (ExitSuccess,3) -> return () + _ -> assertFailure "oops"++countWords :: IO (ExitCode,Int)+countWords = execute+ (pipeo (encoded T.decodeIso8859_1 (pure id) $+ fromFold $ P.sum . G.folds const () (const 1) . view T.words))+ (shell "{ echo aaa ; echo bbb ; echo ccc ; }")++-------------------------------------------------------------------------------+testSingletonPipeline :: TestTree+testSingletonPipeline = testCase "singletonPipeline" $ do+ r <- singletonPipeline + case r of+ (Right ("ooo\nppp\n","eee\nffff\n")) -> return ()+ _ -> assertFailure "oops"++singletonPipeline :: IO (Either Int (BL.ByteString,BL.ByteString))+singletonPipeline = executePipelineFallibly + (pipeoe (fromFold B.toLazyM) + (fromFold B.toLazyM)) + (pure $ stage (linePolicy T.decodeUtf8 (pure ())) pipefail $ + shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ; echo ffff 1>&2 ; }"+ ) ++-------------------------------------------------------------------------------+testBasicPipeline :: TestTree+testBasicPipeline = testCase "basicPipeline" $ do+ r <- basicPipeline + case r of + Right ((),"aaaccc\n") -> return () + _ -> assertFailure "oops"++basicPipeline :: IO (Either Int ((),BL.ByteString))+basicPipeline = executePipelineFallibly + (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") + (fromFold B.toLazyM)) + (fmap (stage (linePolicy T.decodeUtf8 (pure ())) pipefail) $ + Node (shell "grep aaa") [Node (shell "grep ccc") []] )++-------------------------------------------------------------------------------++testBranchingPipeline :: TestTree+testBranchingPipeline = testCase "branchingPipeline" $ do+ exists <- doesFileExist branchingPipelineFile+ when exists $ removeFile branchingPipelineFile+ r <- branchingPipeline + case r of + (Right ("ppp\v","eee\nffff\n")) -> return () + _ -> assertFailure "oops"+ fileContents <- withFile branchingPipelineFile ReadMode $ \hIn -> do+ B.toLazyM $ B.fromHandle hIn + assertBool "file contexts" $ BL.isPrefixOf "yyy" fileContents ++branchingPipelineFile :: String +branchingPipelineFile = "dist/test/process-streaming-pipeline-text.txt"++branchingPipeline :: IO (Either Int (BL.ByteString, BL.ByteString))+branchingPipeline = executePipelineFallibly+ (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) + (Node rootStage + [ Node branch1 [Node terminalStage1 []]+ , Node branch2 [Node terminalStage2 []]+ ] + )+ where+ succStage = P.map (Data.ByteString.map succ)++ rootStage :: Stage Int + rootStage = stage (linePolicy T.decodeIso8859_1 (pure ())) + pipefail+ (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ; echo ffff 1>&2 ; }")++ branch1 :: Stage Int + branch1 = stage (linePolicy T.decodeIso8859_1 (pure ())) + pipefail+ (shell "grep ooo")+ branch2 :: Stage Int + branch2 = stage (linePolicy T.decodeIso8859_1 (pure ())) + pipefail+ (shell "grep xxx")++ terminalStage1 :: Stage Int + terminalStage1 = inbound (\p -> p >-> succStage) $+ stage (linePolicy T.decodeIso8859_1 (pure ())) + pipefail+ (shell "tr -d b")++ terminalStage2 :: Stage Int+ terminalStage2 = inbound (\p -> p >-> succStage) $+ stage (linePolicy T.decodeIso8859_1 (pure ())) + pipefail+ (shell $ "cat > " ++ branchingPipelineFile)++-------------------------------------------------------------------------------++testDrainageDeadlock :: TestTree+testDrainageDeadlock = localOption (mkTimeout $ 20*(10^6)) $+ testCase "drainageDeadlock" $ do+ execute nopiping $ shell "chmod u+x tests/alternating.sh"+ r <- drainageDeadlock+ case r of+ (ExitSuccess,((),())) -> return ()+ _ -> assertFailure "oops"++-- A bug caused some streams not to be drained, and this caused problems+-- due to full output buffers.+drainageDeadlock :: IO (ExitCode,((),()))+drainageDeadlock = execute+ (pipeoe (pure ()) (fromFold $ \producer -> next producer >> pure ()))+ (proc "tests/alternating.sh" [])+++-------------------------------------------------------------------------------++testAlternatingWithCombined :: TestTree+testAlternatingWithCombined = localOption (mkTimeout $ 20*(10^6)) $+ testCase "testAlternatingWithCombined" $ do+ execute nopiping $ shell "chmod u+x tests/alternating.sh"+ r <- alternatingWithCombined + case r of + (ExitSuccess,80000) -> return ()+ _ -> assertFailure "oops"+ r <- alternatingWithCombined2 + case r of + (ExitSuccess,(80000,80000)) -> return ()+ _ -> assertFailure "oops"++alternatingWithCombined :: IO (ExitCode,Integer)+alternatingWithCombined = execute+ (pipeoec lp lp countLines)+ (proc "tests/alternating.sh" [])+ where+ lp = linePolicy T.decodeIso8859_1 (pure ()) + countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines+++alternatingWithCombined2 :: IO (ExitCode,(Integer,Integer))+alternatingWithCombined2 = execute+ (pipeoec lp lp $ (,) <$> countLines <*> countLines)+ (proc "tests/alternating.sh" [])+ where+ lp = linePolicy T.decodeIso8859_1 (pure ()) + countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines++