diff --git a/Data/Conduit/Process.hs b/Data/Conduit/Process.hs --- a/Data/Conduit/Process.hs +++ b/Data/Conduit/Process.hs @@ -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 diff --git a/examples/TestConduit.hs b/examples/TestConduit.hs deleted file mode 100644 --- a/examples/TestConduit.hs +++ /dev/null @@ -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 diff --git a/examples/TestQQ.hs b/examples/TestQQ.hs deleted file mode 100644 --- a/examples/TestQQ.hs +++ /dev/null @@ -1,9 +0,0 @@ -{-# LANGUAGE QuasiQuotes #-} - -import System.Process.QQ -import System.Environment - -main :: IO () -main = do - [url] <- getArgs - print =<< [cmd|curl #{url}|] diff --git a/process-conduit.cabal b/process-conduit.cabal --- a/process-conduit.cabal +++ b/process-conduit.cabal @@ -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: -Homepage: http://github.com/tanakh/process-conduit -License: BSD3 -License-file: LICENSE -Author: Hideyuki Tanaka -Maintainer: Hideyuki Tanaka -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 +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 diff --git a/test.hs b/test.hs new file mode 100644 --- /dev/null +++ b/test.hs @@ -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"]