reflex-process 0.3.0.0 → 0.3.1.0
raw patch · 5 files changed
+53/−38 lines, 5 filesdep ~basedep ~reflex
Dependency ranges changed: base, reflex
Files
- ChangeLog.md +5/−1
- README.lhs +6/−5
- README.md +6/−5
- reflex-process.cabal +7/−4
- src/Reflex/Process.hs +29/−23
ChangeLog.md view
@@ -1,11 +1,15 @@ # Revision history for reflex-process +## 0.3.1.0++* Support reflex 0.8+* Fix a handle leak (#23) and a thread leak (#24)+ ## 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`.
README.lhs view
@@ -1,12 +1,12 @@ 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)+[](https://haskell.org) [](https://hackage.haskell.org/package/reflex-process) [](https://matrix.hackage.haskell.org/#/package/reflex-process) [](https://travis-ci.org/reflex-frp/reflex-process) [](https://github.com/reflex-frp/reflex-process/blob/master/LICENSE) -Functional-reactive shell commands-----------------------------------+Functional-reactive system processes+------------------------------------ -This library provides a functional-reactive interface for running shell commands from [reflex](https://github.com/reflex-frp/reflex).+Run and interact with system processes from within a [Reflex FRP](https://reflex-frp.org/) application. Example -------@@ -46,7 +46,8 @@ --------------------- You can get `ghcid` running for working on the code with the command:-```++```bash nix-shell -E '((import ./reflex-platform {}).ghc.callCabal2nix "reflex-process" ./. {}).env' --run ghcid ```
README.md view
@@ -1,12 +1,12 @@ 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)+[](https://haskell.org) [](https://hackage.haskell.org/package/reflex-process) [](https://matrix.hackage.haskell.org/#/package/reflex-process) [](https://travis-ci.org/reflex-frp/reflex-process) [](https://github.com/reflex-frp/reflex-process/blob/master/LICENSE) -Functional-reactive shell commands-----------------------------------+Functional-reactive system processes+------------------------------------ -This library provides a functional-reactive interface for running shell commands from [reflex](https://github.com/reflex-frp/reflex).+Run and interact with system processes from within a [Reflex FRP](https://reflex-frp.org/) application. Example -------@@ -46,7 +46,8 @@ --------------------- You can get `ghcid` running for working on the code with the command:-```++```bash nix-shell -E '((import ./reflex-platform {}).ghc.callCabal2nix "reflex-process" ./. {}).env' --run ghcid ```
reflex-process.cabal view
@@ -1,8 +1,11 @@ cabal-version: >=1.10 name: reflex-process-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+version: 0.3.1.0+synopsis: Reflex FRP interface for running system processes+description:+ Run and interact with system processes from within a Reflex FRP application.+ .+ <https://reflex-frp.org> bug-reports: https://github.com/reflex-frp/reflex-process/issues license: BSD3 license-file: LICENSE@@ -22,7 +25,7 @@ , bytestring >=0.10 && < 0.11 , data-default >= 0.2 && < 0.8 , process >= 1.6.4 && < 1.7- , reflex >= 0.7.1 && < 0.8+ , reflex >= 0.7.1 && < 0.9 , unix >= 2.7 && < 2.8 hs-source-dirs: src default-language: Haskell2010
src/Reflex/Process.hs view
@@ -4,6 +4,7 @@ -} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-}+{-# LANGUAGE RecursiveDo #-} {-# LANGUAGE ScopedTypeVariables #-} module Reflex.Process@@ -19,11 +20,12 @@ , createRedirectedProcess ) where -import Control.Concurrent.Async (Async, async, waitBoth)+import Control.Concurrent.Async (Async, async, race_, 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 Control.Monad.Fix (MonadFix) import Data.ByteString (ByteString) import qualified Data.ByteString as BS import Data.Default (Default, def)@@ -40,13 +42,11 @@ data SendPipe i = SendPipe_Message i- -- ^ A message that's sent to the underlying process+ -- ^ A message that's sent to the underlying process. This does NOT include a trailing newline when sending your message. | SendPipe_EOF- -- ^ Send an EOF to the underlying process+ -- ^ Send an EOF to the underlying process. Once this is sent no further messages will be processed. | 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+ -- ^ Send the last message along with an EOF. Once this is sent no further messages will be processed. -- | The inputs to a process data ProcessConfig t i = ProcessConfig@@ -81,7 +81,7 @@ } -- | Create a process feeding it input using an 'Event' and exposing its output--- 'Event's representing the process exit code, stdout, and stderr.+-- 'Event's representing the process exit code, @stdout@, and @stderr@. -- -- The @stdout@ and @stderr@ 'Handle's are line-buffered. --@@ -92,7 +92,7 @@ -- provided 'CreateProcess' are replaced with new pipes and all output is redirected -- to those pipes. createProcess- :: (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m))+ :: (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m), MonadFix m) => P.CreateProcess -- ^ Specification of process to create -> ProcessConfig t (SendPipe ByteString) -- ^ Reflex-level configuration for the process -> m (Process t ByteString ByteString)@@ -106,11 +106,12 @@ -- -- The @stdout@ and @stderr@ 'Handle's are line-buffered. ----- For example, you may use 'Chan' for an unbounded buffer (like 'createProcess' does) like this:+-- 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'.+-- 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.@@ -124,31 +125,32 @@ -- provided 'CreateProcess' are replaced with new pipes and all output is redirected -- to those pipes. createProcessBufferingInput- :: (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m))+ :: (MonadIO m, TriggerEvent t m, PerformEvent t m, MonadIO (Performable m), MonadFix m) => 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.+ -- This will run in a separate thread and 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)-createProcessBufferingInput readBuffer writeBuffer = unsafeCreateProcessWithHandles input output output+createProcessBufferingInput readBuffer writeBuffer spec config = do+ rec p <- unsafeCreateProcessWithHandles (input $ _process_handle p) output output spec config+ pure p where- input :: Handle -> IO (SendPipe ByteString -> IO ())- input h = do+ input :: ProcessHandle -> Handle -> IO (SendPipe ByteString -> IO ())+ input ph h = do H.hSetBuffering h H.NoBuffering- void $ liftIO $ async $ fix $ \loop -> do+ void $ liftIO $ async $ race_ (waitForProcess ph) $ 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_Message m -> BS.hPutStr h m *> loop+ 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@@ -172,9 +174,11 @@ 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+ -- ^ Builder for the standard input handler.+ -- This is provided so you can link any new threads to this one to avoid leaking threads.+ -- 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.+ -- This function 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@@ -193,8 +197,6 @@ (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@@ -220,6 +222,10 @@ waited <- waitForProcess ph _ <- waitBoth outThread errThread ecTrigger waited -- Output events should never fire after process completion++ writeInput :: i -> IO () <- liftIO $ mkWriteStdInput hIn+ performEvent_ $ liftIO . writeInput <$> input+ return $ Process { _process_exit = ecOut , _process_stdout = out