polysemy-process 0.8.0.1 → 0.9.0.0

raw patch · 15 files changed

+470/−172 lines, 15 filesdep ~basedep ~incipit-coredep ~polysemy-concPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, incipit-core, polysemy-conc, polysemy-resume, polysemy-test, polysemy-time

API changes (from Hackage documentation)

Files

• changelog.md +0/−16
• lib/Polysemy/Process.hs +40/−8
• lib/Polysemy/Process/Data/ProcessOutputParseResult.hs +21/−0
• lib/Polysemy/Process/Effect/Process.hs +19/−26
• lib/Polysemy/Process/Effect/ProcessInput.hs +15/−0
• lib/Polysemy/Process/Effect/ProcessOutput.hs +11/−2
• lib/Polysemy/Process/Interpreter/Process.hs +173/−68
• lib/Polysemy/Process/Interpreter/ProcessInput.hs +22/−0
• lib/Polysemy/Process/Interpreter/ProcessOutput.hs +86/−4
• lib/Polysemy/Process/Interpreter/ProcessStdio.hs +9/−5
• lib/Polysemy/Process/Interpreter/Pty.hs +2/−2
• lib/Polysemy/Process/Interpreter/SystemProcess.hs +12/−2
• polysemy-process.cabal +26/−22
• readme.md +0/−8
• test/Polysemy/Process/Test/ProcessTest.hs +34/−9

@@ -1,16 +0,0 @@-# 0.8.0.0-* Add `ProcessOptions`, replacing the primitive parameters of `Process` interpreters.-* Add an option for `Process` that determines whether to kill the process after exiting the scope.--# 0.6.0.0--* Allow `Process` to emit custom chunks constructed by an interpreter of `ProcessOutput` instead of `ByteString`s-  containing whatever the `Handle` produced.-* Rename stdio interpreters.-* Add helper `resolveExecutable` that looks up names in `$PATH` and ensures the files are executable.-* Add effect `Pty` for interacting with pseudo terminals.-* Add low-level process abstraction effect, `SystemProcess`.--# 0.5.0.0--* Add the effect `Process`, wrapping `System.Process.Typed` using `Scoped`.

@@ -5,9 +5,8 @@    -- * Effects   -- ** Process-  Process (..),+  Process,   recv,-  recvError,   send,   withProcess,   ProcessOptions (ProcessOptions),@@ -15,7 +14,12 @@    -- ** ProcessOutput   ProcessOutput,+  OutputPipe (Stdout, Stderr),+  ProcessOutputParseResult (..), +  -- ** ProcessInput+  ProcessInput,+   -- ** SystemProcess   SystemProcess,   withSystemProcess,@@ -35,13 +39,29 @@   interpretProcessByteStringLines,   interpretProcessText,   interpretProcessTextLines,+  interpretInputOutputProcess,+  interpretInputHandleBuffered,+  interpretInputHandle,+  interpretOutputHandleBuffered,+  interpretOutputHandle,+  interpretProcessIO,+  interpretProcessHandles,+  interpretProcessCurrent,    -- ** ProcessOutput+  interpretProcessOutputIgnore,   interpretProcessOutputId,+  interpretProcessOutputLeft,+  interpretProcessOutputRight,   interpretProcessOutputLines,   interpretProcessOutputText,   interpretProcessOutputTextLines,+  interpretProcessOutputIncremental, +  -- ** ProcessInput+  interpretProcessInputId,+  interpretProcessInputText,+   -- ** SystemProcess   interpretSystemProcessWithProcess,   interpretSystemProcessNativeSingle,@@ -61,24 +81,36 @@  import Polysemy.Process.Data.ProcessKill (ProcessKill (..)) import Polysemy.Process.Data.ProcessOptions (ProcessOptions (ProcessOptions))-import Polysemy.Process.Effect.Process (Process (..), recv, recvError, send, withProcess)-import Polysemy.Process.Effect.ProcessOutput (ProcessOutput)+import Polysemy.Process.Data.ProcessOutputParseResult (ProcessOutputParseResult (..))+import Polysemy.Process.Effect.Process (Process (..), recv, send, withProcess)+import Polysemy.Process.Effect.ProcessInput (ProcessInput)+import Polysemy.Process.Effect.ProcessOutput (OutputPipe (Stderr, Stdout), ProcessOutput) import Polysemy.Process.Effect.Pty (Pty, withPty)-import Polysemy.Process.Effect.SystemProcess (-  SystemProcess,-  withSystemProcess,-  )+import Polysemy.Process.Effect.SystemProcess (SystemProcess, withSystemProcess) import Polysemy.Process.Executable (resolveExecutable) import Polysemy.Process.Interpreter.Process (+  interpretInputHandle,+  interpretInputHandleBuffered,+  interpretInputOutputProcess,+  interpretOutputHandle,+  interpretOutputHandleBuffered,   interpretProcess,   interpretProcessByteString,   interpretProcessByteStringLines,+  interpretProcessCurrent,+  interpretProcessHandles,+  interpretProcessIO,   interpretProcessText,   interpretProcessTextLines,   )+import Polysemy.Process.Interpreter.ProcessInput (interpretProcessInputId, interpretProcessInputText) import Polysemy.Process.Interpreter.ProcessOutput (   interpretProcessOutputId,+  interpretProcessOutputIgnore,+  interpretProcessOutputIncremental,+  interpretProcessOutputLeft,   interpretProcessOutputLines,+  interpretProcessOutputRight,   interpretProcessOutputText,   interpretProcessOutputTextLines,   )

@@ -0,0 +1,21 @@+-- |A data type encoding the result of an incremental parser for process output.+module Polysemy.Process.Data.ProcessOutputParseResult where++import Text.Show (showParen, showString, showsPrec)++-- |An incremental parse result, potentially a partial result containing a continuation function.+data ProcessOutputParseResult a =+  Done { value :: a, leftover :: ByteString }+  |+  Partial { continue :: ByteString -> ProcessOutputParseResult a }+  |+  Fail { error :: Text }++instance Show a => Show (ProcessOutputParseResult a) where+  showsPrec d = \case+    Done {..} ->+      showParen (d > 10) (showString "Done { value = " <> showsPrec 11 value <> showString ", leftover = " <> showsPrec 11 leftover <> showString " }")+    Partial _ ->+      showString "Partial"+    Fail e ->+      showParen (d > 10) (showString "Fail { error = " <> showString (show e) <> showString " }")

