process-conduit 0.4.1 → 0.5.0

raw patch · 5 files changed

+171/−166 lines, 5 filesdep +HUnitdep +control-monad-loopdep +hspecdep ~bytestringdep ~conduitdep ~mtl

Dependencies added: HUnit, control-monad-loop, hspec

Dependency ranges changed: bytestring, conduit, mtl, process, shakespeare-text, text

Files

• Data/Conduit/Process.hs +98/−104
• examples/TestConduit.hs +0/−8
• examples/TestQQ.hs +0/−9
• process-conduit.cabal +36/−45
• test.hs +37/−0

@@ -1,104 +1,98 @@-{-# LANGUAGE FlexibleContexts, OverloadedStrings, DoAndIfThenElse, BangPatterns #-}-module Data.Conduit.Process (-  -- * Run process-  pipeProcess,-  sourceProcess,-  conduitProcess,-  -  -- * Run shell command-  pipeCmd,-  sourceCmd,-  conduitCmd,-  -  -- * Convenience re-exports-  shell,-  proc,-  CreateProcess(..),-  CmdSpec(..),-  StdStream(..),-  ProcessHandle,-  ) where--import Control.Concurrent-import Control.Concurrent.MVar-import Control.Monad-import Control.Monad.Trans-import Control.Monad.Trans.Resource-import qualified Data.ByteString as B-import Data.Conduit-import qualified Data.Conduit.List as C-import System.Exit-import System.IO-import System.Process--bufSize :: Int-bufSize = 64 * 1024---- | Pipe of process-pipeProcess-  :: MonadResource m-     => CreateProcess-     -> Pipe B.ByteString B.ByteString m ()-pipeProcess cp = flip PipeM (return ()) $ do-  (_, (Just cin, Just cout, _, ph)) <- allocate createp closep-  mvar <- liftIO newEmptyMVar-  return $ go cin cout ph mvar False B.hGetNonBlocking-  where-    createp = createProcess cp-      { std_in  = CreatePipe-      , std_out = CreatePipe-      }--    closep (Just cin, Just cout, _, ph) = do-      hClose cin-      hClose cout-      _ <- waitForProcess ph-      return ()--    go !cin !cout !ph !mvar !wait !rd = do-      out <- liftIO $ rd cout bufSize-      if B.null out then do-        end <- liftIO $ getProcessExitCode ph-        case end of-          Just ec -> do-            lift $ when (ec /= ExitSuccess) $ monadThrow ec-            Done Nothing ()-          Nothing ->-            if wait then do-              emp <- liftIO $ isEmptyMVar mvar-              if emp then do-                go cin cout ph mvar wait rd-              else do-                liftIO $ takeMVar mvar-                go cin cout ph mvar False rd-            else do-              NeedInput-                (\inp -> do-                    liftIO $ do-                      B.hPut cin inp-                      forkIO (hFlush cin >>= putMVar mvar)-                    go cin cout ph mvar True rd)-                (do liftIO (hClose cin)-                    go cin cout ph mvar wait B.hGetSome)-      else do-        HaveOutput (go cin cout ph mvar wait rd) (return ()) out---- | Source of process-sourceProcess :: MonadResource m => CreateProcess -> Source m B.ByteString-sourceProcess cp = C.sourceNull $= conduitProcess cp---- | Conduit of process-conduitProcess :: MonadResource m => CreateProcess -> Conduit B.ByteString m B.ByteString-conduitProcess = pipeProcess---- | Pipe of shell command-pipeCmd :: MonadResource m => String -> Pipe B.ByteString B.ByteString m ()-pipeCmd = pipeProcess . shell---- | Source of shell command-sourceCmd :: MonadResource m => String -> Source m B.ByteString-sourceCmd = sourceProcess . shell---- | Conduit of shell command-conduitCmd :: MonadResource m => String -> Conduit B.ByteString m B.ByteString-conduitCmd = conduitProcess . shell+{-# LANGUAGE FlexibleContexts, OverloadedStrings, BangPatterns #-}
+module Data.Conduit.Process (
+  -- * Run process
+  sourceProcess,
+  conduitProcess,
+
+  -- * Run shell command
+  sourceCmd,
+  conduitCmd,
+
+  -- * Convenience re-exports
+  shell,
+  proc,
+  CreateProcess(..),
+  CmdSpec(..),
+  StdStream(..),
+  ProcessHandle,
+  ) where
+
+import qualified Control.Exception as E
+import Control.Monad
+import Control.Monad.Trans
+import Control.Monad.Trans.Loop
+import Control.Monad.Trans.Resource
+import qualified Data.ByteString as S
+import Data.Conduit
+import qualified Data.Conduit.List as CL
+import Data.Maybe
+import System.Exit
+import System.IO
+import System.Process
+
+bufSize :: Int
+bufSize = 64 * 1024
+
+-- | Conduit of process
+conduitProcess
+  :: MonadResource m
+     => CreateProcess
+     -> GConduit S.ByteString m S.ByteString
+conduitProcess cp = do
+  (_, (Just cin, Just cout, _, ph)) <- lift $ allocate createp closep
+
+  repeatLoopT $ do
+    -- if process's outputs are available, then yields them.
+    repeatLoopT $ do
+      b <- liftIO $ hReady' cout
+      when (not b) exit
+      out <- liftIO $ S.hGetSome cout bufSize
+      void $ lift . lift $ yield out
+
+    -- if process exited, then exit
+    end <- liftIO $ getProcessExitCode ph
+    when (isJust end) exit
+
+    -- if upper stream ended, then exit
+    inp <- lift await
+    when (isNothing inp) exit
+
+    -- put input to process
+    liftIO $ S.hPut cin $ fromJust inp
+    liftIO $ hFlush cin
+
+  -- uppstream or process is done.
+  -- process rest outputs.
+  liftIO $ hClose cin
+  repeatLoopT $ do
+    out <- liftIO $ S.hGetSome cout bufSize
+    when (S.null out) exit
+    lift $ yield out
+  ec <- liftIO $ waitForProcess ph
+  lift $ when (ec /= ExitSuccess) $ monadThrow ec
+
+  where
+    createp = createProcess cp
+      { std_in  = CreatePipe
+      , std_out = CreatePipe
+      }
+
+    closep (Just cin, Just cout, _, ph) = do
+      hClose cin
+      hClose cout
+      _ <- waitForProcess ph
+      return ()
+
+    hReady' h = hReady h `E.catch` \(E.SomeException _) -> return False
+
+-- | Source of process
+sourceProcess :: MonadResource m => CreateProcess -> GSource m S.ByteString
+sourceProcess cp = CL.sourceNull >+> conduitProcess cp
+
+-- | Conduit of shell command
+conduitCmd :: MonadResource m => String -> GConduit S.ByteString m S.ByteString
+conduitCmd = conduitProcess . shell
+
+-- | Source of shell command
+sourceCmd :: MonadResource m => String -> GSource m S.ByteString
+sourceCmd = sourceProcess . shell

@@ -1,8 +0,0 @@-import Data.Conduit-import Data.Conduit.Binary-import Data.Conduit.Process-import System.IO--main :: IO ()-main = runResourceT $ do-  sourceCmd "ls" $= conduitCmd "sort" $$ sinkHandle stdout

@@ -1,9 +0,0 @@-{-# LANGUAGE QuasiQuotes #-}--import System.Process.QQ-import System.Environment--main :: IO ()-main = do-  [url] <- getArgs-  print =<< [cmd|curl #{url}|]

@@ -1,60 +1,51 @@-Name:                process-conduit-Version:             0.4.1-Synopsis:            Conduits for processes+name:                process-conduit+version:             0.5.0+synopsis:            Conduits for processes -Description:+description:   Conduits for processes.   For more details: <https://github.com/tanakh/process-conduit/blob/master/README.md> -Homepage:            http://github.com/tanakh/process-conduit-License:             BSD3-License-file:        LICENSE-Author:              Hideyuki Tanaka-Maintainer:          Hideyuki Tanaka <tanaka.hideyuki@gmail.com>-Copyright:           (c) 2011-2012, Hideyuki Tanaka-Category:            System, Conduit-Build-type:          Simple-Cabal-version:       >=1.8--Extra-source-files:  README.md+homepage:            http://github.com/tanakh/process-conduit+license:             BSD3+license-file:        LICENSE+author:              Hideyuki Tanaka+maintainer:          Hideyuki Tanaka <tanaka.hideyuki@gmail.com>+copyright:           (c) 2011-2012, Hideyuki Tanaka+category:            System, Conduit+build-type:          Simple+cabal-version:       >=1.8 -Source-repository head-  Type:                git-  Location:            git://github.com/tanakh/process-conduit.git+extra-source-files:  README.md -Flag examples-  Description:         Build examples-  Default:             False+source-repository head+  type:                git+  location:            git://github.com/tanakh/process-conduit.git -Library-  Exposed-modules:     Data.Conduit.Process+library+  exposed-modules:     Data.Conduit.Process                        System.Process.QQ   -  Build-depends:       base             == 4.*+  build-depends:       base             == 4.*                      , template-haskell >= 2.4 && < 2.8-                     , mtl              >= 2.0 && < 2.2-                     , bytestring       == 0.9.*-                     , text             == 0.11.*-                     , process          >= 1.0 && < 1.2-                     , conduit          == 0.4.*+                     , mtl              >= 2.0+                     , control-monad-loop == 0.1.*+                     , bytestring       >= 0.9+                     , text             >= 0.11+                     , process          >= 1.0+                     , conduit          == 0.5.*                      , resourcet-                     , shakespeare-text+                     , shakespeare-text >= 1.0 -  Ghc-options:         -Wall+  ghc-options:         -Wall -Executable process-conduit-  Hs-source-dirs:      examples-  Main-is:             TestConduit.hs-  Build-depends:       base            == 4.*+test-suite process-conduit-test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      dist+  main-is:             ../test.hs+  build-depends:       base            == 4.*+                     , bytestring+                     , hspec           >= 1.2+                     , HUnit           >= 1.2                      , conduit                      , process-conduit-  if !flag(examples)-    Buildable:         False--Executable process-qq-  Hs-source-dirs:      examples-  Main-is:             TestQQ.hs-  Build-depends:       base            == 4.*-                     , process-conduit-  if !flag(examples)-    Buildable:         False

@@ -0,0 +1,37 @@+{-# LANGUAGE OverloadedStrings, QuasiQuotes #-}
+
+import Data.Conduit.Process
+import System.Process.QQ
+
+import qualified Data.ByteString.Lazy.Char8 as L
+import Data.Conduit
+import qualified Data.Conduit.Binary as CB
+import Test.Hspec.Monadic
+import Test.Hspec.HUnit
+import Test.HUnit
+
+main :: IO ()
+main = hspec $ do
+  describe "process conduit" $ do
+    it "get process's output" $ do
+      r <- runResourceT $ sourceCmd "echo abc def" $$ CB.take (10^9)
+      L.words r @?= ["abc", "def"]
+
+    it "act as conduit" $ do
+      r <- runResourceT $
+        sourceProcess (proc "echo" ["zxc\nasd\nqwe"])
+        $$ conduitCmd "sort"
+        =$ CB.take (10^9)
+      L.words r @?= ["asd", "qwe", "zxc"]
+
+  describe "quasiquoter" $ do
+    it "get process's output" $ do
+      r <- [cmd|echo abc def|]
+      L.words r @?= ["abc", "def"]
+
+    it "act as conduit" $ do
+      r <- runResourceT $
+        sourceProcess (proc "echo" ["zxc\nasd\nqwe"])
+        $$ [ccmd|sort|]
+        =$ CB.take (10^9)
+      L.words r @?= ["asd", "qwe", "zxc"]