reflex-process 0.2.1.0 → 0.3.0.0
raw patch · 6 files changed
+352/−101 lines, 6 filesdep +asyncdep +dependent-sumdep +hspecdep ~basedep ~bytestringdep ~containersnew-uploader
Dependencies added: async, dependent-sum, hspec, mtl, primitive, ref-tf
Dependency ranges changed: base, bytestring, containers, data-default, process, reflex, unix, vty
Files
- ChangeLog.md +12/−0
- README.lhs +10/−1
- README.md +10/−1
- reflex-process.cabal +26/−4
- src/Reflex/Process.hs +182/−95
- test/Main.hs +112/−0
ChangeLog.md view
@@ -1,5 +1,17 @@ # Revision history for reflex-process +## 0.3.0.0++* ([#15](https://github.com/reflex-frp/reflex-process/pull/15), [#13](https://github.com/reflex-frp/reflex-process/pull/13)) **(Breaking change)** Introduce `SendPipe` type for encoding when an input stream should send EOF. Change `createProcess` to take a `ProcessConfig t (SendPipe ByteString)` so that sending EOF is possible.+ * **IMPORTANT**: For `createProcess` messages to `stdin` must now be wrapped in `SendPipe_Message` *and* have a `"\n"` manually appended to regain the old behavior. Previously `createProcess` implicitly added a `"\n"` to all messages sent to the process. This has been removed and *you must now manually add any necessary new lines to your messages.* This change allows `createProcess` to work with processes in a encoding-agnostic way on `stdin`.+* ([#17](https://github.com/reflex-frp/reflex-process/pull/17)) Deprecate `createRedirectedProcess` in favor of a new, scarier name: `unsafeCreateProcessWithHandles`. This was done to communicate that it does not enforce necessary guarantees for the process to be handled correctly.+* ([#15](https://github.com/reflex-frp/reflex-process/pull/15), [#13](https://github.com/reflex-frp/reflex-process/pull/13)) **(Breaking change)** Sending a message with `SendPipe_Message` `createProcess`+* ([#11](https://github.com/reflex-frp/reflex-process/pull/11)) Add `createProcessBufferingInput` for buffering input to processes and change `createProcess` to use an unbounded buffer instead of blocking the FRP network when the process blocks on its input handle.+* ([#11](https://github.com/reflex-frp/reflex-process/pull/11), [#14](https://github.com/reflex-frp/reflex-process/pull/14)) `ProcessConfig` now includes a `_processConfig_createProcess` field for customizing how the process is created.+* ([#13](https://github.com/reflex-frp/reflex-process/pull/13)) Fix race condition between process completion `Event`s and process `stdout`/`stderr` `Event`s. Process completion is now always the very last `Event` to fire for a given `Process`.+* ([#17](https://github.com/reflex-frp/reflex-process/pull/17)) Add `defProcessConfig` to avoid forcing users to depend on `data-default`.++ ## 0.2.1.0 * `createProcess`: Ensure that handle is open before attempting to check whether it is readable
README.lhs view
@@ -1,4 +1,4 @@-reflex-process +reflex-process ============== [](https://hackage.haskell.org/package/reflex-process) [](https://matrix.hackage.haskell.org/#/package/reflex-process) [](https://travis-ci.org/reflex-frp/reflex-process)@@ -41,3 +41,12 @@ > stretch $ text $ T.decodeUtf8 <$> current stdout > pure $ () <$ exit ```++Developer environment+---------------------++You can get `ghcid` running for working on the code with the command:+```+nix-shell -E '((import ./reflex-platform {}).ghc.callCabal2nix "reflex-process" ./. {}).env' --run ghcid+```+
README.md view
@@ -1,4 +1,4 @@-reflex-process +reflex-process ============== [](https://hackage.haskell.org/package/reflex-process) [](https://matrix.hackage.haskell.org/#/package/reflex-process) [](https://travis-ci.org/reflex-frp/reflex-process)@@ -41,3 +41,12 @@ > stretch $ text $ T.decodeUtf8 <$> current stdout > pure $ () <$ exit ```++Developer environment+---------------------++You can get `ghcid` running for working on the code with the command:+```+nix-shell -E '((import ./reflex-platform {}).ghc.callCabal2nix "reflex-process" ./. {}).env' --run ghcid+```+
reflex-process.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: reflex-process-version: 0.2.1.0+version: 0.3.0.0 synopsis: reflex-frp interface for running shell commands description: Run shell commands from within reflex applications and interact with them over a functional-reactive interface bug-reports: https://github.com/reflex-frp/reflex-process/issues@@ -18,10 +18,11 @@ library exposed-modules: Reflex.Process build-depends: base >=4.12 && <4.14+ , async >= 2 && < 3 , bytestring >=0.10 && < 0.11- , data-default >= 0.7.1 && < 0.8+ , data-default >= 0.2 && < 0.8 , process >= 1.6.4 && < 1.7- , reflex >= 0.6.2.4 && < 0.7+ , reflex >= 0.7.1 && < 0.8 , unix >= 2.7 && < 2.8 hs-source-dirs: src default-language: Haskell2010@@ -38,5 +39,26 @@ , reflex-process , reflex-vty >= 0.1.2.1 && < 0.2 , text >= 1.2.3 && < 1.3- , vty >= 5.21 && < 5.26+ , vty default-language: Haskell2010++test-suite tests+ default-language: Haskell2010+ type: exitcode-stdio-1.0+ main-is: Main.hs+ hs-source-dirs: test+ ghc-options: -O2 -Wall -rtsopts -threaded+ build-depends:+ async,+ base,+ bytestring,+ containers,+ dependent-sum,+ hspec,+ mtl,+ primitive,+ process,+ ref-tf,+ reflex,+ reflex-process,+ unix
src/Reflex/Process.hs view
@@ -1,44 +1,71 @@ {-| Module: Reflex.Process-Description: Run interactive shell commands in reflex+Description: Run processes and interact with them in reflex -} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+ module Reflex.Process ( createProcess- , createRedirectedProcess+ , createProcessBufferingInput+ , defProcessConfig+ , unsafeCreateProcessWithHandles , Process(..) , ProcessConfig(..)+ , SendPipe (..)++ -- Deprecations+ , createRedirectedProcess ) where -import Control.Concurrent (forkIO, killThread)-import Control.Exception (mask_)+import Control.Concurrent.Async (Async, async, waitBoth)+import Control.Concurrent.Chan (newChan, readChan, writeChan)+import Control.Exception (finally) import Control.Monad (void, when) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.ByteString (ByteString) import qualified Data.ByteString as BS-import qualified Data.ByteString.Char8 as Char8-import Data.Default-import qualified GHC.IO.Handle as H+import Data.Default (Default, def)+import Data.Function (fix)+import Data.Traversable (for) import GHC.IO.Handle (Handle)+import qualified GHC.IO.Handle as H import System.Exit (ExitCode) import qualified System.Posix.Signals as P-import qualified System.Process as P import System.Process hiding (createProcess)+import qualified System.Process as P import Reflex +data SendPipe i+ = SendPipe_Message i+ -- ^ A message that's sent to the underlying process+ | SendPipe_EOF+ -- ^ Send an EOF to the underlying process+ | SendPipe_LastMessage i+ -- ^ Send the last message (an EOF will be added). This option is offered for+ -- convenience, because it has the same effect of sending a Message and then+ -- the EOF signal+ -- | The inputs to a process data ProcessConfig t i = ProcessConfig { _processConfig_stdin :: Event t i- -- ^ "stdin" input to be fed to the process+ -- ^ @stdin@ input to be fed to the process , _processConfig_signal :: Event t P.Signal -- ^ Signals to send to the process }- instance Reflex t => Default (ProcessConfig t i) where- def = ProcessConfig never never+ -- | An alias for 'defProcessConfig'.+ def = defProcessConfig +-- | A default 'ProcessConfig' where @stdin@ and signals are never sent.+--+-- You can also use 'Data.Default.def'.+defProcessConfig :: Reflex t => ProcessConfig t i+defProcessConfig = ProcessConfig never never++ -- | The output of a process data Process t o e = Process { _process_handle :: P.ProcessHandle@@ -47,107 +74,167 @@ , _process_stderr :: Event t e -- ^ Fires whenever there's some new stderr output. See note on '_process_stdout'. , _process_exit :: Event t ExitCode+ -- ^ Fires when the process is over and no @stdout@ or @stderr@ data is left.+ -- Once this fires, no other 'Event's for the process will fire again. , _process_signal :: Event t P.Signal -- ^ Fires when a signal has actually been sent to the process (via '_processConfig_signal'). } --- | Runs a process and uses the given input and output handler functions to--- interact with the process via the standard streams. Used to implement--- 'createProcess'.+-- | Create a process feeding it input using an 'Event' and exposing its output+-- 'Event's representing the process exit code, stdout, and stderr. ----- NB: The 'std_in', 'std_out', and 'std_err' parameters of the+-- The @stdout@ and @stderr@ 'Handle's are line-buffered.+--+-- N.B. The process input is buffered with an unbounded channel! For more control of this,+-- use 'createProcessBufferingInput' directly.+--+-- N.B.: The 'std_in', 'std_out', and 'std_err' parameters of the -- provided 'CreateProcess' are replaced with new pipes and all output is redirected -- to those pipes.-createRedirectedProcess+createProcess :: (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m))- => (Handle -> IO (i -> IO ()))- -- ^ Builder for the standard input handler- -> (Handle -> (o -> IO ()) -> IO (IO ()))- -- ^ Builder for the standard output handler- -> (Handle -> (e -> IO ()) -> IO (IO ()))- -- ^ Builder for the standard error handler- -> CreateProcess- -> ProcessConfig t i- -> m (Process t o e)-createRedirectedProcess mkWriteStdInput mkReadStdOutput mkReadStdError p (ProcessConfig input signal) = do- let redirectedProc = p- { std_in = CreatePipe- , std_out = CreatePipe- , std_err = CreatePipe- }- po@(mi, mout, merr, ph) <- liftIO $ P.createProcess redirectedProc- case (mi, mout, merr) of- (Just hIn, Just hOut, Just hErr) -> do- writeInput <- liftIO $ mkWriteStdInput hIn- performEvent_ $ liftIO . writeInput <$> input- sigOut <- performEvent $ ffor signal $ \sig -> liftIO $ do- mpid <- P.getPid ph- case mpid of- Nothing -> return Nothing- Just pid -> do- P.signalProcess sig pid >> return (Just sig)- let output h = do- (e, trigger) <- newTriggerEvent- reader <- liftIO $ mkReadStdOutput h trigger- t <- liftIO $ forkIO reader- return (e, t)-- let err_output h = do- (e, trigger) <- newTriggerEvent- reader <- liftIO $ mkReadStdError h trigger- t <- liftIO $ forkIO reader- return (e, t)- (out, outThread) <- output hOut- (err, errThread) <- err_output hErr- (ecOut, ecTrigger) <- newTriggerEvent- void $ liftIO $ forkIO $ waitForProcess ph >>= \ec -> mask_ $ do- ecTrigger ec- P.cleanupProcess po- killThread outThread- killThread errThread- return $ Process- { _process_exit = ecOut- , _process_stdout = out- , _process_stderr = err- , _process_signal = fmapMaybe id sigOut- , _process_handle = ph- }- _ -> error "Reflex.Vty.Process.createRedirectedProcess: Created pipes were not returned by System.Process.createProcess."+ => P.CreateProcess -- ^ Specification of process to create+ -> ProcessConfig t (SendPipe ByteString) -- ^ Reflex-level configuration for the process+ -> m (Process t ByteString ByteString)+createProcess p procConfig = do+ channel <- liftIO newChan+ createProcessBufferingInput (readChan channel) (writeChan channel) p procConfig --- | Run a shell process, feeding it input using an 'Event' and exposing its output--- 'Event's representing the process exit code, stdout and stderr.+-- | Create a process feeding it input using an 'Event' and exposing its output with 'Event's+-- for its exit code, @stdout@, and @stderr@. The input is fed via a buffer represented by a+-- reading action and a writing action. ----- The input 'Handle' is not buffered and the output 'Handle's are line-buffered.+-- The @stdout@ and @stderr@ 'Handle's are line-buffered. ----- NB: The 'std_in', 'std_out', and 'std_err' parameters of the+-- For example, you may use 'Chan' for an unbounded buffer (like 'createProcess' does) like this:+-- > channel <- liftIO newChan+-- > createProcessBufferingInput (readChan channel) (writeChan channel) myConfig+--+-- Similarly you could use 'TChan'.+--+-- Bounded buffers may cause the Reflex network to block when you trigger an 'Event' that would+-- cause more data to be sent to a process whose @stdin@ is blocked.+--+-- If an unbounded channel would lead to too much memory usage you will want to consider+-- * speeding up the consuming process.+-- * buffering with the file system or another persistent storage to reduce memory usage.+-- * if your usa case allows, dropping 'Event's or messages that aren't important.+--+-- N.B.: The 'std_in', 'std_out', and 'std_err' parameters of the -- provided 'CreateProcess' are replaced with new pipes and all output is redirected -- to those pipes.-createProcess+createProcessBufferingInput :: (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m))- => CreateProcess- -> ProcessConfig t ByteString+ => IO (SendPipe ByteString)+ -- ^ An action that reads a value from the input stream buffer.+ -- This must block when the buffer is empty or not ready.+ -> (SendPipe ByteString -> IO ())+ -- ^ An action that writes a value to the input stream buffer.+ -> P.CreateProcess -- ^ Specification of process to create+ -> ProcessConfig t (SendPipe ByteString) -- ^ Reflex-level configuration for the process -> m (Process t ByteString ByteString)-createProcess = createRedirectedProcess input output output+createProcessBufferingInput readBuffer writeBuffer = unsafeCreateProcessWithHandles input output output where+ input :: Handle -> IO (SendPipe ByteString -> IO ()) input h = do H.hSetBuffering h H.NoBuffering- let go b = do- open <- H.hIsOpen h- when open $ do- writable <- H.hIsWritable h- when writable $ Char8.hPutStrLn h b- return go+ void $ liftIO $ async $ fix $ \loop -> do+ newMessage <- readBuffer+ open <- H.hIsOpen h+ when open $ do+ writable <- H.hIsWritable h+ when writable $ do+ case newMessage of+ SendPipe_Message m -> BS.hPutStr h m+ SendPipe_LastMessage m -> BS.hPutStr h m >> H.hClose h+ SendPipe_EOF -> H.hClose h+ loop+ return writeBuffer output h trigger = do H.hSetBuffering h H.LineBuffering- let go = do- open <- H.hIsOpen h- when open $ do- readable <- H.hIsReadable h- when readable $ do- out <- BS.hGetSome h 32768- if BS.null out- then return ()- else do- void $ trigger out- go- return go+ pure $ fix $ \go -> do+ open <- H.hIsOpen h+ when open $ do+ readable <- H.hIsReadable h+ when readable $ do+ out <- BS.hGetSome h 32768+ if BS.null out+ then H.hClose h+ else void (trigger out) *> go++-- | Runs a process and uses the given input and output handler functions to+-- interact with the process via the standard streams. Used to implement+-- 'createProcess'.+--+-- N.B.: The 'std_in', 'std_out', and 'std_err' parameters of the+-- provided 'CreateProcess' are replaced with new pipes and all output is redirected+-- to those pipes.+unsafeCreateProcessWithHandles+ :: forall t m i o e. (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m))+ => (Handle -> IO (i -> IO ()))+ -- ^ Builder for the standard input handler. The 'Handle' is the write end of the process' @stdin@ and+ -- the resulting @i -> IO ()@ is a function that writes each input 'Event t i' to into 'Handle'.+ -- This functios must not block or the entire Reflex network will block.+ -> (Handle -> (o -> IO ()) -> IO (IO ()))+ -- ^ Builder for the standard output handler. The 'Handle' is the read end of the process' @stdout@ and+ -- the @o -> IO ()@ is a function that will trigger the output @Event t o@ when called. The resulting+ -- @IO ()@ will be run in a separate thread and must block until there is no more data in the 'Handle' to+ -- process.+ -> (Handle -> (e -> IO ()) -> IO (IO ()))+ -- ^ Builder for the standard error handler. The 'Handle' is the read end of the process' @stderr@ and+ -- the @e -> IO ()@ is a function that will trigger the output @Event t e@ when called. The resulting+ -- @IO ()@ will be run in a separate thread and must block until there is no more data in the 'Handle' to+ -- process.+ -> P.CreateProcess -- ^ Specification of process to create+ -> ProcessConfig t i -- ^ Reflex-level configuration for the process+ -> m (Process t o e)+unsafeCreateProcessWithHandles mkWriteStdInput mkReadStdOutput mkReadStdError p (ProcessConfig input signal) = do+ po <- liftIO $ P.createProcess p { std_in = P.CreatePipe, std_out = P.CreatePipe, std_err = P.CreatePipe }+ (hIn, hOut, hErr, ph) <- case po of+ (Just hIn, Just hOut, Just hErr, ph) -> pure (hIn, hOut, hErr, ph)+ _ -> error "Reflex.Process.unsafeCreateProcessWithHandles: Created pipes were not returned by System.Process.createProcess."+ writeInput :: i -> IO () <- liftIO $ mkWriteStdInput hIn+ performEvent_ $ liftIO . writeInput <$> input+ sigOut :: Event t (Maybe P.Signal) <- performEvent $ ffor signal $ \sig -> liftIO $ do+ mpid <- P.getPid ph+ for mpid $ \pid -> sig <$ P.signalProcess sig pid+ let+ output :: Handle -> m (Event t o, Async ())+ output h = do+ (e, trigger) <- newTriggerEvent+ reader <- liftIO $ mkReadStdOutput h trigger+ t <- liftIO $ async reader+ return (e, t)++ errOutput :: Handle -> m (Event t e, Async ())+ errOutput h = do+ (e, trigger) <- newTriggerEvent+ reader <- liftIO $ mkReadStdError h trigger+ t <- liftIO $ async reader+ return (e, t)++ (out, outThread) <- output hOut+ (err, errThread) <- errOutput hErr+ (ecOut, ecTrigger) <- newTriggerEvent+ void $ liftIO $ async $ flip finally (P.cleanupProcess (Just hIn, Just hOut, Just hErr, ph)) $ do+ waited <- waitForProcess ph+ _ <- waitBoth outThread errThread+ ecTrigger waited -- Output events should never fire after process completion+ return $ Process+ { _process_exit = ecOut+ , _process_stdout = out+ , _process_stderr = err+ , _process_signal = fmapMaybe id sigOut+ , _process_handle = ph+ }++{-# DEPRECATED createRedirectedProcess "Use unsafeCreateProcessWithHandles instead." #-}+createRedirectedProcess+ :: forall t m i o e. (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m))+ => (Handle -> IO (i -> IO ()))+ -> (Handle -> (o -> IO ()) -> IO (IO ()))+ -> (Handle -> (e -> IO ()) -> IO (IO ()))+ -> P.CreateProcess+ -> ProcessConfig t i+ -> m (Process t o e)+createRedirectedProcess = unsafeCreateProcessWithHandles
+ test/Main.hs view
@@ -0,0 +1,112 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecursiveDo #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}++module Main where++import Control.Concurrent (MVar, newEmptyMVar, takeMVar, threadDelay, tryPutMVar)+import Control.Concurrent.Async (race)+import Control.Exception (finally)+import Control.Monad (guard, void)+import Control.Monad.IO.Class (liftIO)+import Data.Bifunctor (first, second)+import Data.ByteString (ByteString)+import Data.IORef (newIORef, writeIORef, readIORef)+import Data.Foldable (traverse_)+import Reflex+import System.Timeout (timeout)+import qualified Data.ByteString.Char8 as BS+import qualified System.Process as P++import Reflex.Host.Headless+import Reflex.Process+import Test.Hspec++main :: IO ()+main = hspec $ do+ describe "reflex-process" $ do+ it "isn't blocked by a downstream non-blocking process" $ do+ timeoutWrapperAsync (checkFRPBlocking $ P.proc "cat" []) `shouldReturn` Right (Just Exit)+ it "isn't blocked by a downstream blocking process" $ do+ timeoutWrapperAsync (checkFRPBlocking $ P.proc "sleep" ["infinity"]) `shouldReturn` Right (Just Exit)+ it "sends messages on stdin and receives messages on stdout and stderr" $ runHeadlessApp $ do+ let+ -- Produces an event when the given message is seen on both stdout and stderr of the given process events+ getSawMessage procOut msg = do+ let filterMsg = mapMaybe (guard . (== msg))+ seen <- foldDyn ($) (False, False) $+ mergeWith (.)+ [ first (const True) <$ filterMsg (_process_stdout procOut)+ , second (const True) <$ filterMsg (_process_stderr procOut)+ ]+ pure $ mapMaybe (guard . (== (True, True))) $ updated seen++ rec+ procOut <- createProcess (P.proc "tee" ["/dev/stderr"]) $ ProcessConfig send never+ aWasSeen <- getSawMessage procOut "a\n"+ bWasSeen <- getSawMessage procOut "b\n"+ pb <- getPostBuild+ let+ send = leftmost+ [ SendPipe_Message "a\n" <$ pb+ , SendPipe_Message "b\n" <$ aWasSeen+ , SendPipe_LastMessage "c\n" <$ bWasSeen+ ]++ getSawMessage procOut "c\n"++ it "sends signals" $ runHeadlessApp $ void . _process_exit <$> sendSignalTest+ it "fires event when signal is sent" $ runHeadlessApp $ void . _process_signal <$> sendSignalTest++ where+ sendSignalTest :: MonadHeadlessApp t m => m (Process t ByteString ByteString)+ sendSignalTest = do+ (signal, signalTrigger) <- newTriggerEvent+ procOut <- createProcess (P.proc "sleep" ["infinity"]) $ ProcessConfig never signal+ liftIO $ threadDelay 1000000 *> signalTrigger 15 -- SIGTERM+ pure procOut+++-- This datatype signals that the FRP network was able to exit on its own.+data Exit = Exit deriving (Show, Eq)++-- This creates the MVar with which the FRP network sends the exit signals, and+-- checks if a response from the FRP networks comes back in the allotted time.+timeoutWrapperAsync :: (MVar Exit -> IO ()) -> IO (Either () (Maybe Exit))+timeoutWrapperAsync wrapped = do+ exitCommMVar :: MVar Exit <- newEmptyMVar+ race (wrapped exitCommMVar) (timeout (3*1000000) (takeMVar exitCommMVar))++-- The frp network spawns, starts a timer, and tries to send some long input to+-- the created process. If the underlying process blocks and is able to block+-- the FRP network, the first tick of the timer will never happen, and, the Exit+-- signal will never be put in the MVar.+checkFRPBlocking :: P.CreateProcess -> MVar Exit -> IO ()+checkFRPBlocking downstreamProcess exitMVar = do+ spawnedProcess <- newIORef Nothing++ finally+ (runHeadlessApp $ do+ timer <- tickLossyFromPostBuildTime 1+ void $ performEvent $ liftIO (tryPutMVar exitMVar Exit) <$ timer++ (ev, evTrigger :: SendPipe ByteString -> IO ()) <- newTriggerEvent+ processOutput <- createProcess downstreamProcess $ ProcessConfig ev never+ liftIO $ writeIORef spawnedProcess (Just $ _process_handle processOutput)++ liftIO $ evTrigger $ SendPipe_Message $ veryLongByteString 'a'+ liftIO $ evTrigger $ SendPipe_Message $ veryLongByteString 'b'+ liftIO $ evTrigger $ SendPipe_LastMessage $ veryLongByteString 'c'++ void $ performEvent $ liftIO . BS.putStrLn <$> _process_stdout processOutput+ pure never+ )+ (readIORef spawnedProcess >>= traverse_ P.terminateProcess)++-- It's important to try this with long bytestrings to be sure that they're not+-- put in an operative system inter-process buffer.+veryLongByteString :: Char -> ByteString+veryLongByteString = BS.replicate 100000+--------------------------------------------------------------------------------