@@ -4,12 +4,12 @@ module Polysemy.Process.Effect.Process where  import Polysemy.Conc.Effect.Scoped (Scoped, scoped)+import Polysemy.Input (Input (Input))+import Polysemy.Output (Output (Output)) import Polysemy.Resume (interpretResumable, restop, type (!!)) import Prelude hiding (send)-import Polysemy.Input (Input(Input))-import Polysemy.Output (Output(Output)) --- |Abstraction of a process with stdin/stdout/stderr.+-- |Abstraction of a process with input and output. -- -- This effect is intended to be used in a scoped manner: --@@ -19,7 +19,7 @@ -- import Polysemy.Process -- import qualified System.Process.Typed as System ----- prog :: Member (Scoped resource (Process Text Text e !! err)) r => Sem r Text+-- prog :: Member (Scoped resource (Process Text Text !! err)) r => Sem r Text -- prog = --  resumeAs "failed" do --    withProcess do@@ -31,50 +31,43 @@ --   out <- runConc $ interpretProcessNative (System.proc "cat" []) prog --   putStrLn out -- @-data Process i o e :: Effect where-  Recv :: Process i o e m o-  RecvError :: Process i o e m e-  Send :: i -> Process i o e m ()+data Process i o :: Effect where+  Recv :: Process i o m o+  Send :: i -> Process i o m ()  makeSem_ ''Process --- |Obtain a chunk of stdout.+-- |Obtain a chunk of output. recv ::-  ∀ i o e r .-  Member (Process i o e) r =>+  ∀ i o r .+  Member (Process i o) r =>   Sem r o --- |Obtain a chunk of stderr.-recvError ::-  ∀ i o e r .-  Member (Process i o e) r =>-  Sem r e- -- |Send data to stdin. send ::-  ∀ i o e r .-  Member (Process i o e) r =>+  ∀ i o r .+  Member (Process i o) r =>   i ->   Sem r ()  -- |Create a scoped resource for 'Process'. withProcess ::-  ∀ resource i o e r .-  Member (Scoped resource (Process i o e)) r =>-  InterpreterFor (Process i o e) r+  ∀ resource i o r .+  Member (Scoped resource (Process i o)) r =>+  InterpreterFor (Process i o) r withProcess =   scoped @resource  -- |Convert 'Output' and 'Input' to 'Process'. runProcessIO ::-  ∀ i o e err r .-  Member (Process i o e !! err) r =>+  ∀ i o err r .+  Member (Process i o !! err) r =>   InterpretersFor [Output i !! err, Input o !! err] r runProcessIO =   interpretResumable \case     Input ->-      restop @err @(Process i o e) (recv @i @o @e)+      restop @err @(Process i o) (recv @i @o)   .   interpretResumable \case     Output o ->-      restop @err @(Process i o e) (send @i @o @e o)+      restop @err @(Process i o) (send @i @o o)

