reflex-process 0.3.2.1 → 0.3.3.1
raw patch · 5 files changed
+134/−33 lines, 5 filesdep ~containersdep ~reflex-vty
Dependency ranges changed: containers, reflex-vty
Files
- ChangeLog.md +10/−2
- reflex-process.cabal +5/−3
- src/Reflex/Process.hs +2/−2
- src/Reflex/Process/Lines.hs +86/−0
- test/Main.hs +31/−26
ChangeLog.md view
@@ -1,8 +1,16 @@ # Revision history for reflex-process -## 0.3.2.1+## 0.3.3.1 -* Allow reflex-vty 0.5.*+* Support for reflex-0.9.3++## 0.3.3.0++* Add `Lines` for keeping track of accumulated output, including both terminated and unterminated lines++## 0.3.2.0-r1++* Allow reflex-vty 0.5 ## 0.3.2.0
reflex-process.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: reflex-process-version: 0.3.2.1+version: 0.3.3.1 synopsis: Reflex FRP interface for running system processes description: Run and interact with system processes from within a Reflex FRP application.@@ -25,9 +25,11 @@ library exposed-modules: Reflex.Process+ Reflex.Process.Lines build-depends: base >=4.12 && <4.19 , async >= 2 && < 3- , bytestring >=0.10 && < 0.12+ , bytestring >= 0.10 && < 0.12+ , containers >= 0.6 && < 0.7 , data-default >= 0.2 && < 0.8 , process >= 1.6.4 && < 1.7 , reflex >= 0.7.1 && < 1@@ -45,7 +47,7 @@ , process , reflex , reflex-process- , reflex-vty >= 0.2 && < 0.6+ , reflex-vty >= 0.2 && < 0.7 , text >= 1.2.3 && < 2.1 , vty default-language: Haskell2010
src/Reflex/Process.hs view
@@ -141,7 +141,7 @@ where input :: ProcessHandle -> Handle -> IO (SendPipe ByteString -> IO ()) input ph h = do- H.hSetBuffering h H.NoBuffering+ H.hIsOpen h >>= \open -> if open then H.hSetBuffering h H.LineBuffering else return () void $ liftIO $ async $ race_ (waitForProcess ph) $ fix $ \loop -> do newMessage <- readBuffer open <- H.hIsOpen h@@ -154,7 +154,7 @@ SendPipe_EOF -> H.hClose h return writeBuffer output h trigger = do- H.hSetBuffering h H.LineBuffering+ H.hIsOpen h >>= \open -> if open then H.hSetBuffering h H.LineBuffering else return () pure $ fix $ \go -> do open <- H.hIsOpen h when open $ do
+ src/Reflex/Process/Lines.hs view
@@ -0,0 +1,86 @@+{-# Language OverloadedStrings #-}+module Reflex.Process.Lines where++import Control.Monad.Fix (MonadFix)+import qualified Data.ByteString.Char8 as C8+import Data.ByteString.Char8 (ByteString)+import Data.Foldable (toList)+import Data.Maybe (fromMaybe)+import Data.Sequence (Seq)+import qualified Data.Sequence as Seq+import Reflex++-- * Output lines++-- | Accumulator for line-based output that keeps track of any dangling,+-- unterminated line+data Lines = Lines+ { _lines_terminated :: Seq C8.ByteString+ , _lines_unterminated :: Maybe C8.ByteString+ }+ deriving (Show, Eq, Ord, Read)++-- | Empty output+emptyLines :: Lines+emptyLines = Lines Seq.empty Nothing++-- | Add some raw output to a 'Lines'. This will chop the raw output up into lines.+addLines :: ByteString -> Lines -> Lines+addLines new (Lines t u) =+ let newLines' = Seq.fromList $ filter (not . C8.null) (C8.lines new)+ in+ case u of+ Nothing -> if "\n" `C8.isSuffixOf` new+ then Lines (t <> newLines') Nothing+ else case Seq.viewr newLines' of+ Seq.EmptyR -> Lines t Nothing+ (t' Seq.:> u') -> Lines (t <> t') (Just u')+ Just u' -> addLines (u' <> new) $ Lines t Nothing++-- | Convert a 'ByteString' into a 'Lines'+linesFromBS :: C8.ByteString -> Lines+linesFromBS = flip addLines mempty++instance Semigroup Lines where+ a <> b = addLines (unLines b) a++instance Monoid Lines where+ mempty = emptyLines++-- | Convert a 'Lines' back into a 'ByteString'+unLines :: Lines -> ByteString+unLines (Lines t u) =+ C8.unlines (toList t) <> fromMaybe "" u++-- | Convenience accessor for the last whole line received by a 'Lines'.+-- Ignores any unterminated line that may follow.+lastWholeLine :: Lines -> Maybe C8.ByteString+lastWholeLine (Lines t _) = case Seq.viewr t of+ Seq.EmptyR -> Nothing+ _ Seq.:> x -> Just x++-- | Split lines into two. The sequence that satisfies the predicate is+-- consumed and will not appear in either resulting 'Lines'.+splitLinesOn :: (ByteString -> Bool) -> Lines -> Maybe (Lines, Lines)+splitLinesOn test (Lines t u) = + let (before, after) = Seq.breakl test t+ in if Seq.null after then Nothing else Just (Lines before Nothing, Lines (Seq.drop 1 after) u)++-- | Given an event of raw bytes, fire an output event of *terminated* lines.+-- Unterminated lines are held until the line they belong to is completed or+-- until the flush event fires.+newLines+ :: (Reflex t, MonadHold t m, MonadFix m)+ => Event t ByteString+ -> Event t () -- ^ Event that flushes any remaining unterminated lines+ -> m (Event t Lines) -- ^ These will be complete lines except when the flush event fires, in which it may include unterminated lines+newLines e flush = do+ x <- foldDyn ($) (mempty, mempty) $ mergeWith (.)+ [ ffor e $ \new (_, old) ->+ let Lines t u = addLines new old+ in (Lines t Nothing, Lines mempty u)+ , ffor flush $ \_ (_, old) -> (old, emptyLines)+ ]+ pure $ fforMaybe (updated x) $ \(terminatedLines, _) -> if terminatedLines == mempty+ then Nothing+ else Just terminatedLines
test/Main.hs view
@@ -31,35 +31,41 @@ 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+ it "sends messages on stdin and receives messages on stdout and stderr" $ do+ () <- runHeadlessApp $ do let- send = leftmost- [ SendPipe_Message "a\n" <$ pb- , SendPipe_Message "b\n" <$ aWasSeen- , SendPipe_LastMessage "c\n" <$ bWasSeen- ]+ -- 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 - getSawMessage procOut "c\n"+ 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+ ] - it "sends signals" $ runHeadlessApp $ void . _process_exit <$> sendSignalTest- it "fires event when signal is sent" $ runHeadlessApp $ void . _process_signal <$> sendSignalTest+ getSawMessage procOut "c\n"+ pure () + it "sends signals" $ do+ () <- runHeadlessApp $ void . _process_exit <$> sendSignalTest+ pure ()+ it "fires event when signal is sent" $ do+ () <- runHeadlessApp $ void . _process_signal <$> sendSignalTest+ pure ()+ where sendSignalTest :: MonadHeadlessApp t m => m (Process t ByteString ByteString) sendSignalTest = do@@ -67,7 +73,6 @@ 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)