streamly-process 0.3.1 → 0.4.0
raw patch · 6 files changed
+241/−61 lines, 6 filesdep ~QuickCheckdep ~streamlydep ~streamly-core
Dependency ranges changed: QuickCheck, streamly, streamly-core, tasty-bench
Files
- CHANGELOG.md +5/−0
- src/DocTestProcess.hs +3/−1
- src/Streamly/Internal/System/Process.hs +186/−48
- src/Streamly/System/Process.hs +13/−1
- streamly-process.cabal +9/−6
- test/Streamly/System/Process.hs +25/−5
CHANGELOG.md view
@@ -1,5 +1,10 @@ # Changelog +## 0.4.0 (Sep 2025)++* Remove buffering from the pipe chunked APIs.+* Add a `toBytesWith` function to modify process configuration+ ## 0.3.1 (Dec 2023) * Allow streamly-0.10.0 and streamly-core-0.2.0
src/DocTestProcess.hs view
@@ -2,6 +2,7 @@ >>> :set -XFlexibleContexts >>> :set -XScopedTypeVariables+>>> :set -Wno-deprecations >>> import Data.Char (toUpper) >>> import Data.Function ((&)) >>> import qualified Streamly.Console.Stdio as Stdio@@ -13,7 +14,8 @@ For APIs that have not been released yet. ->>> import qualified Streamly.Internal.Console.Stdio as Stdio (putChars, putChunks)+>>> import Streamly.Internal.System.IO (defaultChunkSize)+>>> import qualified Streamly.Internal.Console.Stdio as Stdio (putChars, putChunks, readChunks) >>> import qualified Streamly.Internal.FileSystem.Dir as Dir (readFiles) >>> import qualified Streamly.Internal.System.Process as Process >>> import qualified Streamly.Internal.Unicode.Stream as Unicode (lines)
src/Streamly/Internal/System/Process.hs view
@@ -8,6 +8,7 @@ -- {-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE ScopedTypeVariables #-} -- TODO:@@ -42,7 +43,6 @@ -- -- - Replace FilePath with a typed path. ---{-# LANGUAGE FlexibleContexts #-} module Streamly.Internal.System.Process (@@ -77,6 +77,8 @@ , inheritStdin , inheritStdout , pipeStdErr+ , pipeStdin+ , pipeStdout -- * Exceptions , ProcessFailure (..)@@ -84,6 +86,7 @@ -- * Generation -- | stdout of the process is redirected to output stream. , toBytes+ , toBytesWith , toChunks , toChunksWith , toChars@@ -138,7 +141,7 @@ import Streamly.Data.Fold (Fold) import Streamly.Data.Stream.Prelude (MonadAsync, Stream) import System.Exit (ExitCode(..))-import System.IO (hClose, Handle)+import System.IO (hClose, Handle, hSetBuffering, BufferMode(..)) #if !defined(mingw32_HOST_OS) import System.Posix.Types (CUid (..), CGid (..)) #endif@@ -177,6 +180,18 @@ #include "DocTestProcess.hs" -------------------------------------------------------------------------------+-- Compatibility+-------------------------------------------------------------------------------++#if MIN_VERSION_streamly_core(0,3,0)+#define UNFOLD_EACH Stream.unfoldEach+#define CHUNKS_OF Array.chunksOf+#else+#define UNFOLD_EACH Stream.unfoldMany+#define CHUNKS_OF Stream.chunksOf+#endif++------------------------------------------------------------------------------- -- Config ------------------------------------------------------------------------------- @@ -214,6 +229,12 @@ pipeStdErr :: Config -> Config pipeStdErr (Config _) = Config True +pipeStdin :: Config -> Config+pipeStdin (Config _) = Config True++pipeStdout :: Config -> Config+pipeStdout (Config _) = Config True+ inheritStdin :: Config -> Config inheritStdin (Config _) = Config True @@ -453,6 +474,12 @@ pipeStdErr :: Config -> Config pipeStdErr (Config cfg) = Config $ cfg { std_err = CreatePipe } +pipeStdin :: Config -> Config+pipeStdin (Config cfg) = Config $ cfg { std_in = CreatePipe }++pipeStdout :: Config -> Config+pipeStdout (Config cfg) = Config $ cfg { std_out = CreatePipe }+ inheritStdin :: Config -> Config inheritStdin (Config cfg) = Config $ cfg { std_in = Inherit } @@ -485,7 +512,7 @@ ------------------------------------------------------------------------------- -- Transformation ----------------------------------------------------------------------------------+ -- | On normal cleanup we do not need to close the pipe handles as they are -- already guaranteed to be closed (we can assert that) by the time we reach -- here. We should not kill the process, rather wait for it to terminate@@ -518,7 +545,7 @@ -- still hanging around. cleanupException :: (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> IO ()-cleanupException (Just stdinH, Just stdoutH, stderrMaybe, ph) = do+cleanupException (stdinMaybe, stdoutMaybe, stderrMaybe, ph) = do -- Send a SIGTERM to the process #ifdef USE_NATIVE terminate ph@@ -529,8 +556,8 @@ -- Ideally we should be closing the handle without flushing the buffers so -- that we cannot get a SIGPIPE. But there seems to be no way to do that as -- of now so we just ignore the SIGPIPE.- hClose stdinH `catch` eatSIGPIPE- hClose stdoutH+ whenJust (\stdinH -> hClose stdinH `catch` eatSIGPIPE) stdinMaybe+ whenJust hClose stdoutMaybe whenJust hClose stderrMaybe -- Non-blocking wait for the process to go away@@ -553,7 +580,6 @@ _ -> False eatSIGPIPE e = unless (isSIGPIPE e) $ throwIO e-cleanupException _ = error "cleanupProcess: Not reachable" -- | Creates a system process from an executable path and arguments. For the -- default attributes used to create the process see 'mkConfig'.@@ -574,9 +600,16 @@ -- XXX Read the exception channel and reap the process if it failed before -- exec. parent+ hSetBuffering inp NoBuffering+ hSetBuffering out NoBuffering+ hSetBuffering err NoBuffering return (Just inp, Just out, err, proc) #else- createProcess cfg+ r@(inp, out, err, _) <- createProcess cfg+ mapM_ (`hSetBuffering` NoBuffering) inp+ mapM_ (`hSetBuffering` NoBuffering) out+ mapM_ (`hSetBuffering` NoBuffering) err+ return r #endif where@@ -611,6 +644,11 @@ alloc = createProc' modCfg path args +-- Note: It is allowed to inheritStdout or inheritStderr but not both as that+-- would not generate a stream for further processing.+-- inheritStdin has no affect, we always pipe the input stream to the process's+-- stdin+ -- | Like 'pipeChunksEither' but use the specified configuration to run the -- process. {-# INLINE pipeChunksEitherWith #-}@@ -622,10 +660,12 @@ -> Stream m (Array Word8) -- ^ Input stream -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output stream pipeChunksEitherWith modifier path args input =- pipeChunksWithAction run (modifier . pipeStdErr) path args+ pipeChunksWithAction run (pipeStdin . modifier . pipeStdErr) path args where + run (_, Nothing, Nothing, _) =+ error "pipeChunksEitherWith: only one of stdout or stderr can be inherited" run (Just stdinH, Just stdoutH, Just stderrH, _) = putChunksClose stdinH input `parallel` fmap Left (toChunksClose stderrH)@@ -634,6 +674,11 @@ -- | Like 'pipeChunks' but also includes stderr as 'Left' stream in the -- 'Either' output.+--+-- Definition:+--+-- >>> pipeChunksEither = pipeChunksEitherWith id+-- {-# INLINE pipeChunksEither #-} pipeChunksEither :: (MonadCatch m, MonadAsync m)@@ -643,11 +688,14 @@ -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output stream pipeChunksEither = pipeChunksEitherWith id --- | @pipeBytesEither path args input@ runs the executable at @path@ using @args@--- as arguments and @input@ stream as its standard input. The error stream of--- the executable is presented as 'Left' values in the resulting stream and--- output stream as 'Right' values.+-- | @pipeBytesEither path args input@ runs the executable at @path@ using+-- @args@ as arguments and @input@ stream as its standard input. The error+-- stream of the executable is presented as 'Left' values in the resulting+-- stream and output stream as 'Right' values. The input to the pipe is+-- buffered with a buffer size of 'defaultChunkSize'. --+-- For control over the buffer use your own chunking and chunk based APIs.+-- -- Raises 'ProcessFailure' exception in case of failure. -- -- For example, the following is equivalent to the shell command @echo "hello@@ -671,12 +719,16 @@ -> Stream m Word8 -- ^ Input Stream -> Stream m (Either Word8 Word8) -- ^ Output Stream pipeBytesEither path args input =- let input1 = Stream.chunksOf defaultChunkSize input+ let input1 = CHUNKS_OF defaultChunkSize input output = pipeChunksEither path args input1 leftRdr = fmap Left Array.reader rightRdr = fmap Right Array.reader- in Stream.unfoldMany (Unfold.either leftRdr rightRdr) output+ in UNFOLD_EACH (Unfold.either leftRdr rightRdr) output +-- Note: inheritStdin, inheritStdout have no affect, we always pipe+-- the input stream to the process's stdin and pipe the stdout to the+-- resulting stream+ -- | Like 'pipeChunks' but use the specified configuration to run the process. {-# INLINE pipeChunksWith #-} pipeChunksWith ::@@ -687,17 +739,19 @@ -> Stream m (Array Word8) -- ^ Input stream -> Stream m (Array Word8) -- ^ Output stream pipeChunksWith modifier path args input =- pipeChunksWithAction run modifier path args+ pipeChunksWithAction run (pipeStdout . pipeStdin . modifier) path args where run (Just stdinH, Just stdoutH, _, _) =- putChunksClose stdinH input `parallel` toChunksClose stdoutH- run _ = error "pipeChunksWith: Not reachable"+ putChunksClose stdinH input+ `parallel` toChunksClose stdoutH+ run _ = error "pipeChunksWith: unreachable" -- | @pipeChunks file args input@ runs the executable @file@ specified by -- its name or path using @args@ as arguments and @input@ stream as its--- standard input. Returns the standard output of the executable as a stream.+-- standard input. Returns the standard output of the process as a stream+-- of chunks of bytes (Array Word8). -- -- If only the name of an executable file is specified instead of its path then -- the file name is searched in the directories specified by the PATH@@ -719,6 +773,10 @@ -- :} --HELLO WORLD --+-- Definition:+--+-- >>> pipeChunks = pipeChunksWith id+-- -- /pre-release/ {-# INLINE pipeChunks #-} pipeChunks ::@@ -739,9 +797,13 @@ -> Stream m (Array Word8) -- ^ Output stream processChunks = pipeChunks --- | Like 'pipeChunks' except that it works on a stream of bytes instead of--- a stream of chunks.+-- | Like 'pipeChunks' except that its input and output is stream of bytes+-- instead of a stream of chunks. The input to the pipe is buffered with a+-- buffer size of 'defaultChunkSize'. --+-- For control over the input buffer use your own chunking and chunk based+-- APIs.+-- -- We can write the example in 'pipeChunks' as follows. -- -- >>> :{@@ -760,9 +822,9 @@ -> Stream m Word8 -- ^ Input Stream -> Stream m Word8 -- ^ Output Stream pipeBytes path args input = -- rights . pipeBytesEither path args- let input1 = Stream.chunksOf defaultChunkSize input+ let input1 = CHUNKS_OF defaultChunkSize input output = pipeChunks path args input1- in Stream.unfoldMany Array.reader output+ in UNFOLD_EACH Array.reader output {-# DEPRECATED processBytes "Please use pipeBytes instead." #-} {-# INLINE processBytes #-}@@ -774,9 +836,18 @@ -> Stream m Word8 -- ^ Output Stream processBytes = pipeBytes --- | Like 'pipeChunks' except that it works on a stream of chars instead of--- a stream of chunks.+whenJustS :: Applicative m => (a -> Stream m b) -> Maybe a -> Stream m b+whenJustS action mb = maybe Stream.nil action mb++-- | Like 'pipeChunks' except that its input and output is stream of chars+-- instead of a stream of chunks. The input to the pipe is buffered with a+-- buffer size of 'defaultChunkSize'. --+-- For control over the input buffer use your own chunking and chunk based+-- APIs.+--+-- NOTE: This API uses UTF-8 encoding.+-- -- >>> :{ -- Process.toChars "echo" ["hello world"] -- & Process.pipeChars "tr" ["[a-z]", "[A-Z]"]@@ -811,6 +882,10 @@ -- Generation ------------------------------------------------------------------------------- +-- Note: It is allowed to inheritStdout or inheritStderr but not both as that+-- would not generate a stream for further processing and would result in unintuitive+-- behaviour+ -- | Like 'toChunksEither' but use the specified configuration to run the -- process. {-# INLINE toChunksEitherWith #-}@@ -821,15 +896,19 @@ -> [String] -- ^ Arguments -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output stream toChunksEitherWith modifier path args =- pipeChunksWithAction run (modifier . inheritStdin . pipeStdErr) path args+ pipeChunksWithAction run (modifier . inheritStdin . pipeStdErr . pipeStdout) path args where - run (_, Just stdoutH, Just stderrH, _) =- fmap Left (toChunksClose stderrH)- `parallel` fmap Right (toChunksClose stdoutH)- run _ = error "toChunksEitherWith: Not reachable"+ run (_, Nothing, Nothing, _) =+ error "toChunksEitherWith: only one of stdout or stderr can be inherited"+ run (_, stdoutMaybe, stderrMaybe, _) =+ fmap Left (whenJustS toChunksClose stderrMaybe)+ `parallel` fmap Right (whenJustS toChunksClose stdoutMaybe) +-- Note: inheritStdout has no affect, we always pipe stdout to the resulting+-- stream+ -- | Like 'toChunks' but use the specified configuration to run the process. {-# INLINE toChunksWith #-} toChunksWith ::@@ -839,16 +918,17 @@ -> [String] -- ^ Arguments -> Stream m (Array Word8) -- ^ Output stream toChunksWith modifier path args =- pipeChunksWithAction run (modifier . inheritStdin) path args+ pipeChunksWithAction run (pipeStdout . modifier . inheritStdin) path args where run (_, Just stdoutH, _, _) = toChunksClose stdoutH- run _ = error "toChunksWith: Not reachable"+ run _ = error "toChunksWith: Not reachable" -- | @toBytesEither path args@ runs the executable at @path@ using @args@ as--- arguments and returns a stream of 'Either' bytes. The 'Left' values are from--- @stderr@ and the 'Right' values are from @stdout@ of the executable.+-- arguments and returns the output of the process as a stream of 'Either'+-- bytes. The 'Left' values are from @stderr@ and the 'Right' values are from+-- @stdout@ of the executable. -- -- Raises 'ProcessFailure' exception in case of failure. --@@ -875,11 +955,15 @@ let output = toChunksEither path args leftRdr = fmap Left Array.reader rightRdr = fmap Right Array.reader- in Stream.unfoldMany (Unfold.either leftRdr rightRdr) output+ in UNFOLD_EACH (Unfold.either leftRdr rightRdr) output --- | The following code is equivalent to the shell command @echo "hello--- world"@:+-- | @toBytes path args@ runs the executable specified by @path@ using @args@+-- as arguments and returns the output of the process as a stream of bytes. --+-- Raises 'ProcessFailure' exception in case of failure.+--+-- The following code is equivalent to the shell command @echo "hello world"@:+-- -- >>> :{ -- Process.toBytes "echo" ["hello world"] -- & Stream.fold Stdio.write@@ -895,8 +979,22 @@ -> Stream m Word8 -- ^ Output Stream toBytes path args = let output = toChunks path args- in Stream.unfoldMany Array.reader output+ in UNFOLD_EACH Array.reader output +-- | Like 'toBytes' but use the specified configuration to run the process.+--+-- @since 0.3.2+{-# INLINE toBytesWith #-}+toBytesWith ::+ (MonadAsync m, MonadCatch m)+ => (Config -> Config)+ -> FilePath -- ^ Executable name or path+ -> [String] -- ^ Arguments+ -> Stream m Word8 -- ^ Output Stream+toBytesWith modCfg path args =+ let output = toChunksWith modCfg path args+ in UNFOLD_EACH Array.reader output+ -- | Like 'toBytesEither' but generates a stream of @Array Word8@ instead of a stream -- of @Word8@. --@@ -921,9 +1019,14 @@ -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output Stream toChunksEither = toChunksEitherWith id --- | The following code is equivalent to the shell command @echo "hello--- world"@:+-- | @toChunks path args@ runs the executable specified by @path@ using @args@+-- as arguments and returns the output of the process as a stream of chunks of+-- bytes (Array Word8). --+-- Raises 'ProcessFailure' exception in case of failure.+--+-- The following code is equivalent to the shell command @echo "hello world"@:+-- -- >>> :{ -- Process.toChunks "echo" ["hello world"] -- & Stream.fold Stdio.writeChunks@@ -941,7 +1044,15 @@ -> Stream m (Array Word8) -- ^ Output Stream toChunks = toChunksWith id --- |+-- | @toChars path args@ runs the executable specified by @path@ using @args@+-- as arguments and returns the output of the process as a stream of chars.+--+-- NOTE: This API uses UTF-8 encoding.+--+-- Raises 'ProcessFailure' exception in case of failure.+--+-- Definition:+-- -- >>> toChars path args = toBytes path args & Unicode.decodeUtf8 -- {-# INLINE toChars #-}@@ -952,9 +1063,22 @@ -> Stream m Char -- ^ Output Stream toChars path args = toBytes path args & Unicode.decodeUtf8 --- |--- >>> toLines path args f = toChars path args & Unicode.lines f+-- | @toLines f path args@ runs the executable specified by @path@ using @args@+-- as arguments and folds the output of the process at line breaks, using the+-- fold @f@, to return a stream of folded lines. --+-- NOTE: This API uses UTF-8 encoding.+--+-- Raises 'ProcessFailure' exception in case of failure.+--+-- To return a stream of lines as strings:+--+-- >>> toStrings = toLines Fold.toList+--+-- Definition:+--+-- >>> toLines f path args = toChars path args & Unicode.lines f+-- {-# INLINE toLines #-} toLines :: (MonadAsync m, MonadCatch m)@@ -964,18 +1088,28 @@ -> Stream m a -- ^ Output Stream toLines f path args = toChars path args & Unicode.lines f --- |--- >>> toString path args = toChars path args & Stream.fold Fold.toList+-- | @toString path args@ runs the executable specified by @path@ using @args@+-- as arguments and folds the entire output of the process as a single string. --+-- NOTE: This API uses UTF-8 encoding.+--+-- Definition:+--+-- >>> toString path args = toChars path args & Stream.toList+-- {-# INLINE toString #-} toString :: (MonadAsync m, MonadCatch m) => FilePath -- ^ Executable name or path -> [String] -- ^ Arguments -> m String-toString path args = toChars path args & Stream.fold Fold.toList+toString path args = toChars path args & Stream.toList --- |+-- | @toStdout path args@ runs the executable specified by @path@ using @args@+-- as arguments and returns the output of the process on stdout.+--+-- Definition:+-- -- >>> toStdout path args = toChunks path args & Stdio.putChunks -- {-# INLINE toStdout #-}@@ -992,7 +1126,11 @@ return () -} --- |+-- | @toNull path args@ runs the executable specified by @path@ using @args@+-- as arguments and discards the output of the process.+--+-- Definition:+-- -- >>> toNull path args = toChunks path args & Stream.fold Fold.drain -- {-# INLINE toNull #-}
src/Streamly/System/Process.hs view
@@ -38,7 +38,9 @@ -- -- >>> :{ -- Process.toBytes "echo" ["hello world"]--- & Unicode.decodeLatin1 & fmap toUpper & Unicode.encodeLatin1+-- & Unicode.decodeLatin1+-- & fmap toUpper+-- & Unicode.encodeLatin1 -- & Stream.fold Stdio.write -- :} -- HELLO WORLD@@ -74,6 +76,15 @@ -- & Stream.fold Stdio.writeChunks -- :} --+-- = Running Interactive Programs (e.g. ghci)+--+-- >>> :{+-- ghci =+-- Stdio.readChunks+-- & Process.pipeChunks "ghci" []+-- & Stdio.putChunks+-- :}+-- -- = Experimental APIs -- -- See "Streamly.Internal.System.Process" for unreleased functions.@@ -121,6 +132,7 @@ , toChunks , toChunksWith , toBytes+ , toBytesWith , toChars , toLines
streamly-process.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: streamly-process-version: 0.3.1+version: 0.4.0 synopsis: Use OS processes as stream transformation functions description: Use operating system (OS) commands in Haskell programs as if they were@@ -18,7 +18,10 @@ copyright: Composewell Technologies category: Streamly, Streaming, System stability: Experimental-tested-with: GHC==9.4.4+tested-with: GHC==9.10.1+ , GHC==9.8.1+ , GHC==9.6.3+ , GHC==9.4.4 , GHC==9.2.7 , GHC==9.0.2 , GHC==8.10.7@@ -97,8 +100,8 @@ base >= 4.8 && < 5 , exceptions >= 0.8 && < 0.11 -- Uses internal APIs- , streamly >= 0.9 && < 0.10- , streamly-core >= 0.1 && < 0.2+ , streamly >= 0.9 && < 0.12+ , streamly-core >= 0.1 && < 0.4 if !flag(use-native) build-depends: process >= 1.0 && < 1.7 else@@ -127,7 +130,7 @@ , directory >= 1.2.2 && < 1.4 -- Uses internal APIs , streamly-core- , tasty-bench >= 0.2.5 && < 0.4+ , tasty-bench >= 0.2.5 && < 0.5 if flag(fusion-plugin) && !impl(ghc < 8.6) build-depends:@@ -147,6 +150,6 @@ , directory >= 1.2.2 && < 1.4 , exceptions >= 0.8 && < 0.11 , hspec >= 2.0 && < 3- , QuickCheck >= 2.10 && < 2.15+ , QuickCheck >= 2.10 && < 2.16 -- Uses internal APIs , streamly-core
test/Streamly/System/Process.hs view
@@ -2,6 +2,10 @@ module Main (main) where +-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------+ import Data.Function ((&)) import Data.List ((\\)) import Data.Maybe (fromMaybe)@@ -36,6 +40,22 @@ import qualified Streamly.Internal.FileSystem.Handle as FH (putBytes, read) import qualified Streamly.Internal.System.Command as Cmd (quotedWord) +-------------------------------------------------------------------------------+-- Compatibility+-------------------------------------------------------------------------------++#if MIN_VERSION_streamly_core(0,3,0)+#define UNFOLD_EACH Stream.unfoldEach+#define CHUNKS_OF Array.chunksOf+#else+#define UNFOLD_EACH Stream.unfoldMany+#define CHUNKS_OF Stream.chunksOf+#endif++-------------------------------------------------------------------------------+-- Tests+-------------------------------------------------------------------------------+ newtype SimpleError = SimpleError String deriving Show @@ -172,7 +192,7 @@ {-# INLINE concatArr #-} concatArr :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a-concatArr = S.unfoldMany Array.reader+concatArr = UNFOLD_EACH Array.reader toChunks1 :: Property toChunks1 =@@ -270,7 +290,7 @@ Proc.pipeChunks trBinary ["[a-z]", "[A-Z]"]- (S.chunksOf arrayChunkSize inputStream)+ (CHUNKS_OF arrayChunkSize inputStream) charUpperStrm = fmap toUpper inputStream @@ -290,7 +310,7 @@ Proc.pipeChunks path ["-n", "1"]- (S.chunksOf arrayChunkSize inputStream)+ (CHUNKS_OF arrayChunkSize inputStream) headLine = S.foldMany@@ -409,7 +429,7 @@ Proc.pipeChunksEither trBinary ["[a-z]", "[A-Z]"]- (S.chunksOf arrayChunkSize inputStream)+ (CHUNKS_OF arrayChunkSize inputStream) charUpperStrm = fmap toUpper inputStream @@ -427,7 +447,7 @@ Proc.pipeChunksEither interpreterFile [interpreterArg, executableFile, "[a-z]", "[A-Z]"]- (S.chunksOf arrayChunkSize inputStream)+ (CHUNKS_OF arrayChunkSize inputStream) charUpperStrm = fmap toUpper inputStream