@@ -0,0 +1,15 @@+{-# options_haddock prune #-}++-- |Description: ProcessInput effect, Internal.+module Polysemy.Process.Effect.ProcessInput where++-- |This effect is used by the effect 'Polysemy.Process.Process' to encode values for process input.+-- example using a parser.+data ProcessInput a :: Effect where+  -- |Encode a value for enqueueing it to a process' stdin.+  Encode ::+    -- |The value to encode.+    a ->+    ProcessInput a m ByteString++makeSem ''ProcessInput

@@ -3,11 +3,20 @@ -- |Description: ProcessOutput effect, Internal. module Polysemy.Process.Effect.ProcessOutput where +-- |Kind tag for selecting the 'ProcessOutput' handler for stdout/stderr.+data OutputPipe =+  -- |Tag for stdout.+  Stdout+  |+  -- |Tag for stderr.+  Stderr+  deriving stock (Eq, Show)+ -- |This effect is used by the effect 'Polysemy.Process.Process' to accumulate and decode chunks of 'ByteString's, for -- example using a parser. -- The interpreter may be stateful or stateless, since the constructor 'Chunk' is expected to be called with both the -- accumulated unprocessed output as well as the new chunk.-data ProcessOutput a :: Effect where+data ProcessOutput (p :: OutputPipe) a :: Effect where   -- |Add a chunk of output to the accumulator, returning any number of successfully parsed values and the leftover   -- output.   Chunk ::@@ -15,6 +24,6 @@     ByteString ->     -- |The new chunk read from the process.     ByteString ->-    ProcessOutput a m ([a], ByteString)+    ProcessOutput p a m ([a], ByteString)  makeSem ''ProcessOutput

@@ -4,6 +4,7 @@ module Polysemy.Process.Interpreter.Process where  import Control.Concurrent.STM.TBMQueue (TBMQueue)+import Data.ByteString (hGetSome, hPut) import qualified Polysemy.Conc as Conc import Polysemy.Conc.Async (withAsync_) import qualified Polysemy.Conc.Data.QueueResult as QueueResult@@ -13,20 +14,27 @@ import Polysemy.Conc.Effect.Scoped (Scoped) import Polysemy.Conc.Interpreter.Queue.TBM (interpretQueueTBMWith, withTBMQueue) import Polysemy.Conc.Interpreter.Scoped (interpretScopedResumableWith_)-import Polysemy.Resume (Stop, resumeOr, resume_, stop, type (!!))+import Polysemy.Input (Input (Input))+import Polysemy.Output (Output (Output))+import Polysemy.Resume (Stop, interpretResumable, resumeOr, resume_, stop, stopNote, type (!!)) import Prelude hiding (fromException)+import System.IO (BufferMode (NoBuffering), Handle, hSetBuffering, stdin, stdout)  import Polysemy.Process.Data.ProcessError (ProcessError (Terminated)) import Polysemy.Process.Data.ProcessKill (ProcessKill (KillAfter, KillImmediately, KillNever)) import Polysemy.Process.Data.ProcessOptions (ProcessOptions (ProcessOptions)) import qualified Polysemy.Process.Effect.Process as Process import Polysemy.Process.Effect.Process (Process)+import qualified Polysemy.Process.Effect.ProcessInput as ProcessInput+import Polysemy.Process.Effect.ProcessInput (ProcessInput) import qualified Polysemy.Process.Effect.ProcessOutput as ProcessOutput-import Polysemy.Process.Effect.ProcessOutput (ProcessOutput)+import Polysemy.Process.Effect.ProcessOutput (OutputPipe (Stderr, Stdout), ProcessOutput) import qualified Polysemy.Process.Effect.SystemProcess as SystemProcess import Polysemy.Process.Effect.SystemProcess (SystemProcess, withSystemProcess)+import Polysemy.Process.Interpreter.ProcessInput (interpretProcessInputId, interpretProcessInputText) import Polysemy.Process.Interpreter.ProcessOutput (   interpretProcessOutputId,+  interpretProcessOutputIgnore,   interpretProcessOutputLines,   interpretProcessOutputText,   interpretProcessOutputTextLines,@@ -44,26 +52,24 @@   Err { unErr :: a }   deriving stock (Eq, Show) -data ProcessQueues o e =+data ProcessQueues i o =   ProcessQueues {-    pqIn :: TBMQueue (In ByteString),-    pqOut :: TBMQueue (Out o),-    pqErr :: TBMQueue (Err e)+    pqIn :: TBMQueue (In i),+    pqOut :: TBMQueue (Out o)   }  interpretQueues ::   Members [Resource, Race, Embed IO] r =>-  ProcessQueues o e ->-  InterpretersFor [Queue (In ByteString), Queue (Out o), Queue (Err e)] r-interpretQueues (ProcessQueues inQ outQ errQ) =-  interpretQueueTBMWith errQ .+  ProcessQueues i o ->+  InterpretersFor [Queue (In i), Queue (Out o)] r+interpretQueues (ProcessQueues inQ outQ) =   interpretQueueTBMWith outQ .   interpretQueueTBMWith inQ  handleProcessWithQueues ::-  ∀ o e m r a .-  Members [Queue (In ByteString), Queue (Out o), Queue (Err e), Stop ProcessError] r =>-  Process ByteString o e m a ->+  ∀ i o m r a .+  Members [Queue (In i), Queue (Out o), Stop ProcessError] r =>+  Process i o m a ->   Sem r a handleProcessWithQueues = \case   Process.Recv ->@@ -74,67 +80,59 @@         stop (Terminated "impossible: empty")       QueueResult.Success (Out msg) ->         pure msg-  Process.RecvError ->-    Queue.read >>= \case-      QueueResult.Closed ->-        stop (Terminated "closed")-      QueueResult.NotAvailable ->-        stop (Terminated "impossible: empty")-      QueueResult.Success (Err msg) ->-        pure msg   Process.Send msg -> do-    whenM (Queue.closed @(In ByteString)) (stop (Terminated "closed"))+    whenM (Queue.closed @(In i)) (stop (Terminated "closed"))     Queue.write (In msg)  withSTMResources ::-  ∀ o e r a .+  ∀ i o r a .   Members [Resource, Embed IO] r =>   Int ->-  (ProcessQueues o e -> Sem r a) ->+  (ProcessQueues i o -> Sem r a) ->   Sem r a withSTMResources qSize action = do   withTBMQueue qSize \ inQ ->     withTBMQueue qSize \ outQ ->-      withTBMQueue qSize \ errQ ->-        action (ProcessQueues inQ outQ errQ)+      action (ProcessQueues inQ outQ)  withQueues ::+  ∀ i o r .   Members [Race, Resource, Embed IO] r =>   Int ->-  InterpretersFor [Queue (In ByteString), Queue (Out o), Queue (Err e)] r+  InterpretersFor [Queue (In i), Queue (Out o)] r withQueues qSize action =   withSTMResources qSize \ qs -> interpretQueues qs action  outputQueue ::-  ∀ pipe chunk err r .-  Coercible (pipe chunk) chunk =>-  Members [SystemProcess !! err, ProcessOutput chunk, Queue (pipe chunk), Embed IO] r =>+  ∀ p chunk err eff r .+  Members [eff !! err, ProcessOutput p chunk, Queue (Out chunk), Embed IO] r =>   Bool ->-  Sem (SystemProcess : r) ByteString ->+  Sem (eff : r) ByteString ->   Sem r () outputQueue discardWhenFull readChunk = do   spin ""   where     spin buffer =-      resumeOr @err readChunk (write buffer) (const (Queue.close @(pipe chunk)))+      resumeOr @err readChunk (write buffer) (const (Queue.close @(Out chunk)))     write buffer msg = do-      (chunks, newBuffer) <- ProcessOutput.chunk buffer msg-      for_ chunks \ (coerce @chunk @(pipe chunk) -> c) ->+      (chunks, newBuffer) <- ProcessOutput.chunk @p @chunk buffer msg+      for_ chunks \ (Out -> c) ->         if discardWhenFull then void (Queue.tryWrite c) else Queue.write c       spin newBuffer  inputQueue ::-  ∀ err r .-  Members [SystemProcess !! err, Queue (In ByteString), Embed IO] r =>-  (ByteString -> Sem (SystemProcess : r) ()) ->+  ∀ i err eff r .+  Members [eff !! err, ProcessInput i, Queue (In i), Embed IO] r =>+  (ByteString -> Sem (eff : r) ()) ->   Sem r () inputQueue writeChunk =   spin   where     spin =-      Queue.read >>= \case-        QueueResult.Success (In msg) ->-          resumeOr @err (writeChunk msg) (const spin) (const (Queue.close @(In ByteString)))+      Queue.read @(In i) >>= \case+        QueueResult.Success (In msg) -> do+          bytes <- ProcessInput.encode msg+          resumeOr @err (writeChunk bytes) (const spin) (const (Queue.close @(In i)))         _ ->           unit @@ -161,78 +159,185 @@     void SystemProcess.pid     handleKill kill -type ScopeEffects o e err =-  [Queue (In ByteString), Queue (Out o), Queue (Err e), SystemProcess !! err]+type ScopeEffects i o err =+  [Queue (In i), Queue (Out o), SystemProcess !! err]  scope ::-  ∀ o e resource err r .+  ∀ i o resource err r .   Member (Scoped resource (SystemProcess !! err)) r =>-  Members [ProcessOutput e, ProcessOutput o, Resource, Race, Async, Embed IO] r =>+  Members [ProcessInput i, ProcessOutput 'Stdout o, ProcessOutput 'Stderr o, Resource, Race, Async, Embed IO] r =>   ProcessOptions ->-  InterpretersFor (ScopeEffects o e err) r+  InterpretersFor (ScopeEffects i o err) r scope (ProcessOptions discard qSize kill) =   withSystemProcess @resource .   withQueues qSize .-  withAsync_ (outputQueue @Err @e @err discard SystemProcess.readStderr) .-  withAsync_ (outputQueue @Out @o @err discard SystemProcess.readStdout) .-  withAsync_ (inputQueue @err SystemProcess.writeStdin) .+  withAsync_ (outputQueue @'Stderr @o @err @SystemProcess discard SystemProcess.readStderr) .+  withAsync_ (outputQueue @'Stdout @o @err @SystemProcess discard SystemProcess.readStdout) .+  withAsync_ (inputQueue @i @err @SystemProcess SystemProcess.writeStdin) .   withKill @err kill  -- |Interpret 'Process' with a system process resource whose file descriptors are connected to three 'TBMQueue's, -- deferring decoding of stdout and stderr to the interpreters of two 'ProcessOutput' effects. interpretProcess ::-  ∀ resource err o e r .+  ∀ resource err i o r .   Member (Scoped resource (SystemProcess !! err)) r =>-  Members [ProcessOutput o, ProcessOutput e, Resource, Race, Async, Embed IO] r =>+  Members [ProcessOutput 'Stdout o, ProcessOutput 'Stderr o, ProcessInput i, Resource, Race, Async, Embed IO] r =>   ProcessOptions ->-  InterpreterFor (Scoped () (Process ByteString o e) !! ProcessError) r+  InterpreterFor (Scoped () (Process i o) !! ProcessError) r interpretProcess options =-  interpretScopedResumableWith_ @(ScopeEffects o e err) (scope @o @e @resource options) handleProcessWithQueues+  interpretScopedResumableWith_ @(ScopeEffects i o err) (scope @i @o @resource options) handleProcessWithQueues --- |Interpret 'Process' with a system process resource whose file descriptors are connected to three 'TBMQueue's,+-- |Interpret 'Process' with a system process resource whose stdin/stdout are connected to two 'TBMQueue's, -- producing 'ByteString's.+-- Silently discards stderr. interpretProcessByteString ::   ∀ resource err r .   Members [Scoped resource (SystemProcess !! err), Resource, Race, Async, Embed IO] r =>   ProcessOptions ->-  InterpreterFor (Scoped () (Process ByteString ByteString ByteString) !! ProcessError) r+  InterpreterFor (Scoped () (Process ByteString ByteString) !! ProcessError) r interpretProcessByteString options =-  interpretProcessOutputId .+  interpretProcessOutputIgnore @'Stderr @ByteString .+  interpretProcessOutputId @'Stdout .+  interpretProcessInputId .   interpretProcess @resource @err options .-  raiseUnder+  raiseUnder3 --- |Interpret 'Process' with a system process resource whose file descriptors are connected to three 'TBMQueue's,+-- |Interpret 'Process' with a system process resource whose stdin/stdout are connected to two 'TBMQueue's, -- producing chunks of lines of 'ByteString's.+-- Silently discards stderr. interpretProcessByteStringLines ::   ∀ resource err r .   Members [Scoped resource (SystemProcess !! err), Resource, Race, Async, Embed IO] r =>   ProcessOptions ->-  InterpreterFor (Scoped () (Process ByteString ByteString ByteString) !! ProcessError) r+  InterpreterFor (Scoped () (Process ByteString ByteString) !! ProcessError) r interpretProcessByteStringLines options =-  interpretProcessOutputLines .+  interpretProcessOutputIgnore @'Stderr @ByteString .+  interpretProcessOutputLines @'Stdout .+  interpretProcessInputId .   interpretProcess @resource @err options .-  raiseUnder+  raiseUnder3 --- |Interpret 'Process' with a system process resource whose file descriptors are connected to three 'TBMQueue's,+-- |Interpret 'Process' with a system process resource whose stdin/stdout are connected to two 'TBMQueue's, -- producing 'Text's.+-- Silently discards stderr. interpretProcessText ::   ∀ resource err r .   Members [Scoped resource (SystemProcess !! err), Resource, Race, Async, Embed IO] r =>   ProcessOptions ->-  InterpreterFor (Scoped () (Process ByteString Text Text) !! ProcessError) r+  InterpreterFor (Scoped () (Process Text Text) !! ProcessError) r interpretProcessText options =-  interpretProcessOutputText .+  interpretProcessOutputIgnore @'Stderr @Text .+  interpretProcessOutputText @'Stdout .+  interpretProcessInputText .   interpretProcess @resource @err options .-  raiseUnder+  raiseUnder3 --- |Interpret 'Process' with a system process resource whose file descriptors are connected to three 'TBMQueue's,+-- |Interpret 'Process' with a system process resource whose stdin/stdout are connected to two 'TBMQueue's, -- producing chunks of lines of 'Text's.+-- Silently discards stderr. interpretProcessTextLines ::   ∀ resource err r .   Members [Scoped resource (SystemProcess !! err), Resource, Race, Async, Embed IO] r =>   ProcessOptions ->-  InterpreterFor (Scoped () (Process ByteString Text Text) !! ProcessError) r+  InterpreterFor (Scoped () (Process Text Text) !! ProcessError) r interpretProcessTextLines options =-  interpretProcessOutputTextLines .+  interpretProcessOutputIgnore @'Stderr @Text .+  interpretProcessOutputTextLines @'Stdout .+  interpretProcessInputText .   interpretProcess @resource @err options .-  raiseUnder+  raiseUnder3++-- |Reinterpret 'Input' and 'Output' as 'Process'.+interpretInputOutputProcess ::+  ∀ i o r .+  Member (Process i o) r =>+  InterpretersFor [Input o, Output i] r+interpretInputOutputProcess =+  runOutputSem (Process.send @i @o) .+  runInputSem (Process.recv @i @o)++-- |Interpret 'Input ByteString' by polling a 'Handle' and stopping with 'ProcessError' when it fails.+interpretInputHandleBuffered ::+  Member (Embed IO) r =>+  Handle ->+  InterpreterFor (Input ByteString !! ProcessError) r+interpretInputHandleBuffered handle =+  interpretResumable \case+    Input ->+      stopNote (Terminated "handle closed") =<< tryMaybe (hGetSome handle 4096)++-- |Interpret 'Input ByteString' by polling a 'Handle' and stopping with 'ProcessError' when it fails.+-- This variant deactivates buffering for the 'Handle'.+interpretInputHandle ::+  Member (Embed IO) r =>+  Handle ->+  InterpreterFor (Input ByteString !! ProcessError) r+interpretInputHandle handle sem = do+  tryAny_ (hSetBuffering handle NoBuffering)+  interpretInputHandleBuffered handle sem++-- |Interpret 'Output ByteString' by writing to a 'Handle' and stopping with 'ProcessError' when it fails.+interpretOutputHandleBuffered ::+  Member (Embed IO) r =>+  Handle ->+  InterpreterFor (Output ByteString !! ProcessError) r+interpretOutputHandleBuffered handle =+  interpretResumable \case+    Output o ->+      stopNote (Terminated "handle closed") =<< tryMaybe (hPut handle o)++-- |Interpret 'Output ByteString' by writing to a 'Handle' and stopping with 'ProcessError' when it fails.+-- This variant deactivates buffering for the 'Handle'.+interpretOutputHandle ::+  Member (Embed IO) r =>+  Handle ->+  InterpreterFor (Output ByteString !! ProcessError) r+interpretOutputHandle handle sem = do+  tryAny_ (hSetBuffering handle NoBuffering)+  interpretOutputHandleBuffered handle sem++-- |Interpret 'Process' in terms of 'Input' and 'Output'.+-- Since the @i@ and @o@ parameters correspond to the abstraction of stdio fds of an external system process, @i@ is+-- written by 'Output' and @o@ is read from 'Input'.+-- This is useful to abstract the current process's stdio as an external process, with input and output swapped.+interpretProcessIO ::+  ∀ i o ie oe r .+  Members [Input ByteString !! ie, Output ByteString !! oe] r =>+  Members [ProcessInput i, ProcessOutput 'Stdout o, Resource, Race, Async, Embed IO] r =>+  ProcessOptions ->+  InterpreterFor (Process i o !! ProcessError) r+interpretProcessIO (ProcessOptions discard qSize _) =+  withQueues @i @o qSize .+  withAsync_ (outputQueue @'Stdout @o @ie @(Input ByteString) discard input) .+  withAsync_ (inputQueue @i @oe @(Output ByteString) output) .+  interpretResumable handleProcessWithQueues .+  raiseUnder2++-- |Interpret 'Process' in terms of two 'Handle's.+-- This is useful to abstract the current process's stdio as an external process, with input and output swapped.+-- The first 'Handle' argument corresponds to the @o@ parameter, the second one to @i@, despite the first one usually+-- being the current process's stdin.+-- This is due to 'Process' abstracting an external process to whose stdin would be /written/, while the current one's+-- is /read/.+interpretProcessHandles ::+  ∀ i o r .+  Members [ProcessInput i, ProcessOutput 'Stdout o, Resource, Race, Async, Embed IO] r =>+  ProcessOptions ->+  Handle ->+  Handle ->+  InterpreterFor (Process i o !! ProcessError) r+interpretProcessHandles options hIn hOut =+  interpretOutputHandle hOut .+  interpretInputHandle hIn .+  interpretProcessIO @i @o @ProcessError @ProcessError options .+  raiseUnder2++-- |Interpret 'Process' using the current process's stdin and stdout.+-- This mirrors the usual abstraction of an external process, to whose stdin would be /written/, while the current one's+-- is /read/.+interpretProcessCurrent ::+  Members [ProcessInput i, ProcessOutput 'Stdout o, Resource, Race, Async, Embed IO] r =>+  ProcessOptions ->+  InterpreterFor (Process i o !! ProcessError) r+interpretProcessCurrent options =+  interpretProcessHandles options stdin stdout

@@ -0,0 +1,22 @@+{-# options_haddock prune #-}++-- |Description: ProcessInput Interpreters, Internal+module Polysemy.Process.Interpreter.ProcessInput where++import Polysemy.Process.Effect.ProcessInput (ProcessInput (Encode))++-- |Interpret 'ProcessInput' by passing 'ByteString' through.+interpretProcessInputId :: InterpreterFor (ProcessInput ByteString) r+interpretProcessInputId =+  interpret \case+    Encode value ->+      pure value+{-# inline interpretProcessInputId #-}++-- |Interpret 'ProcessInput' by UTF-8-encoding 'Text'.+interpretProcessInputText :: InterpreterFor (ProcessInput Text) r+interpretProcessInputText =+  interpret \case+    Encode value ->+      pure (encodeUtf8 value)+{-# inline interpretProcessInputText #-}

@@ -5,16 +5,52 @@  import qualified Data.ByteString as ByteString +import Polysemy.Process.Data.ProcessOutputParseResult (ProcessOutputParseResult (Done, Fail, Partial)) import Polysemy.Process.Effect.ProcessOutput (ProcessOutput (Chunk))+import qualified Polysemy.Process.Effect.ProcessOutput as ProcessOutput +-- |Interpret 'ProcessOutput' by discarding any output.+interpretProcessOutputIgnore ::+  ∀ p a r .+  InterpreterFor (ProcessOutput p a) r+interpretProcessOutputIgnore =+  interpret \case+    Chunk _ _ ->+      pure ([], "")+{-# inline interpretProcessOutputIgnore #-}+ -- |Interpret 'ProcessOutput' by immediately emitting raw 'ByteString's without accumulation.-interpretProcessOutputId :: InterpreterFor (ProcessOutput ByteString) r+interpretProcessOutputId ::+  ∀ p r .+  InterpreterFor (ProcessOutput p ByteString) r interpretProcessOutputId =   interpret \case     Chunk buffer new ->       pure ([buffer <> new], "") {-# inline interpretProcessOutputId #-} +-- |Transformer for 'ProcessOutput' that lifts results into 'Left', creating 'ProcessOutput p (Either a b)' from+-- 'ProcessOutput p a'.+interpretProcessOutputLeft ::+  ∀ p a b r .+  Member (ProcessOutput p a) r =>+  InterpreterFor (ProcessOutput p (Either a b)) r+interpretProcessOutputLeft =+  interpret \case+    Chunk buf new ->+      first (fmap Left) <$> ProcessOutput.chunk @p buf new++-- |Transformer for 'ProcessOutput' that lifts results into 'Right', creating 'ProcessOutput p (Either a b)' from+-- 'ProcessOutput p b'.+interpretProcessOutputRight ::+  ∀ p a b r .+  Member (ProcessOutput p b) r =>+  InterpreterFor (ProcessOutput p (Either a b)) r+interpretProcessOutputRight =+  interpret \case+    Chunk buf new ->+      first (fmap Right) <$> ProcessOutput.chunk @p buf new+ splitLines :: ByteString -> ByteString -> ([ByteString], ByteString) splitLines buffer new =   second fold (foldr' folder ([], Nothing) parts)@@ -27,7 +63,9 @@       (a : z, Just r)  -- |Interpret 'ProcessOutput' by emitting individual 'ByteString' lines of output.-interpretProcessOutputLines :: InterpreterFor (ProcessOutput ByteString) r+interpretProcessOutputLines ::+  ∀ p r .+  InterpreterFor (ProcessOutput p ByteString) r interpretProcessOutputLines =   interpret \case     Chunk buffer new ->@@ -35,7 +73,9 @@ {-# inline interpretProcessOutputLines #-}  -- |Interpret 'ProcessOutput' by immediately emitting 'Text' without accumulation.-interpretProcessOutputText :: InterpreterFor (ProcessOutput Text) r+interpretProcessOutputText ::+  ∀ p r .+  InterpreterFor (ProcessOutput p Text) r interpretProcessOutputText =   interpret \case     Chunk buffer new ->@@ -43,9 +83,51 @@ {-# inline interpretProcessOutputText #-}  -- |Interpret 'ProcessOutput' by emitting individual 'Text' lines of output.-interpretProcessOutputTextLines :: InterpreterFor (ProcessOutput Text) r+interpretProcessOutputTextLines ::+  ∀ p r .+  InterpreterFor (ProcessOutput p Text) r interpretProcessOutputTextLines =   interpret \case     Chunk buffer new ->       pure (first (fmap decodeUtf8) (splitLines buffer new)) {-# inline interpretProcessOutputTextLines #-}++type Parser a =+  ByteString -> ProcessOutputParseResult a++-- |Internal helper for 'interpretProcessOutputIncremental' that repeatedly parses elements from a chunk until the+-- parser returns a failure or a partial result.+parseMany ::+  Parser a ->+  Maybe (Parser a) ->+  ByteString ->+  (Maybe (Parser a), ([Either Text a], ByteString))+parseMany parse =+  spin id+  where+    spin cons cont = \case+      "" ->+        (cont, (cons [], ""))+      chunk ->+        case fromMaybe parse cont chunk of+          Fail e ->+            (Nothing, (cons [Left e], ""))+          Partial c ->+            (Just c, (cons [], ""))+          Done a rest ->+            spin (cons . (Right a :)) Nothing rest++-- |Whenever a chunk of output arrives, call the supplied incremental parser whose result must be converted to+-- 'ProcessOutputParseResult'.+-- If a partial parse result is produced, it is stored in the state and resumed when the next chunk is available.+-- If parsing an @a@ succeeds, the parser recurses until it fails.+interpretProcessOutputIncremental ::+  ∀ p a r .+  (ByteString -> ProcessOutputParseResult a) ->+  InterpreterFor (ProcessOutput p (Either Text a)) r+interpretProcessOutputIncremental parse =+  evalState (Nothing :: Maybe (ByteString -> ProcessOutputParseResult a)) .+  atomicStateToState @(Maybe (ByteString -> ProcessOutputParseResult a)) .+  interpret \case+    Chunk buffer new -> atomicState (flip (parseMany parse) (buffer <> new))+  . raiseUnder2

@@ -21,48 +21,52 @@ import Polysemy.Process.Interpreter.SystemProcess (PipesProcess, interpretSystemProcessNative)  -- |Interpret 'Process' as a native 'Polysemy.Process.SystemProcess', producing unaccumulated chunks of 'ByteString'.+-- Silently discards stderr. interpretProcessByteStringNative ::   Members [Resource, Race, Async, Embed IO] r =>-  -- |Whether to discard output chunks if the queue is full.   ProcessOptions ->+  -- |Basic config. The pipes will be changed to 'System.IO.Handle' by the interpreter.   ProcessConfig () () () ->-  InterpreterFor (Scoped () (Process ByteString ByteString ByteString) !! ProcessError) r+  InterpreterFor (Scoped () (Process ByteString ByteString) !! ProcessError) r interpretProcessByteStringNative options conf =   interpretSystemProcessNative conf .   interpretProcessByteString @PipesProcess @SystemProcessError options .   raiseUnder  -- |Interpret 'Process' as a native 'Polysemy.Process.SystemProcess', producing lines of 'ByteString'.+-- Silently discards stderr. interpretProcessByteStringLinesNative ::   Members [Resource, Race, Async, Embed IO] r =>   ProcessOptions ->   -- |Basic config. The pipes will be changed to 'System.IO.Handle' by the interpreter.   ProcessConfig () () () ->-  InterpreterFor (Scoped () (Process ByteString ByteString ByteString) !! ProcessError) r+  InterpreterFor (Scoped () (Process ByteString ByteString) !! ProcessError) r interpretProcessByteStringLinesNative options conf =   interpretSystemProcessNative conf .   interpretProcessByteStringLines @PipesProcess @SystemProcessError options .   raiseUnder  -- |Interpret 'Process' as a native 'Polysemy.Process.SystemProcess', producing unaccumulated chunks of 'Text'.+-- Silently discards stderr. interpretProcessTextNative ::   Members [Resource, Race, Async, Embed IO] r =>   ProcessOptions ->   -- |Basic config. The pipes will be changed to 'System.IO.Handle' by the interpreter.   ProcessConfig () () () ->-  InterpreterFor (Scoped () (Process ByteString Text Text) !! ProcessError) r+  InterpreterFor (Scoped () (Process Text Text) !! ProcessError) r interpretProcessTextNative options conf =   interpretSystemProcessNative conf .   interpretProcessText @PipesProcess @SystemProcessError options .   raiseUnder  -- |Interpret 'Process' as a native 'Polysemy.Process.SystemProcess', producing lines of 'Text'.+-- Silently discards stderr. interpretProcessTextLinesNative ::   Members [Resource, Race, Async, Embed IO] r =>   ProcessOptions ->   -- |Basic config. The pipes will be changed to 'System.IO.Handle' by the interpreter.   ProcessConfig () () () ->-  InterpreterFor (Scoped () (Process ByteString Text Text) !! ProcessError) r+  InterpreterFor (Scoped () (Process Text Text) !! ProcessError) r interpretProcessTextLinesNative options conf =   interpretSystemProcessNative conf .   interpretProcessTextLines @PipesProcess @SystemProcessError options .

@@ -34,8 +34,8 @@   PtyResources ->   Sem r () releasePty PtyResources {primary, pty} = do-  ignoreException (closePty pty)-  ignoreException (closeFd primary)+  tryAny_ (closePty pty)+  tryAny_ (closeFd primary)  withPty ::   Members [Resource, Embed IO] r =>

@@ -101,6 +101,16 @@ processId process =   terminate "getPid returned Nothing" =<< embed (getPid (unsafeProcessHandle process)) +checkEof ::+  Member (Stop SystemProcessError) r =>+  ByteString ->+  Sem r ByteString+checkEof = \case+  "" ->+    stop (SystemProcessError.Terminated "Process terminated, empty ByteString read from handle")+  b ->+    pure b+ -- |Interpret 'SystemProcess' with a concrete 'System.Process' with connected pipes. interpretSystemProcessWithProcess ::   ∀ r .@@ -115,9 +125,9 @@       pid <- processId process       tryStop "signal failed" (Signal.signalProcess sig pid)     SystemProcess.ReadStdout ->-      tryStop "stdout failed" (hGetSome (getStdout process) 4096)+      checkEof =<< tryStop "stdout failed" (hGetSome (getStdout process) 4096)     SystemProcess.ReadStderr ->-      tryStop "stderr failed" (hGetSome (getStderr process) 4096)+      checkEof =<< tryStop "stderr failed" (hGetSome (getStderr process) 4096)     SystemProcess.WriteStdin msg ->       tryStop "stdin failed" (hPut (getStdin process) msg)     SystemProcess.Wait ->

@@ -5,21 +5,18 @@ -- see: https://github.com/sol/hpack  name:           polysemy-process-version:        0.8.0.1-synopsis:       Polysemy Effects for System Processes-description:    See <https://hackage.haskell.org/package/polysemy-process/docs/Polysemy-Process.html>+version:        0.9.0.0+synopsis:       Polysemy effects for system processes+description:    See https://hackage.haskell.org/package/polysemy-process/docs/Polysemy-Process.html category:       Concurrency homepage:       https://github.com/tek/polysemy-conc#readme bug-reports:    https://github.com/tek/polysemy-conc/issues author:         Torsten Schmits-maintainer:     haskell@tryp.io-copyright:      2021 Torsten Schmits+maintainer:     hackage@tryp.io+copyright:      2022 Torsten Schmits license:        BSD-2-Clause-Patent license-file:   LICENSE build-type:     Simple-extra-source-files:-    readme.md-    changelog.md  source-repository head   type: git@@ -31,15 +28,18 @@       Polysemy.Process.Data.ProcessError       Polysemy.Process.Data.ProcessKill       Polysemy.Process.Data.ProcessOptions+      Polysemy.Process.Data.ProcessOutputParseResult       Polysemy.Process.Data.PtyError       Polysemy.Process.Data.PtyResources       Polysemy.Process.Data.SystemProcessError       Polysemy.Process.Effect.Process+      Polysemy.Process.Effect.ProcessInput       Polysemy.Process.Effect.ProcessOutput       Polysemy.Process.Effect.Pty       Polysemy.Process.Effect.SystemProcess       Polysemy.Process.Executable       Polysemy.Process.Interpreter.Process+      Polysemy.Process.Interpreter.ProcessInput       Polysemy.Process.Interpreter.ProcessOutput       Polysemy.Process.Interpreter.ProcessStdio       Polysemy.Process.Interpreter.Pty@@ -63,6 +63,7 @@       DeriveFoldable       DeriveFunctor       DeriveGeneric+      DeriveLift       DeriveTraversable       DerivingStrategies       DerivingVia@@ -95,6 +96,7 @@       RankNTypes       RecordWildCards       RecursiveDo+      RoleAnnotations       ScopedTypeVariables       StandaloneDeriving       TemplateHaskell@@ -107,16 +109,16 @@       UndecidableInstances       UnicodeSyntax       ViewPatterns-  ghc-options: -Wall -Wredundant-constraints -Wincomplete-uni-patterns -Wmissing-deriving-strategies -Widentities+  ghc-options: -Wall -Wredundant-constraints -Wincomplete-uni-patterns -Wmissing-deriving-strategies -Widentities -Wunused-packages   build-depends:-      base ==4.*-    , incipit-core >=0.2+      base >=4.12 && <5+    , incipit-core >=0.3     , path >=0.7     , path-io >=1.6.2     , polysemy >=1.6-    , polysemy-conc-    , polysemy-resume >=0.3-    , polysemy-time >=0.4+    , polysemy-conc >=0.9+    , polysemy-resume >=0.5+    , polysemy-time >=0.5     , posix-pty >=0.2     , process     , stm-chans >=2@@ -124,8 +126,8 @@     , unix   mixins:       base hiding (Prelude)-  if impl(ghc >= 8.10)-    ghc-options: -Wunused-packages+    , incipit-core (IncipitCore as Prelude)+    , incipit-core hiding (IncipitCore)   default-language: Haskell2010  test-suite polysemy-process-unit@@ -149,6 +151,7 @@       DeriveFoldable       DeriveFunctor       DeriveGeneric+      DeriveLift       DeriveTraversable       DerivingStrategies       DerivingVia@@ -181,6 +184,7 @@       RankNTypes       RecordWildCards       RecursiveDo+      RoleAnnotations       ScopedTypeVariables       StandaloneDeriving       TemplateHaskell@@ -193,22 +197,22 @@       UndecidableInstances       UnicodeSyntax       ViewPatterns-  ghc-options: -Wall -Wredundant-constraints -Wincomplete-uni-patterns -Wmissing-deriving-strategies -Widentities -threaded -rtsopts -with-rtsopts=-N+  ghc-options: -Wall -Wredundant-constraints -Wincomplete-uni-patterns -Wmissing-deriving-strategies -Widentities -Wunused-packages -threaded -rtsopts -with-rtsopts=-N   build-depends:-      base ==4.*-    , incipit-core+      base >=4.12 && <5+    , incipit-core >=0.3     , polysemy     , polysemy-conc     , polysemy-plugin     , polysemy-process     , polysemy-resume-    , polysemy-test+    , polysemy-test >=0.6     , polysemy-time     , tasty     , tasty-expected-failure     , typed-process   mixins:       base hiding (Prelude)-  if impl(ghc >= 8.10)-    ghc-options: -Wunused-packages+    , incipit-core (IncipitCore as Prelude)+    , incipit-core hiding (IncipitCore)   default-language: Haskell2010

@@ -1,8 +0,0 @@-# About--This library provides [Polysemy] effects for system processes and pseudo terminals.--Please visit [Hackage] for documentation.--[Polysemy]: https://hackage.haskell.org/package/polysemy-[Hackage]: https://hackage.haskell.org/package/polysemy-process/docs/Polysemy-Process.html

@@ -2,6 +2,7 @@  module Polysemy.Process.Test.ProcessTest where +import qualified Data.ByteString as ByteString import qualified Polysemy.Conc as Conc import Polysemy.Conc.Effect.Scoped (Scoped) import Polysemy.Conc.Interpreter.Race (interpretRace)@@ -19,7 +20,9 @@ import Polysemy.Process.Data.ProcessOptions (ProcessOptions (kill)) import qualified Polysemy.Process.Effect.Process as Process import Polysemy.Process.Effect.Process (withProcess)+import Polysemy.Process.Interpreter.ProcessOutput (parseMany) import Polysemy.Process.Interpreter.ProcessStdio (interpretProcessByteStringNative, interpretProcessTextLinesNative)+import Polysemy.Process.Data.ProcessOutputParseResult (ProcessOutputParseResult(Done, Partial))  config :: ProcessConfig () () () config =@@ -29,18 +32,18 @@ messageLines =   replicate 4 "line" -message :: ByteString+message :: Text message =-  encodeUtf8 (unlines messageLines)+  unlines messageLines  test_process :: UnitTest test_process =   runTestAuto $ interpretRace $ asyncToIOFinal $ interpretProcessByteStringNative def config do     response <- resumeHoistError @ProcessError @(Scoped _ _) show do       withProcess do-        Process.send message+        Process.send (encodeUtf8 message)         Race.timeout_ (throw "timed out") (Seconds 5) Process.recv-    message === response+    message === decodeUtf8 response  test_processLines :: UnitTest test_processLines =@@ -51,8 +54,8 @@         Race.timeout_ (throw "timed out") (Seconds 5) (replicateM 4 Process.recv)     messageLines === response -test_processKill :: UnitTest-test_processKill =+test_processKillNever :: UnitTest+test_processKillNever =   runTestAuto $ interpretRace $ asyncToIOFinal $ interpretProcessTextLinesNative def { kill = KillNever } config do     result <- resumeHoistError @ProcessError @(Scoped _ _) show do       Conc.timeout unit (MilliSeconds 100) do@@ -63,10 +66,32 @@     -- the process and makes the right side terminate regularly.     assertLeft () result +test_processIncremental :: UnitTest+test_processIncremental =+  runTestAuto do+    (Nothing, ([Right "aa", Right "bb"], "")) === first void (parseMany parse Nothing "aabb")+    let (c1, r1) = parseMany parse Nothing "aabbc"+    ([Right "aa", Right "bb"], "") === r1+    case ($ "c") <$> c1 of+      Just (Done a "") -> "cc" === a+      a -> fail ("not Done: " <> show a)+    let (c2, r2) = parseMany parse Nothing "a"+    ([], "") === r2+    case ($ "a") <$> c2 of+      Just (Done a "") -> "aa" === a+      a -> fail ("not Done: " <> show a)+    (Nothing, ([Right "aa"], "")) === first void (parseMany parse c2 "a")+  where+    parse b+      | ByteString.length b == 1 =+        Partial (parse . (b <>))+      | otherwise =+        Done (ByteString.take 2 b) (ByteString.drop 2 b)+ test_processAll :: TestTree test_processAll =   testGroup "process" [-    unitTest "process" test_process,-    unitTest "process lines" test_processLines,-    ignoreTest (unitTest "process kill" test_processKill)+    unitTest "read raw chunks" test_process,+    unitTest "read lines" test_processLines,+    ignoreTest (unitTest "don't kill the process at the end of the scope" test_processKillNever)   ]