streamly-process 0.2.0.1 → 0.3.0

raw patch · 10 files changed

+1514/−450 lines, 10 filesdep +streamly-coredep −gaugedep ~streamly

Dependencies added: streamly-core

Dependencies removed: gauge

Dependency ranges changed: streamly

Files

• Benchmark/System/Process.hs +48/−52
• CHANGELOG.md +9/−0
• README.md +72/−3
• src/Streamly/Internal/System/Command.hs +318/−0
• src/Streamly/Internal/System/Process.hs +629/−143
• src/Streamly/Internal/System/Process/Posix.hs +6/−0
• src/Streamly/System/Command.hs +158/−0
• src/Streamly/System/Process.hs +89/−76
• streamly-process.cabal +45/−46
• test/Streamly/System/Process.hs +140/−130

@@ -1,11 +1,12 @@ {-# LANGUAGE  ScopedTypeVariables #-}+{-# OPTIONS_GHC -fspec-constr-recursive=4 #-}  module Main (main) where  import Control.Exception (finally) import Data.Either (isRight, fromRight, isLeft, fromLeft) import Data.Word (Word8)-import Gauge (defaultMain, bench, nfIO)+import Test.Tasty.Bench (defaultMain, bench, nfIO) import System.Directory (removeFile, findExecutable) import System.IO     ( Handle@@ -13,25 +14,24 @@     , openFile     , hClose     )-import System.Process (proc, createProcess, waitForProcess, callCommand)+import Streamly.Data.Stream (Stream)  import qualified Streamly.Data.Fold as FL-import qualified Streamly.FileSystem.Handle as FH-import qualified Streamly.Prelude as S+import qualified Streamly.Data.Stream as S import qualified Streamly.System.Process as Proc+import qualified Streamly.Internal.System.Command as Cmd  -- Internal imports-import qualified Streamly.Internal.FileSystem.Handle-    as FH (toBytes, toChunks, putBytes, putChunks)+import qualified Streamly.Internal.FileSystem.Handle as FH import qualified Streamly.Internal.System.Process as Proc  -- XXX replace with streamly versions once they are fixed {-# INLINE rights #-}-rights :: (S.IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b+rights :: Monad m => Stream m (Either a b) -> Stream m b rights = fmap (fromRight undefined) . S.filter isRight  {-# INLINE lefts #-}-lefts :: (S.IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a+lefts :: Monad m => Stream m (Either a b) -> Stream m a lefts = fmap (fromLeft undefined) . S.filter isLeft  -------------------------------------------------------------------------------@@ -71,16 +71,12 @@ generateByteFile :: IO () generateByteFile = do     ddPath <- which "dd"-    let procObj = proc ddPath [-                "if=" ++ devRandom,-                "of=" ++ largeByteFile,-                "count=" ++ show ddBlockCount,-                "bs=" ++ show ddBlockSize-            ]--    (_, _, _, procHandle) <- createProcess procObj-    _ <- waitForProcess procHandle-    return ()+    Cmd.toStdout+        $ ddPath+            ++ " if=" ++ devRandom+            ++ " of=" ++ largeByteFile+            ++ " count=" ++ show ddBlockCount+            ++ " bs=" ++ show ddBlockSize  ------------------------------------------------------------------------------- -- Create a file filled with ascii chars@@ -107,12 +103,12 @@  trToStderrContent :: String trToStderrContent =-    "tr [a-z] [A-Z] <&0 >&2"+    "#!/bin/sh\ntr [a-z] [A-Z] <&0 >&2"  createExecutable :: IO () createExecutable = do     writeFile trToStderr trToStderrContent-    callCommand ("chmod +x " ++ trToStderr)+    Cmd.toStdout ("chmod +x " ++ trToStderr)  ------------------------------------------------------------------------------- -- Create and delete the temp data/exec files@@ -134,36 +130,36 @@ -- Benchmark functions ------------------------------------------------------------------------------- -toBytes' :: String-> Handle -> IO ()-toBytes' catPath outH =+toBytesEither :: String-> Handle -> IO ()+toBytesEither catPath outH =     FH.putBytes outH         $ rights-        $ Proc.toBytes' catPath [largeByteFile]+        $ Proc.toBytesEither catPath [largeByteFile] -toChunks' :: String -> Handle -> IO ()-toChunks' catPath hdl =+toChunksEither :: String -> Handle -> IO ()+toChunksEither catPath hdl =     FH.putChunks hdl         $ rights-        $ Proc.toChunks' catPath [largeByteFile]+        $ Proc.toChunksEither catPath [largeByteFile] -processBytes' :: String-> Handle -> IO ()-processBytes' trPath outputHdl = do+pipeBytesEither :: String-> Handle -> IO ()+pipeBytesEither trPath outputHdl = do     inputHdl <- openFile largeCharFile ReadMode     _ <- S.fold (FL.partition (FH.write outputHdl) (FH.write outputHdl))-        $ Proc.processBytes'+        $ Proc.pipeBytesEither             trPath             ["[a-z]", "[A-Z]"]-        $ FH.toBytes inputHdl+        $ FH.read inputHdl     hClose inputHdl -processBytes :: String-> Handle -> IO ()-processBytes trPath outputHdl = do+pipeBytes :: String-> Handle -> IO ()+pipeBytes trPath outputHdl = do     inputHdl <- openFile largeCharFile ReadMode     FH.putBytes outputHdl-        $ Proc.processBytes+        $ Proc.pipeBytes             trPath             ["[a-z]", "[A-Z]"]-        $ FH.toBytes inputHdl+        $ FH.read inputHdl     hClose inputHdl  processBytesToStderr :: Handle -> IO ()@@ -171,33 +167,33 @@     inputHdl <- openFile largeCharFile ReadMode     FH.putBytes outputHdl         $ lefts-        $ Proc.processBytes'+        $ Proc.pipeBytesEither             trToStderr             ["[a-z]", "[A-Z]"]-        $ FH.toBytes inputHdl+        $ FH.read inputHdl     hClose inputHdl -processChunks :: String -> Handle -> IO ()-processChunks trPath outputHdl = do+pipeChunks :: String -> Handle -> IO ()+pipeChunks trPath outputHdl = do     inputHdl <- openFile largeCharFile ReadMode     FH.putChunks outputHdl $-        Proc.processChunks+        Proc.pipeChunks             trPath             ["[a-z]", "[A-Z]"]-        $ FH.toChunks inputHdl+        $ FH.readChunks inputHdl     hClose inputHdl -processChunks' :: String -> Handle -> IO ()-processChunks' trPath outputHdl = do+pipeChunksEither :: String -> Handle -> IO ()+pipeChunksEither trPath outputHdl = do     inputHdl <- openFile largeCharFile ReadMode     _ <- S.fold             (FL.partition                 (FH.writeChunks outputHdl) (FH.writeChunks outputHdl)             )-        $ Proc.processChunks'+        $ Proc.pipeChunksEither             trPath             ["[a-z]", "[A-Z]"]-            (FH.toChunks inputHdl)+            (FH.readChunks inputHdl)     hClose inputHdl  processChunksToStderr :: Handle -> IO ()@@ -205,10 +201,10 @@     inputHdl <- openFile largeCharFile ReadMode     FH.putChunks outputHdl         $ lefts-        $ Proc.processChunks'+        $ Proc.pipeChunksEither             trToStderr             ["[a-z]", "[A-Z]"]-            (FH.toChunks inputHdl)+            (FH.readChunks inputHdl)     hClose inputHdl  -------------------------------------------------------------------------------@@ -225,13 +221,13 @@     putStrLn "Running benchmarks..."      defaultMain-        [ bench "toBytes'" $ nfIO $ toBytes' catPath nullH-        , bench "toChunks'" $ nfIO $ toChunks' catPath nullH-        , bench "processBytes tr" $ nfIO $ processBytes trPath nullH-        , bench "processBytes' tr" $ nfIO $ processBytes' trPath nullH+        [ bench "toBytesEither" $ nfIO $ toBytesEither catPath nullH+        , bench "toChunksEither" $ nfIO $ toChunksEither catPath nullH+        , bench "pipeBytes tr" $ nfIO $ pipeBytes trPath nullH+        , bench "pipeBytesEither tr" $ nfIO $ pipeBytesEither trPath nullH         , bench "processBytesToStderr tr" $ nfIO $ processBytesToStderr nullH-        , bench "processChunks tr" $ nfIO (processChunks trPath nullH)-        , bench "processChunks' tr" $ nfIO (processChunks' trPath nullH)+        , bench "pipeChunks tr" $ nfIO (pipeChunks trPath nullH)+        , bench "pipeChunksEither tr" $ nfIO (pipeChunksEither trPath nullH)         , bench "processChunksToStderr" $ nfIO $ processChunksToStderr nullH         ] `finally` (do             putStrLn "cleanup ..."

@@ -1,5 +1,14 @@ # Changelog +## 0.3.0 (Apr 2023)++* Added a `Streamly.System.Command` module+* `toChunks` and `toBytes` now make the stdin available to the process being+  executed.+* Allow streamly 0.9.0+* Signature changes - removed the IsStream constraint, now all APIs use the+  monomorphic `Stream` type.+ ## 0.2.0.1 (Mar 2022)  * Fix the test suite.

@@ -1,8 +1,77 @@ # OS processes as streams -Run operating system processes as stream source, sink or transformation-functions. Use them seamlessly in a streaming data pipeline in the same way-as any other Haskell functions.+Use operating system (OS) commands in Haskell programs as if they were+native Haskell functions, by treating their inputs and outputs as+Haskell streams. This allows you to write high-level Haskell scripts+that can perform tasks similar to shell scripts, but with C-like+performance, and with strong safety guarantees, refactorability, and+modularity.++Use the following imports in the examples below:++```haskell+>>> :set -XFlexibleContexts+>>> :set -XScopedTypeVariables+>>> :set -XQuasiQuotes+>>> import Data.Function ((&))+>>> import Streamly.Unicode.String (str)+>>> import qualified Streamly.Data.Array as Array+>>> import qualified Streamly.Console.Stdio as Stdio+>>> import qualified Streamly.Data.Fold as Fold+>>> import qualified Streamly.Data.Stream.Prelude as Stream+>>> import qualified Streamly.System.Command as Command+>>> import qualified Streamly.Internal.FileSystem.Dir as Dir+```++## Commands as streaming functions++The shell command `echo "hello world" | tr [a-z] [A-Z]` can be written as+follows using this package:++```haskell+>>> :{+   Command.toBytes [str|echo "hello world"|] -- Stream IO Word8+ & Command.pipeBytes [str|tr [a-z] [A-Z]|]   -- Stream IO Word8+ & Stream.fold Stdio.write                   -- IO ()+ :}+ HELLO WORLD+```++## Shell commands as streaming functions++If you want to execute the same command using the shell pipe syntax:++```haskell+>>> :{+   Command.toBytes [str|sh "-c" "echo 'hello world' | tr [a-z] [A-Z]"|] -- Stream IO Word8+ & Stream.fold Stdio.write                                              -- IO ()+ :}+ HELLO WORLD+```++## Running Commands Concurrently++You can run commands concurrently by using streamly's concurrent combinators.++Running @grep@ concurrently on many files:++```haskell+>>> :{+grep file =+   Command.toBytes [str|grep -H "pattern" #{file}|]             -- Stream IO Word8+ & Stream.handle (\(_ :: Command.ProcessFailure) -> Stream.nil) -- Stream IO Word8+ & Stream.foldMany (Fold.takeEndBy (== 10) Array.write)         -- Stream IO (Array Word8)+ :}++>>> :{+pgrep =+   Dir.readFiles "."             -- Stream IO FilePath+ & Stream.parConcatMap id grep   -- Stream IO (Array Word8)+ & Stream.fold Stdio.writeChunks -- IO ()+:}+```++## Streamly  Please visit [Streamly homepage](https://streamly.composewell.com) for more details about streamly.

@@ -0,0 +1,318 @@+-- |+-- Module      : Streamly.Internal.System.Command+-- Copyright   : (c) 2022 Composewell Technologies+-- License     : Apache-2.0+-- Maintainer  : streamly@composewell.com+-- Stability   : experimental+-- Portability : GHC+--+{-# LANGUAGE CPP #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}++module Streamly.Internal.System.Command+    (+    -- * Generation+      toBytes+    , toChunks+    , toChars+    , toLines++    -- * Effects+    , toString+    , toStdout+    , toNull++    -- * Transformation+    , pipeBytes+    , pipeChars+    , pipeChunks++    -- * Helpers+    , runWith+    , streamWith+    , pipeWith+    )+where++import Control.Monad.Catch (MonadCatch)+import Data.Char (isSpace)+import Data.Word (Word8)+import Streamly.Data.Array (Array)+import Streamly.Data.Fold (Fold)+import Streamly.Data.Parser (Parser)+import Streamly.Data.Stream.Prelude (MonadAsync, Stream)++import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Parser as Parser+import qualified Streamly.Data.Stream.Prelude as Stream+import qualified Streamly.Internal.System.Process as Process++-- Keep it synced with the released module++-- $setup+-- >>> :set -XFlexibleContexts+-- >>> :set -XQuasiQuotes+-- >>> import Data.Char (toUpper)+-- >>> import Data.Function ((&))+-- >>> import Streamly.Unicode.String (str)+-- >>> import qualified Streamly.Data.Array as Array+-- >>> import qualified Streamly.Console.Stdio as Stdio+-- >>> import qualified Streamly.Data.Fold as Fold+-- >>> import qualified Streamly.Data.Stream.Prelude as Stream+-- >>> import qualified Streamly.System.Command as Command+-- >>> import qualified Streamly.Unicode.Stream as Unicode+--+-- >>> import qualified Streamly.Internal.System.Process as Process+-- >>> import qualified Streamly.Internal.Console.Stdio as Stdio+-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir++{-# INLINE quotedWord #-}+quotedWord :: MonadCatch m => Parser Char m String+quotedWord =+    let toRQuote x =+            case x of+                '"' -> Just x+                '\'' -> Just x+                _ -> Nothing+        trEsc q x = if q == x then Just x else Nothing+     in Parser.wordWithQuotes False trEsc '\\' toRQuote isSpace Fold.toList++-- | A modifier for stream generation APIs in "Streamly.System.Process" to+-- generate streams from command strings.+--+-- For example:+--+-- >>> streamWith Process.toBytes "echo hello" & Stdio.putBytes+-- hello+-- >>> streamWith Process.toChunks "echo hello" & Stdio.putChunks+-- hello+--+-- /Internal/+{-# INLINE streamWith #-}+streamWith :: MonadCatch m =>+    (FilePath -> [String] -> Stream m a) -> String -> Stream m a+streamWith f cmd =+    Stream.concatEffect $ do+        xs <- Stream.fold Fold.toList+                $ Stream.catRights+                $ Stream.parseMany quotedWord+                $ Stream.fromList cmd+        case xs of+            y:ys -> return $ f y ys+            _ -> error "streamWith: empty command"++-- | A modifier for process running APIs in "Streamly.System.Process" to run+-- command strings.+--+-- For example:+--+-- >>> runWith Process.toString "echo hello"+-- "hello\n"+-- >>> runWith Process.toStdout "echo hello"+-- hello+--+-- /Internal/+{-# INLINE runWith #-}+runWith :: MonadCatch m =>+    (FilePath -> [String] -> m a) -> String -> m a+runWith f cmd = do+    xs <- Stream.fold Fold.toList+            $ Stream.catRights+            $ Stream.parseMany quotedWord+            $ Stream.fromList cmd+    case xs of+        y:ys -> f y ys+        _ -> error "streamWith: empty command"++-- | A modifier for process piping APIs in "Streamly.System.Process" to pipe+-- data through processes specified by command strings.+--+-- For example:+--+-- >>> :{+--    toChunks "echo hello"+--  & pipeWith Process.pipeChunks "tr [a-z] [A-Z]"+--  & Stdio.putChunks+--  :}+--HELLO+--+-- /Internal/+pipeWith :: MonadCatch m =>+       (FilePath -> [String] -> Stream m a -> Stream m b)+    -> String+    -> Stream m a+    -> Stream m b+pipeWith f cmd input =+    Stream.concatEffect $ do+        xs <- Stream.fold Fold.toList+                $ Stream.catRights+                $ Stream.parseMany quotedWord+                $ Stream.fromList cmd+        case xs of+            y:ys -> return $ f y ys input+            _ -> error "streamWith: empty command"+++-- | @pipeChunks command input@ runs the executable with arguments specified by+-- @command@ and supplying @input@ stream as its standard input.  Returns the+-- standard output of the executable as a stream of byte arrays.+--+-- 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+-- environment variable.+--+-- If the input stream throws an exception or if the output stream is garbage+-- collected before it could finish then the process is terminated with SIGTERM.+--+-- If the process terminates with a non-zero exit code then a 'ProcessFailure'+-- exception is raised.+--+-- The following code is equivalent to the shell command @echo "hello world" |+-- tr [a-z] [A-Z]@:+--+-- >>> :{+--    toChunks "echo hello world"+--  & pipeChunks "tr [a-z] [A-Z]"+--  & Stdio.putChunks+--  :}+--HELLO WORLD+--+-- /Pre-release/+{-# INLINE pipeChunks #-}+pipeChunks :: (MonadAsync m, MonadCatch m) =>+    String -> Stream m (Array Word8) -> Stream m (Array Word8)+pipeChunks = pipeWith Process.pipeChunks++-- | Like 'pipeChunks' except that it works on a stream of bytes instead of+-- a stream of chunks.+--+-- >>> :{+--    toBytes "echo hello world"+--  & pipeBytes "tr [a-z] [A-Z]"+--  & Stdio.putBytes+--  :}+--HELLO WORLD+--+-- /Pre-release/+{-# INLINE pipeBytes #-}+pipeBytes :: (MonadAsync m, MonadCatch m) =>+    String -> Stream m Word8 -> Stream m Word8+pipeBytes = pipeWith Process.pipeBytes++-- | Like 'pipeChunks' except that it works on a stream of chars instead of+-- a stream of chunks.+--+-- >>> :{+--    toChars "echo hello world"+--  & pipeChars "tr [a-z] [A-Z]"+--  & Stdio.putChars+--  :}+--HELLO WORLD+--+-- /Pre-release/+{-# INLINE pipeChars #-}+pipeChars :: (MonadAsync m, MonadCatch m) =>+    String -> Stream m Char -> Stream m Char+pipeChars = pipeWith Process.pipeChars++-------------------------------------------------------------------------------+-- Generation+-------------------------------------------------------------------------------++-- |+--+-- >>> toBytes = streamWith Process.toBytes+--+-- >>> toBytes "echo hello world" & Stdio.putBytes+--hello world+-- >>> toBytes "echo hello\\ world" & Stdio.putBytes+--hello world+-- >>> toBytes "echo 'hello world'" & Stdio.putBytes+--hello world+-- >>> toBytes "echo \"hello world\"" & Stdio.putBytes+--hello world+--+-- /Pre-release/+{-# INLINE toBytes #-}+toBytes :: (MonadAsync m, MonadCatch m) => String -> Stream m Word8+toBytes = streamWith Process.toBytes++-- |+--+-- >>> toChunks = streamWith Process.toChunks+--+-- >>> toChunks "echo hello world" & Stdio.putChunks+--hello world+--+-- /Pre-release/+{-# INLINE toChunks #-}+toChunks :: (MonadAsync m, MonadCatch m) => String -> Stream m (Array Word8)+toChunks = streamWith Process.toChunks++-- |+-- >>> toChars = streamWith Process.toChars+--+-- >>> toChars "echo hello world" & Stdio.putChars+--hello world+--+-- /Pre-release/+{-# INLINE toChars #-}+toChars :: (MonadAsync m, MonadCatch m) => String -> Stream m Char+toChars = streamWith Process.toChars++-- |+-- >>> toLines f = streamWith (Process.toLines f)+--+-- >>> toLines Fold.toList "echo -e hello\\\\nworld" & Stream.fold Fold.toList+-- ["hello","world"]+--+-- /Pre-release/+{-# INLINE toLines #-}+toLines ::+    (MonadAsync m, MonadCatch m)+    => Fold m Char a+    -> String       -- ^ Command+    -> Stream m a -- ^ Output Stream+toLines f = streamWith (Process.toLines f)++-- |+-- >>> toString = runWith Process.toString+--+-- >>> toString "echo hello world"+--"hello world\n"+--+-- /Pre-release/+{-# INLINE toString #-}+toString ::+    (MonadAsync m, MonadCatch m)+    => String       -- ^ Command+    -> m String+toString = runWith Process.toString++-- |+-- >>> toStdout = runWith Process.toStdout+--+-- >>> toStdout "echo hello world"+-- hello world+--+-- /Pre-release/+{-# INLINE toStdout #-}+toStdout ::+    (MonadAsync m, MonadCatch m)+    => String       -- ^ Command+    -> m ()+toStdout = runWith Process.toStdout++-- |+-- >>> toNull = runWith Process.toNull+--+-- >>> toNull "echo hello world"+--+-- /Pre-release/+{-# INLINE toNull #-}+toNull ::+    (MonadAsync m, MonadCatch m)+    => String -- ^ Command+    -> m ()+toNull = runWith Process.toNull

@@ -49,45 +49,101 @@     -- * Process Configuration       Config +    -- ** Common Config Options+    -- | These options apply to both POSIX and Windows.+    , setCwd+    , setEnv+    {-+    , setStdin+    , setStdout+    , setStderr+    -}+    , closeFiles+    , newProcessGroup+    , setSession++    -- * Posix Only Options+    -- | These options have no effect on Windows.+    , parentIgnoresInterrupt+    , setUserId+    , setGroupId++    -- * Windows Only Options+    -- | These options have no effect on Posix.+    , waitForChildTree++    -- * Internal+    , inheritStdin+    , inheritStdout+    , pipeStdErr+     -- * Exceptions     , ProcessFailure (..)      -- * Generation+    -- | stdout of the process is redirected to output stream.     , toBytes-    , toBytes'     , toChunks-    , toChunks'+    , toChunksWith+    , toChars+    , toLines+    , toString+    , toStdout+    , toNull      -- * Transformation+    -- | The input stream is redirected to the stdin of the process, stdout of+    -- the process is redirected to the output stream.+    , pipeBytes+    , pipeChunks+    , pipeChunksWith+    , pipeChars++    -- * Stderr+    -- | Like other "Generation" routines but along with stdout, stderr is also+    -- included in the output stream. stdout is converted to 'Right' values in+    -- the output stream and stderr is converted to 'Left' values.+    , toBytesEither+    , toChunksEither+    , toChunksEitherWith+    , pipeBytesEither+    , pipeChunksEither+    , pipeChunksEitherWith++    -- * Standalone Processes+    , standalone+    , interactive+    , daemon++    -- * Deprecated     , processBytes-    , processBytes'-    , processChunksWith     , processChunks-    , processChunks'With-    , processChunks'     ) where --- #define USE_NATIVE-+import Control.Concurrent (forkIO)+import Control.Exception (Exception(..), catch, throwIO)+import Control.Monad (void, unless) import Control.Monad.Catch (MonadCatch, throwM) import Control.Monad.IO.Class (MonadIO, liftIO)-import Data.Word (Word8)-import Streamly.Data.Array.Foreign (Array)-import Streamly.Prelude (MonadAsync, parallel, IsStream, adapt)+import Data.Function ((&))+import Data.Word (Word8, Word32)+import Foreign.C.Error (Errno(..), ePIPE)+import GHC.IO.Exception (IOException(..), IOErrorType(..))+import Streamly.Data.Array (Array)+import Streamly.Data.Fold (Fold)+import Streamly.Data.Stream.Prelude (MonadAsync, Stream) import System.Exit (ExitCode(..)) import System.IO (hClose, Handle)+#if !defined(mingw32_HOST_OS)+import System.Posix.Types (CUid (..), CGid (..))+#endif  #ifdef USE_NATIVE-import Control.Exception (Exception(..), catch, throwIO, SomeException)+import Control.Exception (SomeException) import System.Posix.Process (ProcessStatus(..)) import Streamly.Internal.System.Process.Posix #else-import Control.Concurrent (forkIO)-import Control.Exception (Exception(..), catch, throwIO)-import Control.Monad (void, unless)-import Foreign.C.Error (Errno(..), ePIPE)-import GHC.IO.Exception (IOException(..), IOErrorType(..)) import System.Process     ( ProcessHandle     , CreateProcess(..)@@ -96,32 +152,35 @@     , waitForProcess     , CmdSpec(..)     , terminateProcess+    , withCreateProcess     ) #endif -import qualified Streamly.Data.Array.Foreign as Array-import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold  -- Internal imports import Streamly.Internal.System.IO (defaultChunkSize) -import qualified Streamly.Internal.Data.Array.Stream.Foreign-    as ArrayStream (arraysOf)-import qualified Streamly.Internal.Data.Stream.IsStream as Stream (bracket')+import qualified Streamly.Internal.Console.Stdio as Stdio (putChunks)+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Internal.Data.Unfold as Unfold (either) import qualified Streamly.Internal.FileSystem.Handle-    as Handle (toChunks, putChunks)+    as Handle (readChunks, putChunks)+import qualified Streamly.Unicode.Stream as Unicode+import qualified Streamly.Internal.Unicode.Stream as Unicode (lines)  -- $setup -- >>> :set -XFlexibleContexts -- >>> import Data.Char (toUpper) -- >>> import Data.Function ((&))--- >>> import qualified Streamly.Console.Stdio as Stdio+-- >>> import qualified Streamly.Internal.Console.Stdio as Stdio -- >>> import qualified Streamly.Data.Fold as Fold--- >>> import qualified Streamly.Prelude as Stream--- >>> import qualified Streamly.System.Process as Process+-- >>> import qualified Streamly.Data.Stream.Prelude as Stream+-- >>> import qualified Streamly.Internal.System.Process as Process -- >>> import qualified Streamly.Unicode.Stream as Unicode--- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream+-- >>> import qualified Streamly.Internal.Data.Stream as Stream+-- >>> import qualified Streamly.Internal.Unicode.Stream as Unicode  ------------------------------------------------------------------------------- -- Config@@ -133,13 +192,19 @@ -- from the parent process: -- -- * Current working directory--- * Environment+-- * Environment variables -- * Open file descriptors -- * Process group+-- * Terminal session+--+-- On POSIX:+-- -- * Process uid and gid -- * Signal handlers--- * Terminal (Session) --+-- On Windows by default the parent process waits for the entire child process+-- tree to finish.+-- #ifdef USE_NATIVE  type ProcessHandle = Process@@ -154,7 +219,15 @@  pipeStdErr :: Config -> Config pipeStdErr (Config _) = Config True++inheritStdin :: Config -> Config+inheritStdin (Config _) = Config True++inheritStdout :: Config -> Config+inheritStdout (Config _) = Config True+ #else+ newtype Config = Config CreateProcess  -- | Create a default process configuration from an executable file path and@@ -178,7 +251,9 @@     , child_group = Nothing  -- Posix only      -- Signals (Posix only) behavior-    -- Reset SIGINT (Ctrl-C) and SIGQUIT (Ctrl-\) to default handlers.+    -- Ignore SIGINT (Ctrl-C) and SIGQUIT (Ctrl-\) in the parent process until+    -- the child exits i.e. let the child handle it. See+    -- https://www.cons.org/cracauer/sigint.html .     , delegate_ctlc = False      -- Terminal behavior@@ -190,8 +265,183 @@     , use_process_jobs = True -- Windows only     } +-- XXX use osPath++-- | Set the current working directory of the new process. When 'Nothing', the+-- working directory is inherited from the parent process.+--+-- Default is 'Nothing' - inherited from the parent process.+setCwd :: Maybe (FilePath) -> Config -> Config+setCwd path (Config cfg) = Config $ cfg { cwd = path }++-- | Set the environment variables for the new process. When 'Nothing', the+-- environment is inherited from the parent process.+--+-- Default is 'Nothing' - inherited from the parent process.+setEnv :: Maybe [(String, String)] -> Config -> Config+setEnv e (Config cfg) = Config $ cfg { env = e }++{-+-- XXX We should allow setting only those stdio streams which are not used for+-- piping. We can either close those or inherit from parent.+--+-- * In a source we have to close stdin and use stdout+-- * In a pipe we have to use both+-- * In a sink we have to close stdout and use stdin.+--+-- Only stderr may be left for setting - either pipe it to merge it with stdout+-- or inherit or close. Closing may lead to bad behavior in most cases. So it+-- is either inherit or merge with stdout. Merge with stdout can be achieved by+-- using the either combinators. So there is nothing really left to set here+-- for any std stream.++-- UseProc, UsePipe?+-- | What to do with the stdin, stdout, stderr streams of the process.+data Stdio =+      ToHaskell -- ^ Pipe to Haskell Streamly+    | ToProcess -- ^ Connect to the parent process++toStdStream :: Stdio -> StdStream+toStdStream x =+    case x of+        ToProcess -> Inherit+        ToHaskell -> CreatePipe++-- | What to do with the @stdin@ stream of the process.+setStdin :: Stdio -> Config -> Config+setStdin x (Config cfg) = Config $ cfg { std_in = toStdStream x }++-- | What to do with the @stdout@ stream of the process.+setStdout :: Stdio -> Config -> Config+setStdout x (Config cfg) = Config $ cfg { std_out = toStdStream x }++-- | What to do with the @stderr@ stream of the process.+setStderr :: Stdio -> Config -> Config+setStderr x (Config cfg) = Config $ cfg { std_err = toStdStream x }+-}++-- | Close all open file descriptors inherited from the parent process. Note,+-- this does not apply to stdio descriptors - the behavior of those is determined+-- by other configuration settings.+--+-- Default is 'False'.+--+-- Note: if the number of open descriptors is large, it may take a while+-- closing them.+closeFiles :: Bool -> Config -> Config+closeFiles x (Config cfg) = Config $ cfg { close_fds = x }++-- XXX Do these details apply to Windows as well?++-- | If 'True' the new process starts a new process group, becomes a process+-- group leader, its pid becoming the process group id.+--+-- See the POSIX @setpgid@ man page.+--+-- Default is 'False', the new process belongs to the parent's process group.+newProcessGroup :: Bool -> Config -> Config+newProcessGroup x (Config cfg) = Config $ cfg { create_group = x }++-- | 'InheritSession' makes the new process inherit the terminal session from the+-- parent process. This is the default.+--+-- 'NewSession' makes the new process start with a new session without a+-- controlling terminal. On POSIX, @setsid@ is used to create a new process+-- group and session, the pid of the new process is the session id and process+-- group id as well. On Windows @DETACHED_PROCESS@ flag is used to detach the+-- process from inherited console session.+--+-- 'NewConsole' creates a new terminal and attaches the process to the new+-- terminal on Windows, using the CREATE_NEW_CONSOLE flag. On POSIX this does+-- nothing.+--+-- For Windows see+-- * https://learn.microsoft.com/en-us/windows/win32/procthread/process-creation-flags+-- * https://learn.microsoft.com/en-us/windows/console/creation-of-a-console .+--+-- For POSIX see, @setsid@ man page.+data Session =+      InheritSession -- ^ Inherit the parent session+    | NewSession -- ^ Detach process from the current session+    | NewConsole -- ^ Windows only, CREATE_NEW_CONSOLE flag++-- | Define the terminal session behavior for the new process.+--+-- Default is 'InheritSession'.+setSession :: Session -> Config -> Config+setSession x (Config cfg) =+    Config $+        case x of+            InheritSession -> cfg+            NewSession -> cfg { new_session = True}+            NewConsole -> cfg {create_new_console = True}++-- | Use the POSIX @setuid@ call to set the user id of the new process before+-- executing the command. The parent process must have sufficient privileges to+-- set the user id.+--+-- POSIX only. See the POSIX @setuid@ man page.+--+-- Default is 'Nothing' - inherit from the parent.+setUserId :: Maybe Word32 -> Config -> Config+#if defined(mingw32_HOST_OS)+setUserId _ (Config cfg) =+    Config cfg+#else+setUserId x (Config cfg) =+    Config $ cfg { child_user = CUid <$> x }+#endif++-- | Use the POSIX @setgid@ call to set the group id of the new process before+-- executing the command. The parent process must have sufficient privileges to+-- set the group id.+--+-- POSIX only. See the POSIX @setgid@ man page.+--+-- Default is 'Nothing' - inherit from the parent.+setGroupId :: Maybe Word32 -> Config -> Config+#if defined(mingw32_HOST_OS)+setGroupId _ (Config cfg) =+    Config cfg+#else+setGroupId x (Config cfg) =+    Config $ cfg { child_group = CGid <$> x }+#endif++-- See https://www.cons.org/cracauer/sigint.html for more details on signal+-- handling by interactive processes.++-- | When this is 'True', the parent process ignores user interrupt signals+-- @SIGINT@ and @SIGQUIT@ delivered to it until the child process exits. If+-- multiple child processes are started then the default handling in the parent+-- is restored only after the last one exits.+--+-- When a user presses CTRL-C or CTRL-\ on the terminal, a SIGINT or SIGQUIT is+-- sent to all the foreground processes in the terminal session, this includes+-- both the child and the parent. By default, on receiving these signals, the+-- parent process would cleanup and exit, to avoid that and let the child+-- handle these signals we can choose to ignore these signals in the parent+-- until the child exits.+--+-- POSIX only. Default is 'False'.+parentIgnoresInterrupt :: Bool -> Config -> Config+parentIgnoresInterrupt x (Config cfg) = Config $ cfg { delegate_ctlc = x }++-- | On Windows, the parent waits for the entire tree of process i.e. including+-- processes that are spawned by the child process.+--+-- Default is 'True'.+waitForChildTree :: Bool -> Config -> Config+waitForChildTree x (Config cfg) = Config $ cfg { use_process_jobs = x }+ pipeStdErr :: Config -> Config pipeStdErr (Config cfg) = Config $ cfg { std_err = CreatePipe }++inheritStdin :: Config -> Config+inheritStdin (Config cfg) = Config $ cfg { std_in = Inherit }++inheritStdout :: Config -> Config+inheritStdout (Config cfg) = Config $ cfg { std_out = Inherit } #endif  -------------------------------------------------------------------------------@@ -213,6 +463,9 @@     displayException (ProcessFailure exitCode) =         "Process failed with exit code: " ++ show exitCode +parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a+parallel s1 s2 = Stream.parList (Stream.eager True) [s1, s2]+ ------------------------------------------------------------------------------- -- Transformation -------------------------------------------------------------------------------@@ -221,9 +474,9 @@ -- 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 -- normally.-cleanupNormal :: MonadIO m =>-    (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m ()-cleanupNormal (_, _, _, procHandle) = liftIO $ do+cleanupNormal ::+    (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> IO ()+cleanupNormal (_, _, _, procHandle) = do #ifdef USE_NATIVE     -- liftIO $ putStrLn "cleanupNormal waiting"     status <- wait procHandle@@ -247,11 +500,15 @@ -- Since we are using SIGTERM to kill the process, it may block forever. We can -- possibly use a timer and send a SIGKILL after the timeout if the process is -- still hanging around.-cleanupException :: MonadIO m =>-    (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m ()-cleanupException (Just stdinH, Just stdoutH, stderrMaybe, ph) = liftIO $ do+cleanupException ::+    (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> IO ()+cleanupException (Just stdinH, Just stdoutH, stderrMaybe, ph) = do     -- Send a SIGTERM to the process+#ifdef USE_NATIVE+    terminate ph+#else     terminateProcess ph+#endif      -- 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@@ -261,7 +518,11 @@     whenJust hClose stderrMaybe      -- Non-blocking wait for the process to go away+#ifdef USE_NATIVE+    void $ forkIO (void $ wait ph)+#else     void $ forkIO (void $ waitForProcess ph)+#endif      where @@ -307,62 +568,62 @@     Config cfg = modCfg $ mkConfig path args  {-# INLINE putChunksClose #-}-putChunksClose :: (MonadIO m, IsStream t) =>-    Handle -> t m (Array Word8) -> t m a+putChunksClose :: MonadIO m =>+    Handle -> Stream m (Array Word8) -> Stream m a putChunksClose h input =     Stream.before-        (Handle.putChunks h (adapt input) >> liftIO (hClose h))+        (Handle.putChunks h input >> liftIO (hClose h))         Stream.nil  {-# INLINE toChunksClose #-}-toChunksClose :: (IsStream t, MonadAsync m) => Handle -> t m (Array Word8)-toChunksClose h = Stream.after (liftIO $ hClose h) (Handle.toChunks h)+toChunksClose :: MonadAsync m => Handle -> Stream m (Array Word8)+toChunksClose h = Stream.afterIO (hClose h) (Handle.readChunks h) -{-# INLINE processChunksWithAction #-}-processChunksWithAction ::-    (IsStream t, MonadCatch m, MonadAsync m)-    => ((Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> t m a)+{-# INLINE pipeChunksWithAction #-}+pipeChunksWithAction ::+    (MonadCatch m, MonadAsync m)+    => ((Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> Stream m a)     -> (Config -> Config)     -> FilePath             -- ^ Path to Executable     -> [String]             -- ^ Arguments-    -> t m a     -- ^ Output stream-processChunksWithAction run modCfg path args =-    Stream.bracket'+    -> Stream m a     -- ^ Output stream+pipeChunksWithAction run modCfg path args =+    Stream.bracketIO3           alloc cleanupNormal cleanupException cleanupException run      where -    alloc = liftIO $ createProc' modCfg path args+    alloc = createProc' modCfg path args -{-# INLINE processChunks'With #-}-processChunks'With ::-    (IsStream t, MonadCatch m, MonadAsync m)+{-# INLINE pipeChunksEitherWith #-}+pipeChunksEitherWith ::+    (MonadCatch m, MonadAsync m)     => (Config -> Config)   -- ^ Config modifier     -> FilePath             -- ^ Executable name or path     -> [String]             -- ^ Arguments-    -> t m (Array Word8)    -- ^ Input stream-    -> t m (Either (Array Word8) (Array Word8))     -- ^ Output stream-processChunks'With modifier path args input =-    processChunksWithAction run (modifier . pipeStdErr) path args+    -> 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      where      run (Just stdinH, Just stdoutH, Just stderrH, _) =         putChunksClose stdinH input-            `parallel` Stream.map Left (toChunksClose stderrH)-            `parallel` Stream.map Right (toChunksClose stdoutH)-    run _ = error "processChunks': Not reachable"+            `parallel` fmap Left (toChunksClose stderrH)+            `parallel` fmap Right (toChunksClose stdoutH)+    run _ = error "pipeChunksEitherWith: Not reachable" -{-# INLINE processChunks' #-}-processChunks' ::-    (IsStream t, MonadCatch m, MonadAsync m)+{-# INLINE pipeChunksEither #-}+pipeChunksEither ::+    (MonadCatch m, MonadAsync m)     => FilePath             -- ^ Executable name or path     -> [String]             -- ^ Arguments-    -> t m (Array Word8)    -- ^ Input stream-    -> t m (Either (Array Word8) (Array Word8))     -- ^ Output stream-processChunks' = processChunks'With id+    -> Stream m (Array Word8)    -- ^ Input stream+    -> Stream m (Either (Array Word8) (Array Word8))     -- ^ Output stream+pipeChunksEither = pipeChunksEitherWith id --- | @processBytes' path args input@ runs the executable at @path@ using @args@+-- | @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.@@ -373,45 +634,47 @@ -- world" | tr [:lower:] [:upper:]@: -- -- >>> :{---    processBytes' "echo" ["hello world"] Stream.nil---  & Stream.rights---  & processBytes' "tr" ["[:lower:]", "[:upper:]"]---  & Stream.rights+--    pipeBytesEither "echo" ["hello world"] Stream.nil+--  & Stream.catRights+--  & pipeBytesEither "tr" ["[:lower:]", "[:upper:]"]+--  & Stream.catRights --  & Stream.fold Stdio.write --  :} --HELLO WORLD -- -- @since 0.1.0-{-# INLINE processBytes' #-}-processBytes' ::-    (IsStream t, MonadCatch m, MonadAsync m)+{-# INLINE pipeBytesEither #-}+pipeBytesEither ::+    (MonadCatch m, MonadAsync m)     => FilePath         -- ^ Executable name or path     -> [String]         -- ^ Arguments-    -> t m Word8        -- ^ Input Stream-    -> t m (Either Word8 Word8) -- ^ Output Stream-processBytes' path args input =-    let input1 = ArrayStream.arraysOf defaultChunkSize input-        output = processChunks' path args input1-     in Stream.unfoldMany (Unfold.either Array.read) output+    -> Stream m Word8        -- ^ Input Stream+    -> Stream m (Either Word8 Word8) -- ^ Output Stream+pipeBytesEither path args input =+    let input1 = Stream.chunksOf 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 -{-# INLINE processChunksWith #-}-processChunksWith ::-    (IsStream t, MonadCatch m, MonadAsync m)+{-# INLINE pipeChunksWith #-}+pipeChunksWith ::+    (MonadCatch m, MonadAsync m)     => (Config -> Config)   -- ^ Config modifier     -> FilePath             -- ^ Executable name or path     -> [String]             -- ^ Arguments-    -> t m (Array Word8)    -- ^ Input stream-    -> t m (Array Word8)    -- ^ Output stream-processChunksWith modifier path args input =-    processChunksWithAction run modifier path args+    -> Stream m (Array Word8)    -- ^ Input stream+    -> Stream m (Array Word8)    -- ^ Output stream+pipeChunksWith modifier path args input =+    pipeChunksWithAction run modifier path args      where      run (Just stdinH, Just stdoutH, _, _) =         putChunksClose stdinH input `parallel` toChunksClose stdoutH-    run _ = error "processChunks: Not reachable"+    run _ = error "pipeChunksWith: Not reachable" --- | @processChunks file args input@ runs the executable @file@ specified by+-- | @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. --@@ -430,53 +693,136 @@ -- -- >>> :{ --    Process.toChunks "echo" ["hello world"]---  & Process.processChunks "tr" ["[a-z]", "[A-Z]"]+--  & Process.pipeChunks "tr" ["[a-z]", "[A-Z]"] --  & Stream.fold Stdio.writeChunks --  :} --HELLO WORLD ----- @since 0.1.0+-- /pre-release/+{-# INLINE pipeChunks #-}+pipeChunks ::+    (MonadCatch m, MonadAsync m)+    => FilePath             -- ^ Executable name or path+    -> [String]             -- ^ Arguments+    -> Stream m (Array Word8)    -- ^ Input stream+    -> Stream m (Array Word8)    -- ^ Output stream+pipeChunks = pipeChunksWith id++{-# DEPRECATED processChunks "Please use pipeChunks instead." #-} {-# INLINE processChunks #-} processChunks ::-    (IsStream t, MonadCatch m, MonadAsync m)+    (MonadCatch m, MonadAsync m)     => FilePath             -- ^ Executable name or path     -> [String]             -- ^ Arguments-    -> t m (Array Word8)    -- ^ Input stream-    -> t m (Array Word8)    -- ^ Output stream-processChunks = processChunksWith id+    -> Stream m (Array Word8)    -- ^ Input stream+    -> Stream m (Array Word8)    -- ^ Output stream+processChunks = pipeChunks --- | Like 'processChunks' except that it works on a stream of bytes instead of+-- | Like 'pipeChunks' except that it works on a stream of bytes instead of -- a stream of chunks. ----- We can write the example in 'processChunks' as follows. Notice how--- seamlessly we can replace the @tr@ process with the Haskell @toUpper@--- function:+-- We can write the example in 'pipeChunks' as follows. -- -- >>> :{ --    Process.toBytes "echo" ["hello world"]---  & Unicode.decodeLatin1 & Stream.map toUpper & Unicode.encodeLatin1+--  & Process.pipeBytes "tr" ["[a-z]", "[A-Z]"] --  & Stream.fold Stdio.write --  :} --HELLO WORLD ----- @since 0.1.0+-- /pre-release/+{-# INLINE pipeBytes #-}+pipeBytes ::+    (MonadCatch m, MonadAsync m)+    => FilePath     -- ^ Executable name or path+    -> [String]     -- ^ Arguments+    -> Stream m Word8    -- ^ Input Stream+    -> Stream m Word8    -- ^ Output Stream+pipeBytes path args input = -- rights . pipeBytesEither path args+    let input1 = Stream.chunksOf defaultChunkSize input+        output = pipeChunks path args input1+     in Stream.unfoldMany Array.reader output++{-# DEPRECATED processBytes "Please use pipeBytes instead." #-} {-# INLINE processBytes #-} processBytes ::-    (IsStream t, MonadCatch m, MonadAsync m)+    (MonadCatch m, MonadAsync m)     => FilePath     -- ^ Executable name or path     -> [String]     -- ^ Arguments-    -> t m Word8    -- ^ Input Stream-    -> t m Word8    -- ^ Output Stream-processBytes path args input = -- rights . processBytes' path args-    let input1 = ArrayStream.arraysOf defaultChunkSize input-        output = processChunks path args input1-     in Stream.unfoldMany Array.read output+    -> Stream m Word8    -- ^ Input Stream+    -> Stream m Word8    -- ^ Output Stream+processBytes = pipeBytes +-- | Like 'pipeChunks' except that it works on a stream of chars instead of+-- a stream of chunks.+--+-- >>> :{+--    Process.toChars "echo" ["hello world"]+--  & Process.pipeChars "tr" ["[a-z]", "[A-Z]"]+--  & Stdio.putChars+--  :}+--HELLO WORLD+--+-- We can seamlessly replace the @tr@ process with the Haskell @toUpper@+-- function:+--+-- >>> :{+--    Process.toChars "echo" ["hello world"]+--  & fmap toUpper+--  & Stdio.putChars+--  :}+--HELLO WORLD+--+-- /pre-release/+{-# INLINE pipeChars #-}+pipeChars ::+    (MonadCatch m, MonadAsync m)+    => FilePath     -- ^ Executable name or path+    -> [String]     -- ^ Arguments+    -> Stream m Char    -- ^ Input Stream+    -> Stream m Char    -- ^ Output Stream+pipeChars path args input =+    Unicode.encodeUtf8 input+        & pipeBytes path args+        & Unicode.decodeUtf8+ ------------------------------------------------------------------------------- -- Generation ------------------------------------------------------------------------------------- | @toBytes' path args@ runs the executable at @path@ using @args@ as++{-# INLINE toChunksEitherWith #-}+toChunksEitherWith ::+    (MonadCatch m, MonadAsync m)+    => (Config -> Config)   -- ^ Config modifier+    -> FilePath             -- ^ Executable name or path+    -> [String]             -- ^ Arguments+    -> Stream m (Either (Array Word8) (Array Word8))     -- ^ Output stream+toChunksEitherWith modifier path args =+    pipeChunksWithAction run (modifier . inheritStdin . pipeStdErr) path args++    where++    run (_, Just stdoutH, Just stderrH, _) =+        fmap Left (toChunksClose stderrH)+            `parallel` fmap Right (toChunksClose stdoutH)+    run _ = error "toChunksEitherWith: Not reachable"++{-# INLINE toChunksWith #-}+toChunksWith ::+    (MonadCatch m, MonadAsync m)+    => (Config -> Config)   -- ^ Config modifier+    -> FilePath             -- ^ Executable name or path+    -> [String]             -- ^ Arguments+    -> Stream m (Array Word8)    -- ^ Output stream+toChunksWith modifier path args =+    pipeChunksWithAction run (modifier . inheritStdin) path args++    where++    run (_, Just stdoutH, _, _) = toChunksClose stdoutH+    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. --@@ -486,31 +832,29 @@ -- to @stderr@, then uses folds from "Streamly.Console.Stdio" to write them -- back to @stdout@ and @stderr@ respectively: ----- -- >>> :{---   Process.toBytes' "/bin/bash" ["-c", "echo 'hello'; echo 'world' 1>&2"]+--   Process.toBytesEither "/bin/bash" ["-c", "echo 'hello'; echo 'world' 1>&2"] -- & Stream.fold (Fold.partition Stdio.writeErr Stdio.write) -- :} -- world -- hello -- ((),()) ----- >>> toBytes' path args = Process.processBytes' path args Stream.nil--- -- @since 0.1.0-{-# INLINE toBytes' #-}-toBytes' ::-    (IsStream t, MonadAsync m, MonadCatch m)+{-# INLINE toBytesEither #-}+toBytesEither ::+    (MonadAsync m, MonadCatch m)     => FilePath     -- ^ Executable name or path     -> [String]     -- ^ Arguments-    -> t m (Either Word8 Word8)    -- ^ Output Stream-toBytes' path args = processBytes' path args Stream.nil+    -> Stream m (Either Word8 Word8)    -- ^ Output Stream+toBytesEither path args =+    let output = toChunksEither path args+        leftRdr = fmap Left Array.reader+        rightRdr = fmap Right Array.reader+     in Stream.unfoldMany (Unfold.either leftRdr rightRdr) output --- | See 'processBytes'. 'toBytes' is defined as:------ >>> toBytes path args = processBytes path args Stream.nil------ The following code is equivalent to the shell command @echo "hello world"@:+-- | The following code is equivalent to the shell command @echo "hello+-- world"@: -- -- >>> :{ --    Process.toBytes "echo" ["hello world"]@@ -521,41 +865,40 @@ -- @since 0.1.0 {-# INLINE toBytes #-} toBytes ::-    (IsStream t, MonadAsync m, MonadCatch m)+    (MonadAsync m, MonadCatch m)     => FilePath     -- ^ Executable name or path     -> [String]     -- ^ Arguments-    -> t m Word8    -- ^ Output Stream-toBytes path args = processBytes path args Stream.nil+    -> Stream m Word8    -- ^ Output Stream+toBytes path args =+    let output = toChunks path args+     in Stream.unfoldMany Array.reader output  -- | Like 'toBytes' but generates a stream of @Array Word8@ instead of a stream -- of @Word8@. -- -- >>> :{---   toChunks' "bash" ["-c", "echo 'hello'; echo 'world' 1>&2"]+--   toChunksEither "bash" ["-c", "echo 'hello'; echo 'world' 1>&2"] -- & Stream.fold (Fold.partition Stdio.writeErrChunks Stdio.writeChunks) -- :} -- world -- hello -- ((),()) ----- >>> toChunks' path args = processChunks' path args Stream.nil+-- >>> toChunksEither = toChunksEitherWith id ----- Prefer 'toChunks' over 'toBytes' when performance matters.+-- Prefer 'toChunksEither over 'toBytesEither when performance matters. ----- @since 0.1.0-{-# INLINE toChunks' #-}-toChunks' ::-    (IsStream t, MonadAsync m, MonadCatch m)+-- /Pre-release/+{-# INLINE toChunksEither #-}+toChunksEither ::+    (MonadAsync m, MonadCatch m)     => FilePath             -- ^ Executable name or path     -> [String]             -- ^ Arguments-    -> t m (Either (Array Word8) (Array Word8))    -- ^ Output Stream-toChunks' path args = processChunks' path args Stream.nil+    -> Stream m (Either (Array Word8) (Array Word8))    -- ^ Output Stream+toChunksEither = toChunksEitherWith id --- | See 'processChunks'. 'toChunks' is defined as:------ >>> toChunks path args = processChunks path args Stream.nil------ The following code is equivalent to the shell command @echo "hello world"@:+-- | The following code is equivalent to the shell command @echo "hello+-- world"@: -- -- >>> :{ --    Process.toChunks "echo" ["hello world"]@@ -563,11 +906,154 @@ --  :} --hello world --+-- >>> toChunks = toChunksWith id+-- -- @since 0.1.0 {-# INLINE toChunks #-} toChunks ::-    (IsStream t, MonadAsync m, MonadCatch m)+    (MonadAsync m, MonadCatch m)     => FilePath             -- ^ Executable name or path     -> [String]             -- ^ Arguments-    -> t m (Array Word8)    -- ^ Output Stream-toChunks path args = processChunks path args Stream.nil+    -> Stream m (Array Word8)    -- ^ Output Stream+toChunks = toChunksWith id++-- |+-- >>> toChars path args = toBytes path args & Unicode.decodeUtf8+--+{-# INLINE toChars #-}+toChars ::+    (MonadAsync m, MonadCatch m)+    => FilePath       -- ^ Executable name or path+    -> [String]       -- ^ Arguments+    -> Stream m Char -- ^ Output Stream+toChars path args = toBytes path args & Unicode.decodeUtf8++-- |+-- >>> toLines path args f = toChars path args & Unicode.lines f+--+{-# INLINE toLines #-}+toLines ::+    (MonadAsync m, MonadCatch m)+    => Fold m Char a+    -> FilePath       -- ^ Executable name or path+    -> [String]       -- ^ Arguments+    -> 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+--+{-# 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++-- |+-- >>> toStdout path args = toChunks path args & Stdio.putChunks+--+{-# INLINE toStdout #-}+toStdout ::+    (MonadAsync m, MonadCatch m)+    => FilePath -- ^ Executable name or path+    -> [String] -- ^ Arguments+    -> m ()+toStdout path args = toChunks path args & Stdio.putChunks+{-+-- Directly inherits stdout instead.+toStdout path args = do+    _ <- liftIO $ createProc' (inheritStdin . inheritStdout) path args+    return ()+-}++-- |+-- >>> toNull path args = toChunks path args & Stream.fold Fold.drain+--+{-# INLINE toNull #-}+toNull ::+    (MonadAsync m, MonadCatch m)+    => FilePath -- ^ Executable name or path+    -> [String] -- ^ Arguments+    -> m ()+toNull path args = toChunks path args & Stream.fold Fold.drain++-------------------------------------------------------------------------------+-- Process not interacting with the parent process+-------------------------------------------------------------------------------++{-# INLINE standalone #-}+standalone ::+       Bool -- ^ Wait for process to finish?+    -> (Bool, Bool, Bool) -- ^ close (stdin, stdout, stderr)+    -> (Config -> Config)+    -> FilePath -- ^ Executable name or path+    -> [String] -- ^ Arguments+    -> IO (Either ExitCode ProcessHandle)+standalone wait (close_stdin, close_stdout, close_stderr) modCfg path args =+    withCreateProcess cfg postCreate++    where++    postCreate _ _ _ procHandle =+        if wait+        then fmap Left $ waitForProcess procHandle+        else return $ Right procHandle++    cfg =+        let Config c = modCfg $ mkConfig path args+            s_in = if close_stdin then NoStream else Inherit+            s_out = if close_stdout then NoStream else Inherit+            s_err = if close_stderr then NoStream else Inherit+        in c {std_in = s_in, std_out = s_out, std_err = s_err}++-- | Inherits stdin, stdout, and stderr from the parent, so that the user can+-- interact with the process, user interrupts are handled by the child process,+-- the parent waits for the child process to exit.+--+-- This is same as the common @system@ function found in other libraries used+-- to execute commands.+--+-- On Windows you can pass @setSession NewConsole@ to create a new console.+--+{-# INLINE interactive #-}+interactive ::+       (Config -> Config)+    -> FilePath -- ^ Executable name or path+    -> [String] -- ^ Arguments+    -> IO ExitCode+interactive modCfg path args =+    withCreateProcess cfg (\_ _ _ p -> waitForProcess p)++    where++    -- let child handle SIGINT/QUIT+    modCfg1 = (parentIgnoresInterrupt True . modCfg)++    cfg =+        let Config c = modCfg1 $ mkConfig path args+        in c {std_in = Inherit, std_out = Inherit, std_err = Inherit}++-- XXX ProcessHandle can be used to terminate the process. re-export+-- terminateProcess?++-- | Closes stdin, stdout and stderr, creates a new session, detached from the+-- terminal, the parent does not wait for the process to finish.+--+{-# INLINE daemon #-}+daemon ::+       (Config -> Config)+    -> FilePath -- ^ Executable name or path+    -> [String] -- ^ Arguments+    -> IO ProcessHandle+daemon modCfg path args = withCreateProcess cfg (\_ _ _ p -> return p)++    where++    -- Detach terminal+    modCfg1 = (setSession NewSession . modCfg)++    cfg =+        let Config c = modCfg1 $ mkConfig path args+        in c {std_in = NoStream, std_out = NoStream, std_err = NoStream}

@@ -1,3 +1,4 @@+{-# LANGUAGE Safe #-} -- | -- Module      : Streamly.Internal.System.Process.Posix -- Copyright   : (c) 2020 Composewell Technologies@@ -17,6 +18,7 @@     , newProcess     , wait     , getStatus+    , terminate      -- * IPC     , mkPipe@@ -38,6 +40,7 @@ import System.Posix.IO (createPipe, dupTo, closeFd) import System.Posix.Process (forkProcess, executeFile, ProcessStatus) import System.Posix.Types (ProcessID, Fd(..), CDev, CIno)+import System.Posix.Signals (signalProcess, sigTERM) import System.Posix.Internals (fdGetMode)  import qualified GHC.IO.FD as FD@@ -315,3 +318,6 @@         if isDoesNotExistError e         then return (Nothing, Nothing)         else throwIO e++terminate :: Process -> IO ()+terminate (Process pid _ _) = signalProcess sigTERM pid

@@ -0,0 +1,158 @@+-- |+-- Module      : Streamly.System.Command+-- Copyright   : (c) 2023 Composewell Technologies+-- License     : Apache-2.0+-- Maintainer  : streamly@composewell.com+-- Stability   : experimental+-- Portability : GHC+--+-- Use command strings to execute OS processes. These processes can be used+-- just like native Haskell functions - to generate, transform or consume+-- streams. It provides a powerful way to write high-level Haskell scripts to+-- perform tasks similar to shell scripts without requiring the shell.+-- Moreover, the Haskell scripts provide C-like performance.+--+-- This module is a wrapper over the "Streamly.System.Process" module.+--+-- See also: "Streamly.Internal.System.Command".+--+module Streamly.System.Command+    (+    -- * Setup+    -- | To execute the code examples provided in this module in ghci, please+    -- run the following commands first.+    --+    -- $setup++    -- * Overview+    -- $overview++    -- * Types+      ProcessFailure (..)++    -- * Generation+    , toBytes+    , toChunks+    , toChars+    , toLines++    -- * Effects+    , toString+    , toStdout+    , toNull++    -- * Transformation+    , pipeBytes+    , pipeChars+    , pipeChunks++    -- -- * Helpers+    -- , runWith+    -- , streamWith+    -- , pipeWith+    )+where++import Streamly.Internal.System.Command+import Streamly.Internal.System.Process (ProcessFailure (..))++-- Keep it synced with the Internal module++-- $setup+-- >>> :set -XFlexibleContexts+-- >>> :set -XQuasiQuotes+-- >>> import Data.Char (toUpper)+-- >>> import Data.Function ((&))+-- >>> import Streamly.Unicode.String (str)+-- >>> import qualified Streamly.Data.Array as Array+-- >>> import qualified Streamly.Console.Stdio as Stdio+-- >>> import qualified Streamly.Data.Fold as Fold+-- >>> import qualified Streamly.Data.Stream.Prelude as Stream+-- >>> import qualified Streamly.System.Command as Command+-- >>> import qualified Streamly.Unicode.Stream as Unicode+--+-- >>> import qualified Streamly.Internal.System.Process as Process+-- >>> import qualified Streamly.Internal.Console.Stdio as Stdio+-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir++-- Note: Commands are not executed using shell+--+-- You can use this module to execute system commands and compose them with+-- Haskell.  It does not use the system shell to execute commands, they are+-- executed as independent processes. This provides a convenient and powerful+-- way to replace shell scripting with Haskell. Instead of composing commands+-- using shell you can use Haskell Streamly streaming APIs to compose them with+-- better efficiency and type safety.+--+-- Normally, you should not need the system shell but if you want to use shell+-- scripts in your program then you can take a look at the @streamly-shell@+-- package which provides convenient wrapper over "Streamly.System.Process" to+-- execute shell scripts, commands.++-- $overview+--+-- Please see the "Streamly.System.Process" for basics.+--+-- "Streamly.System.Process" module requires specifying the command executable+-- name and its arguments separately (e.g. "ls" "-al") whereas using this+-- module we can specify the executable and its arguments more conveniently as+-- a single command string e.g.  we can execute "ls -al".+--+-- A command string is parsed in the same way as a posix shell would parse it.+-- A command string consists of whitespace separated tokens with the first+-- token treated as the executable name and the rest as arguments. Whitespace+-- can be escaped using @\\@. Alternatively, double quotes or single quotes can+-- be used to enclose tokens with whitespaces. Quotes can be escaped using @\\@.+-- Single quotes inside double quotes or vice-versa are treated as normal+-- characters.+--+-- You can use the string quasiquoter 'Streamly.Unicode.String.str' to write+-- commands conveniently, it allows Haskell variable expansion as well e.g.:+--+-- >>> f = "file name"+-- >>> [str|ls -al "#{f} with spaces"|]+-- "ls -al \"file name with spaces\""+--+-- With the "Streamly.System.Command" module you can write the examples in the+-- "Streamly.System.Process" module more conveniently.+--+-- = Executables as functions+--+-- The shell command @echo "hello world" | tr [a-z] [A-Z]@ can be written as+-- follows using this module:+--+-- >>> :{+--    Command.toBytes [str|echo "hello world"|]+--  & Command.pipeBytes [str|tr [a-z] [A-Z]|]+--  & Stream.fold Stdio.write+--  :}+--  HELLO WORLD+--+-- = Shell commands as functions+--+-- If you want to execute the same command using the shell:+--+-- >>> :{+--    Command.toBytes [str|sh "-c" "echo 'hello world' | tr [a-z] [A-Z]"|]+--  & Stream.fold Stdio.write+--  :}+--  HELLO WORLD+--+-- = Running Commands Concurrently+--+-- Running @grep@ concurrently on many files:+--+-- >>> :{+-- grep file =+--    Command.toBytes [str|grep -H "pattern" #{file}|]+--  & Stream.handle (\(_ :: Command.ProcessFailure) -> Stream.nil)+--  & Stream.foldMany (Fold.takeEndBy (== 10) Array.write)+--  :}+--+-- >>> :{+-- pgrep =+--    Dir.readFiles "."+--  & Stream.parConcatMap id grep+--  & Stream.fold Stdio.writeChunks+-- :}+--

@@ -6,45 +6,107 @@ -- Stability   : experimental -- Portability : GHC ----- Use OS processes as stream transformation functions.------ This module provides functions to run operating system processes as stream--- source, sink or transformation functions. Thus OS processes can be used in--- the same way as Haskell functions and all the streaming combinators in--- streamly can be used to combine them. This allows you to seamlessly--- integrate external programs into your Haskell program.------ We recommend that you use Streamly threads instead of system processes where--- possible as they have a simpler programming model and processes have a--- larger performance overhead.+-- Use OS processes just like native Haskell functions - to generate, transform+-- or consume streams. ----- = Imports for examples+-- See "Streamly.System.Command" module for a higher level wrapper over this+-- module. ----- Use the following imports for examples below.+-- See also: "Streamly.Internal.System.Process" for unreleased functions. --+module Streamly.System.Process+    (+    -- * Setup+    -- | To execute the code examples provided in this module in ghci, please+    -- run the following commands first.+    --+    -- $setup++    -- * Overview+    -- $overview++    -- * Exceptions+    -- | Since we are composing using Streamly's streaming pipeline there is+    -- nothing special about exception handling, it works the same as in+    -- Streamly.  Like the @pipefail@ option in shells, exceptions are+    -- propagated if any of the stages fail.+      ProcessFailure (..)++    -- * Process Configuration+    , Config++    {-+    -- ** Common Config Options+    -- | These options apply to both POSIX and Windows.+    , setCwd+    , setEnv+    , closeFiles+    , newProcessGroup+    , setSession++    -- * Posix Only Options+    -- | These options have no effect on Windows.+    , parentIgnoresInterrupt+    , setUserId+    , setGroupId++    -- * Windows Only Options+    -- | These options have no effect on Posix.+    , waitForChildTree+    -}++    -- * Generation+    , toChunks+    , toBytes++    -- * Transformation+    , pipeChunks+    , pipeBytes++    -- * Deprecated+    , processChunks+    , processBytes+    )+where++import Streamly.Internal.System.Process++-- $setup -- >>> :set -XFlexibleContexts -- >>> :set -XScopedTypeVariables -- >>> import Data.Char (toUpper) -- >>> import Data.Function ((&)) -- >>> import qualified Streamly.Console.Stdio as Stdio--- >>> import qualified Streamly.Data.Array.Foreign as Array+-- >>> import qualified Streamly.Data.Array as Array -- >>> import qualified Streamly.Data.Fold as Fold--- >>> import qualified Streamly.Prelude as Stream+-- >>> import qualified Streamly.Data.Stream.Prelude as Stream -- >>> import qualified Streamly.System.Process as Process -- >>> import qualified Streamly.Unicode.Stream as Unicode+-- -- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir--- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream+-- >>> import qualified Streamly.Internal.Data.Stream as Stream++-- $overview --+-- This module provides functions to run operating system processes as stream+-- source, sink or transformation functions. Thus OS processes can be used in+-- the same way as Haskell functions and all the streaming combinators in+-- streamly can be used to combine them. This allows you to seamlessly+-- integrate external programs into your Haskell program.+--+-- We recommend using Haskell functions with Streamly threads for performing+-- tasks whenever possible. This approach offers a simpler programming model+-- compared to system processes, which also have a larger performance overhead.+-- -- = Executables as functions ----- Streamly provides powerful ways to combine streams. Processes can be--- composed in a streaming pipeline just like a Posix shell command pipeline--- except that we use @&@ instead of @|@. Moreover, we can mix processes and--- Haskell functions seamlessly in a processing pipeline. For example:+-- Processes can be composed in a streaming pipeline just like a Posix shell+-- command pipeline. Moreover, we can mix processes and Haskell functions+-- seamlessly in a processing pipeline. For example: -- -- >>> :{ --    Process.toBytes "echo" ["hello world"]---  & Process.processBytes "tr" ["[a-z]", "[A-Z]"]+--  & Process.pipeBytes "tr" ["[a-z]", "[A-Z]"] --  & Stream.fold Stdio.write --  :} --  HELLO WORLD@@ -53,7 +115,7 @@ -- -- >>> :{ --    Process.toBytes "echo" ["hello world"]---  & Unicode.decodeLatin1 & Stream.map toUpper & Unicode.encodeLatin1+--  & Unicode.decodeLatin1 & fmap toUpper & Unicode.encodeLatin1 --  & Stream.fold Stdio.write --  :} --  HELLO WORLD@@ -69,7 +131,7 @@ --  :} --  HELLO ----- = Concurrent Processing+-- = Running Commands Concurrently -- -- We can run executables or commands concurrently as we would run any other -- functions in Streamly. For example, the following program greps the word@@ -77,63 +139,14 @@ -- -- >>> :{ -- grep file =---    Process.toBytes "grep" ["-H", "to", file]+--    Process.toBytes "grep" ["-H", "pattern", file] --  & Stream.handle (\(_ :: Process.ProcessFailure) -> Stream.nil)---  & Stream.splitWithSuffix (== 10) Array.write+--  & Stream.foldMany (Fold.takeEndBy (== 10) Array.write) --  :} -- -- >>> :{ -- pgrep =---    Dir.toFiles "."---  & Stream.concatMapWith Stream.async grep+--    Dir.readFiles "."+--  & Stream.parConcatMap id grep --  & Stream.fold Stdio.writeChunks -- :}------ = Exception handling------ Since we are composing using Streamly streaming pipeline there is nothing--- special about exception handling, it works the same as in Streamly.  Like--- the @pipefail@ option in shells, exceptions are propagated if any of the--- stages fail.------ = Process Attributes------ When a new process is spawned, the following attributes are inherited from--- the parent process:------ * Current working directory--- * Environment--- * Open file descriptors--- * Process group--- * Process uid and gid--- * Signal handlers--- * Terminal (Session)--module Streamly.System.Process-    ( ProcessFailure (..)--    -- * Generation-    , toChunks-    , toBytes--    -- * Transformation-    , processChunks-    , processBytes-    )-where--import Streamly.Internal.System.Process---- $setup--- >>> :set -XFlexibleContexts--- >>> :set -XScopedTypeVariables--- >>> import Data.Char (toUpper)--- >>> import Data.Function ((&))--- >>> import qualified Streamly.Console.Stdio as Stdio--- >>> import qualified Streamly.Data.Array.Foreign as Array--- >>> import qualified Streamly.Data.Fold as Fold--- >>> import qualified Streamly.Prelude as Stream--- >>> import qualified Streamly.System.Process as Process--- >>> import qualified Streamly.Unicode.Stream as Unicode--- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir--- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream

@@ -1,11 +1,14 @@ cabal-version:       2.2 name:                streamly-process-version:             0.2.0.1+version:             0.3.0 synopsis:            Use OS processes as stream transformation functions description:-    Run operating system processes as stream source, sink or transformation-    functions. Use them seamlessly in a streaming data pipeline in the same way-    as any other Haskell functions.+  Use operating system (OS) commands in Haskell programs as if they were+  native Haskell functions, by treating their inputs and outputs as+  Haskell streams. This allows you to write high-level Haskell scripts+  that can perform tasks similar to shell scripts, but with C-like+  performance, and with strong safety guarantees, refactorability, and+  modularity. homepage:            https://streamly.composewell.com bug-reports:         https://github.com/composewell/streamly-process/issues license:             Apache-2.0@@ -15,12 +18,12 @@ copyright:           Composewell Technologies category:            Streamly, Streaming, System stability:           Experimental-tested-with:         GHC==9.2.1-                   , GHC==9.0.1-                   , GHC==8.10.5+tested-with:         GHC==9.4.4+                   , GHC==9.2.7+                   , GHC==9.0.2+                   , GHC==8.10.7                    , GHC==8.8.4                    , GHC==8.6.5-                   , GHC==8.4.4 build-type:          Simple extra-source-files:   CHANGELOG.md@@ -43,22 +46,29 @@   manual: True   default: False -flag use-gauge-  description: Use gauge instead of tasty-bench for benchmarking+flag use-native+  description: Do not depend on the process package   manual: True   default: False  common compile-options   default-language: Haskell2010   ghc-options:-    -Wall-    -Wcompat-    -Wunrecognised-warning-flags-    -Widentities-    -Wincomplete-record-updates-    -Wincomplete-uni-patterns-    -Wredundant-constraints-    -Wnoncanonical-monad-instances+    -Weverything+    -Wno-implicit-prelude+    -Wno-missing-deriving-strategies+    -Wno-missing-exported-signatures+    -Wno-missing-import-lists+    -Wno-missing-local-signatures+    -Wno-missing-safe-haskell-mode+    -Wno-missed-specialisations+    -Wno-all-missed-specialisations+    -Wno-monomorphism-restriction+    -Wno-prepositive-qualified-module+    -Wno-unsafe+  if !impl(ghc < 9.2)+    ghc-options:+      -Wno-missing-kind-signatures  common optimization-options   ghc-options:@@ -66,34 +76,32 @@     -fdicts-strict     -fspec-constr-recursive=16     -fmax-worker-args=16+  if flag(use-native)+    cpp-options: -DUSE_NATIVE  library   import: compile-options, optimization-options   hs-source-dirs: src   exposed-modules:     Streamly.System.Process+    Streamly.System.Command     Streamly.Internal.System.Process-  if !os(windows)+    Streamly.Internal.System.Command+  if flag (use-native) && !os(windows)     exposed-modules:       Streamly.Internal.System.Process.Posix-  ghc-options:-    -Wall-    -Wcompat-    -Wunrecognised-warning-flags-    -Widentities-    -Wincomplete-record-updates-    -Wincomplete-uni-patterns-    -Wredundant-constraints-    -Wnoncanonical-monad-instances   build-depends:       base              >= 4.8   && < 5-    , process           >= 1.0   && < 1.7-    -- Uses internal APIs-    , streamly          == 0.8.1.* || == 0.8.2.*     , exceptions        >= 0.8   && < 0.11-  if !os(windows)-    build-depends:-      unix              >= 2.5   && < 2.8+    -- Uses internal APIs+    , streamly          == 0.9.0.*+    , streamly-core     == 0.1.0+  if !flag(use-native)+    build-depends: process >= 1.0 && < 1.7+  else+     if !os(windows)+       build-depends:+         unix >= 2.5 && < 2.8  ------------------------------------------------------------------------------- -- Benchmarks@@ -114,21 +122,13 @@       streamly-process     , base              >= 4.8   && < 5     , directory         >= 1.2.2 && < 1.4-    , process           >= 1.0   && < 1.7     -- Uses internal APIs-    , streamly          == 0.8.1.* || == 0.8.2.*+    , streamly-core     == 0.1.0.*+    , tasty-bench >= 0.2.5 && < 0.4    if flag(fusion-plugin) && !impl(ghc < 8.6)     build-depends:         fusion-plugin     >= 0.2   && < 0.3-  if flag(use-gauge)-    build-depends:  gauge >= 0.2.4 && < 0.3-  else-    build-depends:    tasty-bench >= 0.2.5 && < 0.4-    mixins: tasty-bench-      (Test.Tasty.Bench as Gauge-      , Test.Tasty.Bench as Gauge.Main-      )  ------------------------------------------------------------------------------- -- Test Suites@@ -144,7 +144,6 @@     , directory         >= 1.2.2 && < 1.4     , exceptions        >= 0.8   && < 0.11     , hspec             >= 2.0   && < 3-    , process           >= 1.0   && < 1.7     , QuickCheck        >= 2.10  && < 2.15     -- Uses internal APIs-    , streamly          == 0.8.1.* || == 0.8.2.*+    , streamly-core     == 0.1.0.*

@@ -1,6 +1,6 @@ {-# LANGUAGE CPP #-} -module Main where+module Main (main) where  import Data.Function ((&)) import Data.List ((\\))@@ -10,8 +10,11 @@ import Control.Monad (unless) import Control.Monad.Catch (throwM, catch) import Control.Monad.IO.Class (MonadIO(..))+import Streamly.Data.Array (Array, Unbox)+import Streamly.Data.Stream (Stream) import Streamly.System.Process (ProcessFailure (..))-import System.Directory (removeFile, findExecutable)+import System.Directory (removeFile, findExecutable, doesFileExist)+import System.Exit (exitSuccess) import System.IO (IOMode(..), openFile, hClose) import Test.Hspec (hspec, describe) import Test.Hspec.QuickCheck@@ -24,16 +27,14 @@     ) import Test.QuickCheck.Monadic (monadicIO, PropertyM, assert, monitor, run) +import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold-import qualified Streamly.Prelude as S+import qualified Streamly.Data.Stream as S import qualified Streamly.System.Process as Proc --- Internal imports-import qualified Streamly.Internal.Data.Array.Stream.Foreign-    as AS (arraysOf, concat)-import qualified Streamly.Internal.Data.Stream.IsStream as S (nilM, lefts, rights)-import qualified Streamly.Internal.FileSystem.Handle as FH (putBytes, toBytes)+import qualified Streamly.Internal.FileSystem.Handle as FH (putBytes, read) import qualified Streamly.Internal.System.Process as Proc+    (pipeChunksEither, pipeBytesEither, toChunksEither, toBytesEither)  newtype SimpleError = SimpleError String     deriving Show@@ -68,18 +69,12 @@ maxBlockCount :: Int maxBlockCount = 100 -minNumChar :: Int-minNumChar = 1--maxNumChar :: Int-maxNumChar = 100 * 1024- arrayChunkSize :: Int arrayChunkSize = 100  interpreterFile :: FilePath interpreterArg :: String-#if mingw32_HOST_OS == 1+#ifdef mingw32_HOST_OS interpreterFile = "cmd.exe" interpreterArg = "/c" #else@@ -88,21 +83,21 @@ #endif  executableFile :: FilePath-#if mingw32_HOST_OS == 1+#ifdef mingw32_HOST_OS executableFile = "./test/data/writeTrToError.bat" #else executableFile = "./test/data/writeTrToError.sh" #endif  executableFileFail :: FilePath-#if mingw32_HOST_OS == 1+#ifdef mingw32_HOST_OS executableFileFail = "./test/data/failExec.bat" #else executableFileFail = "./test/data/failExec.sh" #endif  executableFilePass :: FilePath-#if mingw32_HOST_OS == 1+#ifdef mingw32_HOST_OS executableFilePass = "./test/data/passExec.bat" #else executableFilePass = "./test/data/passExec.sh"@@ -147,17 +142,17 @@         monadicIO $ do             catBinary <- run $ which "cat"             let-                genStrm = S.rights $ Proc.toBytes' catBinary [charFile]+                genStrm = S.catRights $ Proc.toBytesEither catBinary [charFile]                  ioByteStrm = do                     handle <- openFile charFile ReadMode-                    let strm = FH.toBytes handle-                    return $ S.finally (hClose handle) strm+                    let strm = FH.read handle+                    return $ S.finallyIO (hClose handle) strm              run $ generateCharFile numBlock             byteStrm <- run ioByteStrm-            genList <- run $ S.toList genStrm-            byteList <- run $ S.toList byteStrm+            genList <- run $ S.fold Fold.toList genStrm+            byteList <- run $ S.fold Fold.toList byteStrm             run $ removeFile charFile             listEquals (==) genList byteList @@ -166,8 +161,8 @@     where      action = do-        S.drain $-             Proc.toBytes' interpreterFile [interpreterArg, executableFileFail]+        S.fold Fold.drain $+             Proc.toBytesEither interpreterFile [interpreterArg, executableFileFail]         return False      failAction (ProcessFailure exitCode) =@@ -175,23 +170,27 @@      checkFailAction = catch action failAction +{-# INLINE concatArr #-}+concatArr :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a+concatArr = S.unfoldMany Array.reader+ toChunks1 :: Property toChunks1 =     forAll (choose (minBlockCount, maxBlockCount)) $ \numBlock ->         monadicIO $ do             catBinary <- run $ which "cat"             let-                genStrm = S.rights $ Proc.toChunks' catBinary [charFile]+                genStrm = S.catRights $ Proc.toChunksEither catBinary [charFile]                  ioByteStrm = do                     handle <- openFile charFile ReadMode-                    let strm = FH.toBytes handle-                    return $ S.finally (hClose handle) strm+                    let strm = FH.read handle+                    return $ S.finallyIO (hClose handle) strm              run $ generateCharFile numBlock             byteStrm <- run ioByteStrm-            genList <- run $ S.toList (AS.concat genStrm)-            byteList <- run $ S.toList byteStrm+            genList <- run $ S.fold Fold.toList (concatArr genStrm)+            byteList <- run $ S.fold Fold.toList byteStrm             run $ removeFile charFile             listEquals (==) genList byteList @@ -200,8 +199,8 @@     where      action = do-        S.drain $-             Proc.toChunks' interpreterFile [interpreterArg, executableFileFail]+        S.fold Fold.drain $+             Proc.toChunksEither interpreterFile [interpreterArg, executableFileFail]         return False      failAction (ProcessFailure exitCode) =@@ -209,30 +208,30 @@      checkFailAction = catch action failAction -processBytes1 :: Property-processBytes1 =+pipeBytes1 :: Property+pipeBytes1 =     forAll (listOf (choose(_a, _z))) $ \ls ->         monadicIO $ do             trBinary <- run $ which "tr"             let                 inputStream = S.fromList ls-                genStrm = Proc.processBytes+                genStrm = Proc.pipeBytes                             trBinary                             ["[a-z]", "[A-Z]"]                             inputStream-                charUpperStrm = S.map toUpper inputStream+                charUpperStrm = fmap toUpper inputStream -            genList <- run $ S.toList genStrm-            charList <- run $ S.toList charUpperStrm+            genList <- run $ S.fold Fold.toList genStrm+            charList <- run $ S.fold Fold.toList charUpperStrm             listEquals (==) genList charList -processBytes2 :: Property-processBytes2 = monadicIO $ run checkFailAction+pipeBytes2 :: Property+pipeBytes2 = monadicIO $ run checkFailAction     where      action = do-        S.drain $-           Proc.processBytes+        S.fold Fold.drain $+           Proc.pipeBytes                interpreterFile [interpreterArg, executableFileFail] S.nil         return False @@ -241,13 +240,13 @@      checkFailAction = catch action failAction -processBytes3 :: Property-processBytes3 = monadicIO $ run checkFailAction+pipeBytes3 :: Property+pipeBytes3 = monadicIO $ run checkFailAction     where      action = do-        S.drain $-            Proc.processBytes+        S.fold Fold.drain $+            Proc.pipeBytes             interpreterFile             [interpreterArg, executableFilePass]             (S.nilM $ throwM (SimpleError failErrorMessage))@@ -267,16 +266,16 @@             let                 inputStream = S.fromList ls -                genStrm = AS.concat $-                    Proc.processChunks+                genStrm = concatArr $+                    Proc.pipeChunks                     trBinary                     ["[a-z]", "[A-Z]"]-                    (AS.arraysOf arrayChunkSize inputStream)+                    (S.chunksOf arrayChunkSize inputStream) -                charUpperStrm = S.map toUpper inputStream+                charUpperStrm = fmap toUpper inputStream -            genList <- run $ S.toList genStrm-            charList <- run $ S.toList charUpperStrm+            genList <- run $ S.fold Fold.toList genStrm+            charList <- run $ S.fold Fold.toList charUpperStrm             listEquals (==) genList charList  processChunksConsumePartialInput :: Property@@ -287,17 +286,19 @@             let                 inputStream = S.fromList ls -                procOutput = AS.concat $-                    Proc.processChunks+                procOutput = concatArr $+                    Proc.pipeChunks                     path                     ["-n", "1"]-                    (AS.arraysOf arrayChunkSize inputStream)+                    (S.chunksOf arrayChunkSize inputStream)                  headLine =-                      S.splitWithSuffix (== 10) Fold.toList inputStream-                    & S.head+                      S.foldMany+                          (Fold.takeEndBy (== 10) Fold.toList)+                          inputStream+                    & S.fold Fold.one -            procList <- run $ S.toList procOutput+            procList <- run $ S.fold Fold.toList procOutput             expectedList <- run $ fmap (fromMaybe []) headLine             listEquals (==) procList expectedList @@ -307,8 +308,8 @@     where      runProcess = do-        S.drain $-            Proc.processChunks+        S.fold Fold.drain $+            Proc.pipeChunks                 interpreterFile [interpreterArg, executableFileFail] S.nil         return False @@ -320,8 +321,8 @@     where      runProcess = do-        S.drain $-            Proc.processChunks+        S.fold Fold.drain $+            Proc.pipeChunks             interpreterFile             [interpreterArg, executableFilePass]             (S.nilM $ throwM (SimpleError failErrorMessage))@@ -329,48 +330,48 @@      checkException (SimpleError err) = return (err == failErrorMessage) -processBytes'1 :: Property-processBytes'1 =+pipeBytesEither1 :: Property+pipeBytesEither1 =     forAll (listOf (choose(_a, _z))) $ \ls ->         monadicIO $ do             trBinary <- run $ which "tr"             let                 inputStream = S.fromList ls-                genStrm = S.rights $ Proc.processBytes'+                genStrm = S.catRights $ Proc.pipeBytesEither                             trBinary                             ["[a-z]", "[A-Z]"]                             inputStream-                charUpperStrm = S.map toUpper inputStream+                charUpperStrm = fmap toUpper inputStream -            genList <- run $ S.toList genStrm-            charList <- run $ S.toList charUpperStrm+            genList <- run $ S.fold Fold.toList genStrm+            charList <- run $ S.fold Fold.toList charUpperStrm             listEquals (==) genList charList -processBytes'2 :: Property-processBytes'2 =+pipeBytesEither2 :: Property+pipeBytesEither2 =     forAll (listOf (choose(_a, _z))) $ \ls ->         monadicIO $ do             let                 inputStream = S.fromList ls-                outStream = S.lefts $-                    Proc.processBytes'+                outStream = S.catLefts $+                    Proc.pipeBytesEither                     interpreterFile                     [interpreterArg, executableFile, "[a-z]", "[A-Z]"]                     inputStream -                charUpperStrm = S.map toUpper inputStream+                charUpperStrm = fmap toUpper inputStream -            genList <- run $ S.toList outStream-            charList <- run $ S.toList charUpperStrm+            genList <- run $ S.fold Fold.toList outStream+            charList <- run $ S.fold Fold.toList charUpperStrm             listEquals (==) genList charList -processBytes'3 :: Property-processBytes'3 = monadicIO $ run checkFailAction+pipeBytesEither3 :: Property+pipeBytesEither3 = monadicIO $ run checkFailAction     where      action = do-        S.drain $-            Proc.processBytes'+        S.fold Fold.drain $+            Proc.pipeBytesEither                 interpreterFile [interpreterArg, executableFileFail] S.nil         return False @@ -379,13 +380,13 @@      checkFailAction = catch action failAction -processBytes'4 :: Property-processBytes'4 = monadicIO $ run checkFailAction+pipeBytesEither4 :: Property+pipeBytesEither4 = monadicIO $ run checkFailAction     where      action = do-        S.drain $-            Proc.processBytes'+        S.fold Fold.drain $+            Proc.pipeBytesEither             interpreterFile             [interpreterArg, executableFilePass]             (S.nilM $ throwM (SimpleError failErrorMessage))@@ -396,50 +397,50 @@      checkFailAction = catch action failAction -processChunks'1 :: Property-processChunks'1 =+pipeChunksEither1 :: Property+pipeChunksEither1 =     forAll (listOf (choose(_a, _z))) $ \ls ->         monadicIO $ do             trBinary <- run $ which "tr"             let                 inputStream = S.fromList ls -                genStrm = AS.concat $ S.rights $-                    Proc.processChunks'+                genStrm = concatArr $ S.catRights $+                    Proc.pipeChunksEither                     trBinary                     ["[a-z]", "[A-Z]"]-                    (AS.arraysOf arrayChunkSize inputStream)+                    (S.chunksOf arrayChunkSize inputStream) -                charUpperStrm = S.map toUpper inputStream+                charUpperStrm = fmap toUpper inputStream -            genList <- run $ S.toList genStrm-            charList <- run $ S.toList charUpperStrm+            genList <- run $ S.fold Fold.toList genStrm+            charList <- run $ S.fold Fold.toList charUpperStrm             listEquals (==) genList charList -processChunks'2 :: Property-processChunks'2 =+pipeChunksEither2 :: Property+pipeChunksEither2 =     forAll (listOf (choose(_a, _z))) $ \ls ->         monadicIO $ do             let                 inputStream = S.fromList ls-                outStream = AS.concat $ S.lefts $-                    Proc.processChunks'+                outStream = concatArr $ S.catLefts $+                    Proc.pipeChunksEither                     interpreterFile                     [interpreterArg, executableFile, "[a-z]", "[A-Z]"]-                    (AS.arraysOf arrayChunkSize inputStream)+                    (S.chunksOf arrayChunkSize inputStream) -                charUpperStrm = S.map toUpper inputStream+                charUpperStrm = fmap toUpper inputStream -            genList <- run $ S.toList outStream-            charList <- run $ S.toList charUpperStrm+            genList <- run $ S.fold Fold.toList outStream+            charList <- run $ S.fold Fold.toList charUpperStrm             listEquals (==) genList charList -processChunks'3 :: Property-processChunks'3 = monadicIO $ run checkFailAction+pipeChunksEither3 :: Property+pipeChunksEither3 = monadicIO $ run checkFailAction     where      action = do-        S.drain $ Proc.processChunks'+        S.fold Fold.drain $ Proc.pipeChunksEither             interpreterFile  [interpreterArg, executableFileFail] S.nil         return False @@ -448,13 +449,13 @@      checkFailAction = catch action failAction -processChunks'4 :: Property-processChunks'4 = monadicIO $ run checkFailAction+pipeChunksEither4 :: Property+pipeChunksEither4 = monadicIO $ run checkFailAction     where      action = do-        S.drain $-            Proc.processChunks'+        S.fold Fold.drain $+            Proc.pipeChunksEither             interpreterFile             [interpreterArg, executableFilePass]             (S.nilM $ throwM (SimpleError failErrorMessage))@@ -467,6 +468,15 @@  main :: IO () main = do+    -- Nix does not have "/usr/bin/env", so the execution of test executables+    -- fails. We can create the executables as temporary files to avoid this+    -- issue. Even creating Haskell executables does not work because of+    -- https://github.com/haskell/cabal/issues/7577 .+    r <- doesFileExist "/usr/bin/env"+    unless r $ do+        putStrLn $ "/usr/bin/env does not exist, skipping tests."+        exitSuccess+     hspec $ do         describe "Streamly.System.Process" $ do             -- XXX Add a test for garbage collection case. Also check whether@@ -474,15 +484,15 @@             --             -- Keep the tests in dependency order so that we test the basic             -- things first.-            describe "processChunks'" $ do+            describe "pipeChunksEither" $ do                 prop-                    "AS.concat $ processChunks' tr = map toUpper"-                    processChunks'1+                    "concatArr $ pipeChunksEither tr = map toUpper"+                    pipeChunksEither1                 prop-                    "error stream of processChunks' tr = map toUpper"-                    processChunks'2-                prop "processChunks' on failing executable" processChunks'3-                prop "processChunks' using error stream" processChunks'4+                    "error stream of pipeChunksEither tr = map toUpper"+                    pipeChunksEither2+                prop "pipeChunksEither on failing executable" pipeChunksEither3+                prop "pipeChunksEither using error stream" pipeChunksEither4              describe "processChunks" $ do                 prop "consumeAllInput" processChunksConsumeAllInput@@ -491,24 +501,24 @@                 prop "inputFailure" processChunksInputFailure              -- based on processChunks-            describe "processBytes'" $ do-                prop "processBytes' tr = map toUpper" processBytes'1+            describe "pipeBytesEither" $ do+                prop "pipeBytesEither tr = map toUpper" pipeBytesEither1                 prop-                    "error stream of processBytes' tr = map toUpper"-                    processBytes'2-                prop "processBytes' on failing executable" processBytes'3-                prop "processBytes' using error stream" processBytes'4+                    "error stream of pipeBytesEither tr = map toUpper"+                    pipeBytesEither2+                prop "pipeBytesEither on failing executable" pipeBytesEither3+                prop "pipeBytesEither using error stream" pipeBytesEither4 -            describe "processBytes" $ do-                prop "processBytes tr = map toUpper" processBytes1-                prop "processBytes on failing executable" processBytes2-                prop "processBytes using error stream" processBytes3+            describe "pipeBytes" $ do+                prop "pipeBytes tr = map toUpper" pipeBytes1+                prop "pipeBytes on failing executable" pipeBytes2+                prop "pipeBytes using error stream" pipeBytes3 -            -- Based on processBytes/Chunks-            describe "toChunks'" $ do-                prop "toChunks' cat = FH.toChunks" toChunks1-                prop "toChunks' on failing executable" toChunks2+            -- Based on pipeBytes/Chunks+            describe "toChunksEither" $ do+                prop "toChunksEither cat = FH.toChunks" toChunks1+                prop "toChunksEither on failing executable" toChunks2 -            describe "toBytes'" $ do-                prop "toBytes' cat = FH.toBytes" toBytes1-                prop "toBytes' on failing executable" toBytes2+            describe "toBytesEither" $ do+                prop "toBytesEither cat = FH.toBytes" toBytes1+                prop "toBytesEither on failing executable" toBytes2