streaming-process (empty) → 0.1.0.0
raw patch · 8 files changed
+639/−0 lines, 8 filesdep +QuickCheckdep +basedep +bytestringsetup-changed
Dependencies added: QuickCheck, base, bytestring, directory, exceptions, hspec, lifted-async, monad-control, process, quickcheck-instances, streaming, streaming-bytestring, streaming-commons, streaming-concurrency, streaming-process, streaming-with, transformers, transformers-base
Files
- ChangeLog.md +5/−0
- LICENSE +20/−0
- README.md +38/−0
- Setup.hs +2/−0
- src/Streaming/Process.hs +262/−0
- src/Streaming/Process/Lifted.hs +197/−0
- streaming-process.cabal +58/−0
- test/simple.hs +57/−0
+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Revision history for streaming-process++## 0.1.0.0 -- 2018-05-23++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2017 Ivan Lazar Miljenovic++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,38 @@+streaming-process+=================++[](https://hackage.haskell.org/package/streaming-process) [](https://travis-ci.org/haskell-streaming/streaming-process)++Run a process, streaming data in or out.++A lot of configuration options are available to fine-tune which inputs+and outputs are streamed.++This uses [streaming-with] to handle resource management, and+[streaming-concurrency] for handling both `stdout` and `stderr`+together.++As such, code is typically run in a continuation-passing-style. You+may wish to use the `Streaming.Process.Lifted` module if you have many+of these nested.++[streaming-with]: http://hackage.haskell.org/package/streaming-with+[streaming-concurrency]: http://hackage.haskell.org/package/streaming-concurrency++Exceptions+----------++The functions in this library will all throw+`ProcessExitedUnsuccessfully` if the process/command itself fails.++WARNING+-------++If using this module, you will need to have:++```cabal+ghc-options -threaded+```++in the executable section of your `.cabal` file, otherwise your code+will likely hang!
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ src/Streaming/Process.hs view
@@ -0,0 +1,262 @@+{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses, NamedFieldPuns,+ RankNTypes, RecordWildCards #-}++{- |+ Module : Streaming.Process+ Description : Run system process with support for Streams+ Copyright : (c) Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++ Run system commands in a streaming fashion.++ __WARNING:__ If using this module, you will need to have+ @ghc-options -threaded@ in your @.cabal@ file otherwise it will+ likely hang!++ These functions are typically written to be used in a+ continuation-passing style to allow for proper finalisation. If+ you have many of these nested, it may be easier to use the+ "Streaming.Process.Lifted" module.++ These functions will all throw 'ProcessExitedUnsuccessfully' if the+ process\/command itself fails.++ -}+module Streaming.Process+ ( -- * High level functions+ withStreamingProcess+ , withStreamingCommand+ , streamInput+ , streamInputCommand+ , withStreamingOutput+ , withStreamingOutputCommand+ -- * Lower level+ , StreamProcess(..)+ , switchOutputs+ , WithStream(..)+ , WithStream'+ , SupplyStream(..)+ , withStreamProcess+ , withStreamCommand+ , withProcessHandles+ , processInput+ , withProcessOutput+ -- * Interleaved stdout and stderr+ , StdOutErr+ , withStreamOutputs+ -- * Re-exports+ -- $reexports+ , module Data.Streaming.Process+ , concurrently+ ) where++import Data.ByteString.Streaming (ByteString)+import qualified Data.ByteString.Streaming as SB+import Streaming (hoist)+import Streaming.Concurrent (unbounded, withMergedStreams)+import qualified Streaming.Prelude as S++import Control.Concurrent.Async.Lifted (concurrently)+import Control.Monad.Base (MonadBase)+import Control.Monad.Catch (MonadMask, finally, onException, throwM)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Streaming.Process+import Data.Streaming.Process.Internal (InputSource(..), OutputSink(..))+import System.Exit (ExitCode(..))+import System.IO (hClose)+import System.Process (CreateProcess(..),+ StdStream(CreatePipe), shell)++--------------------------------------------------------------------------------++-- | Feeds the provided data into the specified process, then+-- concurrently streams stdout and stderr into the provided+-- continuation.+--+-- Note that the monad used in the 'StdOutErr' argument to the+-- continuation can be different from the final result, as it's up+-- to the caller to make sure the result is reached.+withStreamingProcess :: (MonadBaseControl IO m, MonadIO m, MonadMask m+ , MonadBase IO n)+ => CreateProcess -> ByteString m v+ -> (StdOutErr n () -> m r) -> m r+withStreamingProcess cp inp = withStreamProcess cp+ . flip (withProcessHandles inp)++-- | As with 'withStreamingProcess', but run the specified command in+-- a shell.+withStreamingCommand :: (MonadBaseControl IO m, MonadIO m, MonadMask m+ , MonadBase IO n)+ => String -> ByteString m v+ -> (StdOutErr n () -> m r) -> m r+withStreamingCommand = withStreamingProcess . shell++-- | Feed input into a process with no expected output.+streamInput :: (MonadIO m, MonadMask m) => CreateProcess+ -> ByteString m r -> m r+streamInput cp = withStreamProcess cp . flip processInput++-- | As with 'streamInput' but run the specified command in a shell.+streamInputCommand :: (MonadIO m, MonadMask m) => String+ -> ByteString m r -> m r+streamInputCommand = streamInput . shell++-- | Obtain the output of a process with no input (ignoring error+-- output).+withStreamingOutput :: (MonadIO n, MonadIO m, MonadMask m)+ => CreateProcess+ -> (ByteString n () -> m r) -> m r+withStreamingOutput cp = withStreamProcess cp . flip withProcessOutput++-- | As with 'withStreamingOutput' but run the specified command in a+-- shell.+withStreamingOutputCommand :: (MonadIO n, MonadIO m, MonadMask m)+ => String+ -> (ByteString n () -> m r) -> m r+withStreamingOutputCommand = withStreamingOutput . shell++--------------------------------------------------------------------------------++-- | Feeds the provided data into the input handle, then concurrently+-- streams stdout and stderr into the provided continuation.+--+-- Note that the monad used in the 'StdOutErr' argument to the+-- continuation can be different from the final result, as it's up+-- to the caller to make sure the result is reached.+withProcessHandles :: (MonadBaseControl IO m, MonadIO m, MonadMask m, MonadBase IO n)+ => ByteString m v+ -> StreamProcess (SupplyStream m)+ (WithStream' m)+ (WithStream' m)+ -> (StdOutErr n () -> m r) -> m r+withProcessHandles inp sp@StreamProcess{..} f =+ snd <$> concurrently withIn withOutErr+ where+ withIn = supplyStream toStdin inp++ withOutErr = withStreamOutputs sp f++-- | Stream input into a process, ignoring any output.+processInput :: (MonadIO m, MonadMask m)+ => StreamProcess (SupplyStream m) ClosedStream ClosedStream+ -> ByteString m r -> m r+processInput StreamProcess{toStdin} = supplyStream toStdin++-- | Read the output from a process, ignoring stdin and stderr.+withProcessOutput :: (MonadIO n, MonadIO m, MonadMask m)+ => StreamProcess ClosedStream (WithStream n m) ClosedStream+ -> (ByteString n () -> m r) -> m r+withProcessOutput StreamProcess{fromStdout} = withStream fromStdout++--------------------------------------------------------------------------------++-- | Represents the input and outputs for a streaming process.+data StreamProcess stdin stdout stderr = StreamProcess+ { toStdin :: !stdin+ , fromStdout :: !stdout+ , fromStderr :: !stderr+ } deriving (Eq, Show)++-- | Switch the two outputs. Useful for example if using+-- 'withStreamProcess' and 'withProcessHandles' but wanting to deal+-- with any potential output from stderr before stdout.+switchOutputs :: StreamProcess stdin stdout stderr+ -> StreamProcess stdin stderr stdout+switchOutputs sp@StreamProcess{fromStdout, fromStderr}+ = sp { fromStdout = fromStderr+ , fromStderr = fromStdout+ }++-- | A variant of 'withCheckedProcess' that will on an exception kill+-- the child process and attempt to perform cleanup (though you+-- should also attempt to do so in your own code).+--+-- Will throw 'ProcessExitedUnsuccessfully' on a non-successful exit code.+--+-- Compared to @withCheckedProcessCleanup@ from @conduit-extra@,+-- this has the three parameters grouped into 'StreamProcess' to+-- make it more of a continuation.+withStreamProcess :: (InputSource stdin, OutputSink stdout, OutputSink stderr+ , MonadIO m, MonadMask m)+ => CreateProcess+ -> (StreamProcess stdin stdout stderr -> m r) -> m r+withStreamProcess cp f = do+ (stdin, stdout, stderr, sph) <- streamingProcess cp+ r <- f (StreamProcess stdin stdout stderr)+ `onException` terminateStreamingProcess sph+ ec <- waitForStreamingProcess sph `finally` closeStreamingProcessHandle sph+ case ec of+ ExitSuccess -> return r+ ExitFailure _ -> throwM (ProcessExitedUnsuccessfully cp ec)++-- | A variant of 'withStreamProcess' that runs the provided+-- command in a shell.+withStreamCommand :: (InputSource stdin, OutputSink stdout, OutputSink stderr+ , MonadIO m, MonadMask m)+ => String+ -> (StreamProcess stdin stdout stderr -> m r) -> m r+withStreamCommand = withStreamProcess . shell++terminateStreamingProcess :: (MonadIO m) => StreamingProcessHandle -> m ()+terminateStreamingProcess = liftIO . terminateProcess . streamingProcessHandleRaw++--------------------------------------------------------------------------------++-- | A representation of the concurrent streaming of both @stdout@ and+-- @stderr@ (contrast to 'SB.hGet').+--+-- Note that if for example you wish to completely discard stderr,+-- you can do so with @'hoist' 'SB.effects'@ (or just process the+-- stdout, then run 'SB.effects' at the end to discard the stderr).+type StdOutErr m r = ByteString (ByteString m) r++-- | Get both stdout and stderr concurrently.+withStreamOutputs :: ( MonadMask m, MonadIO m, MonadBaseControl IO m+ , MonadBase IO n)+ => StreamProcess stdin (WithStream' m) (WithStream' m)+ -> (StdOutErr n () -> m r) -> m r+withStreamOutputs StreamProcess{fromStdout, fromStderr} f =+ withStream fromStdout $ \stdout ->+ withStream fromStderr $ \stderr ->+ let getOut = S.map Left . SB.toChunks $ stdout+ getErr = S.map Right . SB.toChunks $ stderr+ in withMergedStreams unbounded [getOut, getErr] (f . mrg)+ where+ mrg = SB.fromChunks . hoist SB.fromChunks . S.partitionEithers++--------------------------------------------------------------------------------++-- | A wrapper for being able to provide a stream of bytes.+newtype SupplyStream m = SupplyStream { supplyStream :: forall r. ByteString m r -> m r }++instance (MonadMask m, MonadIO m) => InputSource (SupplyStream m) where+ isStdStream = (\(Just h) -> return (SupplyStream $ \inp ->+ SB.hPut h inp `finally` liftIO (hClose h))+ , Just CreatePipe+ )++-- | A wrapper for something taking a continuation with a stream of+-- bytes as input.+newtype WithStream n m = WithStream { withStream :: forall r. (ByteString n () -> m r) -> m r }++-- | An alias for the common case of @n ~ m@.+type WithStream' m = WithStream m m++instance (MonadIO m, MonadMask m, MonadIO n) => OutputSink (WithStream n m) where+ osStdStream = (\(Just h) -> return (WithStream $ \f ->+ f (SB.hGetContents h) `finally` liftIO (hClose h))+ , Just CreatePipe+ )++--------------------------------------------------------------------------------++{- $reexports++All of "Data.Streaming.Process" is available for you to use.++The 'concurrently' function will probably be useful if manually+handling process inputs and outputs.++-}
+ src/Streaming/Process/Lifted.hs view
@@ -0,0 +1,197 @@+{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses, TypeFamilies #-}++{- |+ Module : Streaming.Process.Lifted+ Description : Lifted variants of "Streaming.Process"+ Copyright : (c) Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++ This module defines variants of those in "Streaming.Process" for+ use with the 'Withable' class, found in the @streaming-with@+ package.++ __WARNING:__ If using this module, you will need to have+ @ghc-options -threaded@ in your @.cabal@ file otherwise it will+ likely hang!++ These functions will all throw 'ProcessExitedUnsuccessfully' if the+ process\/command itself fails.++ -}+module Streaming.Process.Lifted+ ( -- * High level functions+ withStreamingProcess+ , withStreamingCommand+ , streamInput+ , streamInputCommand+ , withStreamingOutput+ , withStreamingOutputCommand+ -- * Lower level+ , StreamProcess(..)+ , switchOutputs+ , WithStream(WithStream)+ , WithStream'+ , withStream+ , SupplyStream(SupplyStream)+ , supplyStream+ , withStreamProcess+ , withStreamCommand+ , withProcessHandles+ , processInput+ , withProcessOutput+ -- * Interleaved stdout and stderr+ , StdOutErr+ , withStreamOutputs+ -- * Re-exports+ -- $reexports+ , module Data.Streaming.Process+ , concurrently+ ) where++import Streaming.Process (StdOutErr, StreamProcess(..),+ SupplyStream(SupplyStream),+ WithStream(WithStream), WithStream',+ switchOutputs)+import qualified Streaming.Process as SP++import Data.ByteString.Streaming (ByteString)+import Streaming.With.Lifted (Withable(..))++import Control.Concurrent.Async.Lifted (concurrently)+import Control.Monad.Base (MonadBase)+import Control.Monad.IO.Class (MonadIO)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Streaming.Process+import System.Process (shell)++--------------------------------------------------------------------------------++-- | Feeds the provided data into the specified process, then+-- concurrently streams stdout and stderr into the provided+-- continuation.+--+-- Note that the monad used in the 'StdOutErr' argument to the+-- continuation can be different from the final result, as it's up+-- to the caller to make sure the result is reached.+withStreamingProcess :: ( Withable w, MonadBaseControl IO (WithMonad w)+ , MonadBase IO n)+ => CreateProcess -> ByteString (WithMonad w) v+ -> w (StdOutErr n ())+withStreamingProcess cp inp = liftWith (SP.withStreamingProcess cp inp)++-- | As with 'withStreamingProcess', but run the specified command in+-- a shell.+withStreamingCommand :: ( Withable w, MonadBaseControl IO (WithMonad w)+ , MonadBase IO n)+ => String -> ByteString (WithMonad w) v+ -> w (StdOutErr n ())+withStreamingCommand = withStreamingProcess . shell++-- | Feed input into a process with no expected output.+streamInput :: (Withable w) => CreateProcess+ -> ByteString (WithMonad w) r -> w r+streamInput cp inp = liftAction (SP.streamInput cp inp)++-- | As with 'streamInput' but run the specified command in a shell.+streamInputCommand :: (Withable w) => String+ -> ByteString (WithMonad w) r -> w r+streamInputCommand = streamInput . shell++-- | Obtain the output of a process with no input (ignoring error+-- output).+withStreamingOutput :: (Withable w, MonadIO n)+ => CreateProcess+ -> w (ByteString n ())+withStreamingOutput cp = liftWith (SP.withStreamingOutput cp)++-- | As with 'withStreamingOutput' but run the specified command in a+-- shell.+withStreamingOutputCommand :: (Withable w, MonadIO n)+ => String+ -> w (ByteString n ())+withStreamingOutputCommand = withStreamingOutput . shell++--------------------------------------------------------------------------------++-- | Feeds the provided data into the input handle, then concurrently+-- streams stdout and stderr into the provided continuation.+--+-- Note that the monad used in the 'StdOutErr' argument to the+-- continuation can be different from the final result, as it's up+-- to the caller to make sure the result is reached.+withProcessHandles :: (Withable w, m ~ WithMonad w, MonadBaseControl IO m, MonadBase IO n)+ => ByteString m v+ -> StreamProcess (SupplyStream m)+ (WithStream' m)+ (WithStream' m)+ -> w (StdOutErr n ())+withProcessHandles inp sp = liftWith (SP.withProcessHandles inp sp)++-- | Stream input into a process, ignoring any output.+processInput :: (Withable w)+ => StreamProcess (SupplyStream (WithMonad w)) ClosedStream ClosedStream+ -> ByteString (WithMonad w) r -> w r+processInput sp inp = liftAction (SP.processInput sp inp)++-- | Read the output from a process, ignoring stdin and stderr.+withProcessOutput :: (Withable w, MonadIO n)+ => StreamProcess ClosedStream (WithStream n (WithMonad w)) ClosedStream+ -> w (ByteString n ())+withProcessOutput sp = liftWith (SP.withProcessOutput sp)++--------------------------------------------------------------------------------++-- | A variant of 'withCheckedProcess' that will on an exception kill+-- the child process and attempt to perform cleanup (though you+-- should also attempt to do so in your own code).+--+-- Will throw 'ProcessExitedUnsuccessfully' on a non-successful exit code.+--+-- Compared to @withCheckedProcessCleanup@ from @conduit-extra@,+-- this has the three parameters grouped into 'StreamProcess' to+-- make it more of a continuation.+withStreamProcess :: (InputSource stdin, OutputSink stdout, OutputSink stderr+ , Withable w)+ => CreateProcess -> w (StreamProcess stdin stdout stderr)+withStreamProcess cp = liftWith (SP.withStreamProcess cp)++-- | A variant of 'withStreamProcess' that runs the provided+-- command in a shell.+withStreamCommand :: (InputSource stdin, OutputSink stdout, OutputSink stderr+ , Withable w)+ => String -> w (StreamProcess stdin stdout stderr)+withStreamCommand = withStreamProcess . shell++--------------------------------------------------------------------------------++-- | Get both stdout and stderr concurrently.+withStreamOutputs :: ( Withable w, m ~ WithMonad w, MonadBaseControl IO m+ , MonadBase IO n)+ => StreamProcess stdin (WithStream' m) (WithStream' m)+ -> w (StdOutErr n ())+withStreamOutputs sp = liftWith (SP.withStreamOutputs sp)++--------------------------------------------------------------------------------++-- | Please note that - unlike the version in "Streaming.Process" -+-- this is /not/ a record selector.+supplyStream :: (Withable w) => SupplyStream (WithMonad w)+ -> ByteString (WithMonad w) r -> w r+supplyStream ss inp = liftAction (SP.supplyStream ss inp)++-- | Please note that - unlike the version in "Streaming.Process" -+-- this is /not/ a record selector.+withStream :: (Withable w) => WithStream n (WithMonad w) -> w (ByteString n ())+withStream ws = liftWith (SP.withStream ws)++--------------------------------------------------------------------------------++{- $reexports++All of "Data.Streaming.Process" is available for you to use.++The 'concurrently' function will probably be useful if manually+handling process inputs and outputs.++-}
+ streaming-process.cabal view
@@ -0,0 +1,58 @@+name: streaming-process+version: 0.1.0.0+synopsis: Streaming support for running system process+description:+ Stream data in and out of external commands. Configuration options+ are available to choose which inputs and outputs to use.+license: MIT+license-file: LICENSE+author: Ivan Lazar Miljenovic+maintainer: Ivan.Miljenovic@gmail.com+copyright: Ivan Lazar Miljenovic+category: Data, Streaming+build-type: Simple+extra-source-files: ChangeLog.md, README.md+cabal-version: >=1.10+tested-with: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2,+ GHC == 8.4.1, GHC == 8.5.*++source-repository head+ type: git+ location: https://github.com/ivan-m/streaming-process.git++library+ exposed-modules: Streaming.Process+ , Streaming.Process.Lifted+ build-depends: base == 4.*+ , bytestring+ , directory >= 1.2 && < 1.4+ , exceptions >= 0.6 && < 0.11+ , lifted-async >= 0.9.1 && < 0.11+ , monad-control == 1.*+ , process >= 1.2.0.0 && < 1.7+ , streaming >= 0.1.4.0 && < 0.3+ , streaming-bytestring >= 0.1.4.5 && < 0.2+ , streaming-commons >= 0.1.16 && < 0.3+ , streaming-concurrency == 0.3.*+ , streaming-with == 0.2.*+ , transformers+ , transformers-base+ hs-source-dirs: src+ default-language: Haskell2010+ ghc-options: -Wall++test-suite simple-processes+ type: exitcode-stdio-1.0+ main-is: simple.hs+ other-modules: Paths_streaming_process+ build-depends: streaming-process+ , base+ , bytestring+ , hspec == 2.4.*+ , QuickCheck == 2.*+ , quickcheck-instances+ , streaming+ , streaming-bytestring+ hs-source-dirs: test+ default-language: Haskell2010+ ghc-options: -Wall -threaded
+ test/simple.hs view
@@ -0,0 +1,57 @@+{-# LANGUAGE OverloadedStrings #-}++{- |+ Module : Main+ Description : Simple process testing+ Copyright : (c) Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++ This will probably only run on *nix systems.++ -}+module Main (main) where++import Streaming.Process++import qualified Data.ByteString.Lazy as LB+import Data.ByteString.Streaming (ByteString)+import qualified Data.ByteString.Streaming as B+import Streaming (Of(..))++import Test.Hspec (describe, hspec)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck (Property, ioProperty)+import Test.QuickCheck.Instances ()++--------------------------------------------------------------------------------++main :: IO ()+main = hspec $+ describe "cat" $ do+ prop "stdout only" prop_cat+ prop "stdout and stderr" prop_catBoth++-- Use lazy bytestrings to avoid issues with it being chunked+-- differently.+--+-- But don't take them as arguments+prop_cat :: LB.ByteString -> Property+prop_cat bs = ioProperty (noStdErr "cat" (B.fromLazy bs) (fmap (bs==) . B.toLazy_))++prop_catBoth :: LB.ByteString -> Property+prop_catBoth bs = ioProperty $+ withStreamingCommand "cat | tee /dev/stderr" (B.fromLazy bs) $+ fmap (uncurry' ((&&) . isSame))+ . B.toLazy+ . fmap isSame+ . B.toLazy_+ where+ isSame = (bs==)++noStdErr :: String -> ByteString IO r -> (ByteString IO () -> IO v) -> IO v+noStdErr cmd inp f = withStreamCommand cmd $ \(StreamProcess stdin stdout ClosedStream) ->+ snd <$> concurrently (supplyStream stdin inp) (withStream stdout f)++uncurry' :: (a -> b -> c) -> Of a b -> c+uncurry' f (a :> b) = f a b