shell-conduit 3.0 → 4.0
raw patch · 5 files changed
+249/−258 lines, 5 filesdep +asyncdep +unix
Dependencies added: async, unix
Files
- shell-conduit.cabal +7/−5
- src/Data/Conduit/Shell.hs +29/−25
- src/Data/Conduit/Shell/Process.hs +211/−204
- src/Data/Conduit/Shell/Types.hs +0/−11
- src/Data/Conduit/Shell/Variadic.hs +2/−13
shell-conduit.cabal view
@@ -1,5 +1,5 @@ name: shell-conduit-version: 3.0+version: 4.0 synopsis: Write shell scripts with Conduit description: Write shell scripts with Conduit. See "Data.Conduit.Shell" for documentation. license: BSD3@@ -20,7 +20,8 @@ Data.Conduit.Shell.Process Data.Conduit.Shell.Types Data.Conduit.Shell.Variadic- build-depends: base >= 4 && <5+ build-depends: async >= 2.0.1.5+ , base >= 4 && <5 , bytestring , conduit , conduit-extra@@ -31,10 +32,11 @@ , monads-tf , process , resourcet+ , semigroups , split , template-haskell- , transformers- , transformers-base , text- , semigroups , these+ , transformers+ , transformers-base+ , unix >= 2.7.0.1
src/Data/Conduit/Shell.hs view
@@ -15,35 +15,43 @@ -- The monad instance of Conduit will simply pass along all stdout -- results: ----- >>> run (do echo "Hello"; sed "s/l/a/"; echo "OK!")--- Hello--- OK!--- -- Piping with Conduit's normal pipe will predictably pipe things -- together, as in Bash: ----- >>> run (do shell "echo Hello" $= sed "s/l/a/"; echo "OK!")+-- >>> run (do shell "echo Hello" $| sed "s/l/a/"; echo "OK!") -- Healo -- OK! -- -- Streaming pipes (aka lazy pipes) is also possible: ----- >>> run (do tail' "foo.txt" "-f" $= grep "--line-buffered" "Hello")+-- >>> run (tail' "/tmp/foo.txt" "-f" $| grep "--line-buffered" "Hello") -- Hello, world! -- Oh, hello! -- -- (Remember that @grep@ needs @--line-buffered@ if it is to output -- things line-by-line). --+-- Run custom processes via the @proc@ function:+--+-- >>> run (proc "ls" [])+-- dist LICENSE README.md Setup.hs shell-conduit.cabal src TAGS TODO.org+--+-- Run shell commands via the @shell@ function:+--+-- >>> run (shell "ls")+-- dist LICENSE README.md Setup.hs shell-conduit.cabal src TAGS TODO.org+--+-- Run conduits via the @conduit@ function:+--+-- >>> run (cat "/tmp/foo.txt" $| conduit (do Just x <- await; yield x))+-- Hello!+-- -- == How it works -- -- All executable names in the @PATH@ at compile-time are brought into -- scope as runnable process conduits e.g. @ls@ or @grep@. ----- Stdin/out and stderr are handled as an 'Either' type: 'Chunk' ----- 'Left' is stderr, 'Right' is @stdin@/@stdout@.--- -- All processes are bound as variadic process calling functions, like this: -- -- @@@ -54,9 +62,9 @@ -- But ultimately the types end up being: -- -- @--- rmdir "foo" :: Conduit Chunk m Chunk--- ls :: Conduit Chunk m Chunk--- ls "." :: Conduit Chunk m Chunk+-- rmdir "foo" :: Segment ()+-- ls :: Segment ()+-- ls "." :: Segment () -- @ -- -- Etc.@@ -64,8 +72,7 @@ -- Run all shell scripts with 'run': -- -- @--- run :: (MonadIO m, MonadBaseControl IO m)--- => Conduit Chunk (ShellT m) Chunk -> m ()+-- run :: Segment r -> IO r -- @ -- -- == String types@@ -86,19 +93,16 @@ module Data.Conduit.Shell (-- * Running scripts run- -- * Running custom processes+ -- * Making segments ,shell ,proc- -- * I/O chunks- ,bytes- ,unbytes- ,withRights- ,redirect- ,quiet- ,writeChunks- ,discardChunks- -- * Re-exports- -- $exports+ ,conduit+ -- * Composition of segments+ ,($|)+ ,Segment+ ,ProcessException(..)+ -- * Re-exports+ -- $exports ,module Data.Conduit.Shell.PATH ,module Data.Conduit.Shell.Types ,module Data.Conduit.Shell.Variadic
src/Data/Conduit/Shell/Process.hs view
@@ -1,5 +1,7 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE DeriveDataTypeable #-} -- | Reading from the process. @@ -7,238 +9,243 @@ (-- * Running scripts run -- * Running processes+ ,conduit ,Data.Conduit.Shell.Process.shell ,Data.Conduit.Shell.Process.proc- -- * I/O chunks- ,bytes- ,unbytes- ,withRights- ,redirect- ,quiet- ,writeChunks- ,discardChunks- -- * Low-level internals- ,conduitProcess+ ,($|)+ ,Segment+ ,ProcessException(..)+ ,ToChunk(..) ) where -import Data.Conduit.Shell.Types- import Control.Applicative-import Control.Concurrent (forkIO)-import Control.Concurrent.Chan-import qualified Control.Exception as E+import Control.Concurrent.Async+import Control.Exception import Control.Monad-import Control.Monad.Fix-import Control.Monad.Trans-import Control.Monad.Trans.Resource+import Control.Monad.IO.Class import Data.ByteString (ByteString) import qualified Data.ByteString as S import Data.Conduit-import Data.Conduit.List (sourceList)+import Data.Conduit.Binary import qualified Data.Conduit.List as CL-import Data.Conduit.Process-import Data.Semigroup-import Data.These-import System.Exit (ExitCode(..))+import Data.Typeable+import System.Exit import System.IO-import qualified System.Process+import System.Posix.IO (createPipe, fdToHandle)+import System.Process+import System.Process.Internals (createProcess_) --- | Extract the stdout values from the stream, discarding any errors.-bytes :: Monad m => Conduit Chunk m ByteString-bytes = CL.mapMaybe (either (const Nothing) Just)+-- | A pipeable segment. Either a conduit or a process.+data Segment r+ = SegmentConduit (ConduitM ByteString (Either ByteString ByteString) IO r)+ | SegmentProcess (Handles -> IO r) --- | Extract the stdout values from the stream, discarding any errors.-unbytes :: Monad m => Conduit ByteString m Chunk-unbytes = CL.map Right+instance Monad Segment where+ return = SegmentConduit . return+ SegmentConduit c >>= f =+ SegmentProcess (conduitToProcess c) >>=+ f+ SegmentProcess f >>= g =+ SegmentProcess+ (\handles ->+ do x <- f handles+ case g x of+ SegmentConduit c ->+ conduitToProcess c handles+ SegmentProcess p -> p handles) --- | Run a shell command.-shell :: (MonadResource m) => String -> Conduit Chunk m Chunk-shell = conduitProcess . System.Process.shell+instance Functor Segment where+ fmap = liftM --- | Run a shell command.-proc :: (MonadResource m) => String -> [String] -> Conduit Chunk m Chunk-proc px args = conduitProcess (System.Process.proc px args)+instance Applicative Segment where+ (<*>) = ap; pure = return --- | Size of buffer used to read from process.-bufSize :: Int-bufSize = 64 * 1024+instance MonadIO Segment where+ liftIO x = SegmentProcess (const x) --- | Do something with just the rights.-withRights :: (Monad m)- => Conduit ByteString m ByteString -> Conduit Chunk m Chunk-withRights f =- getZipConduit- (ZipConduit f' *>- ZipConduit g')- where f' =- CL.mapMaybe (either (const Nothing) Just) =$=- f =$=- CL.map Right- g' = CL.filter isLeft+-- | Process handles: @stdin@, @stdout@, @stderr@+data Handles =+ Handles Handle Handle Handle --- | Redirect the given chunk type to the other type.-redirect :: Monad m- => ChunkType -> Conduit Chunk m Chunk-redirect ty =- CL.map (\c' ->- case c' of- Left x' ->- case ty of- Stderr -> Right x'- Stdout -> c'- Right x' ->- case ty of- Stderr -> c'- Stdout -> Left x')+-- | Process running exception.+data ProcessException =+ ProcessException CreateProcess+ ExitCode+ deriving (Typeable) --- | Discard any output from the command: make it quiet.-quiet :: (Monad m,MonadIO m) => Conduit Chunk m Chunk -> Conduit Chunk m Chunk-quiet m = m $= discardChunks+instance Exception ProcessException --- | Run a shell scripting conduit.-run :: (MonadIO m,MonadBaseControl IO m)- => Conduit Chunk (ShellT m) Chunk -> m ()-run p =- runResourceT- (runShellT (sourceList [] $=- p $$- writeChunks))+instance Show ProcessException where+ show (ProcessException cp ec) =+ concat ["The "+ ,case cmdspec cp of+ ShellCommand s -> "shell command " ++ show s+ RawCommand f args ->+ "raw command: " +++ show (f : args)+ ," returned a failure exit code: "+ ,case ec of+ ExitFailure i -> show i+ _ -> show ec] --- | Write chunks to stdout and stderr.-writeChunks :: (MonadIO m)- => Consumer Chunk m ()-writeChunks =- awaitForever- (\c ->- case c of- Left e -> liftIO (S.hPut stderr e)- Right o -> liftIO (S.hPut stdout o))+-- | Convert a process or a conduit to a segment.+class ToSegment a where+ type SegmentResult a+ toSegment :: a -> Segment (SegmentResult a) --- | Discard all chunks.-discardChunks :: (MonadIO m)- => Consumer Chunk m ()-discardChunks = awaitForever (const (return ()))+instance ToSegment (Segment r) where+ type SegmentResult (Segment r) = r+ toSegment = id --- | Conduit of process.-conduitProcess :: (MonadResource m)- => CreateProcess -> Conduit Chunk m Chunk-conduitProcess cp =- bracketP createp closep startProxy- where createp =- createProcess- cp {std_in = CreatePipe- ,std_out = CreatePipe- ,std_err = CreatePipe}- closep (Just cin,Just cout,Just cerr,ph) =- do hClose cin- hClose cout- hClose cerr- _ <- waitForProcess' ph- return ()- closep _ = error "conduitProcess: unexpected arguments to closep"+instance (a ~ ByteString,ToChunk b,m ~ IO) => ToSegment (ConduitM a b m r) where+ type SegmentResult (ConduitM a b m r) = r+ toSegment f =+ SegmentConduit (f `fuseUpstream` CL.map toChunk) --- | Start proxying from conduit to process back to conduit.-startProxy :: (MonadIO m,MonadThrow m)- => (Maybe Handle,Maybe Handle,Maybe Handle,ProcessHandle)- -> ConduitM Chunk Chunk m ()-startProxy (Just cin,Just cout,Just cerr,ph) = interleave- where interleave =- do end <- proxyInterleaved- liftIO (hClose cin)- remainders cout cerr- ec <- liftIO (maybe (waitForProcess' ph) return end)+instance ToSegment CreateProcess where+ type SegmentResult CreateProcess = ()+ toSegment = liftProcess++-- | Used to allow outputting stdout or stderr.+class ToChunk a where+ toChunk :: a -> Either ByteString ByteString++instance ToChunk ByteString where+ toChunk = Left++instance ToChunk (Either ByteString ByteString) where+ toChunk = id++-- | Run a shell command.+shell :: String -> Segment ()+shell = liftProcess . System.Process.shell++-- | Run a process command.+proc :: String -> [String] -> Segment ()+proc name args = liftProcess (System.Process.proc name args)++-- | Run a segment.+run :: Segment r -> IO r+run (SegmentConduit c) =+ run (SegmentProcess (conduitToProcess c))+run (SegmentProcess p) =+ p (Handles stdin stdout stderr)++-- | Fuse two segments (either processes or conduits).+($|) :: Segment () -> Segment b -> Segment b+x $| y = x `fuseSegment` y+infixl 0 $|++-- | Lift a conduit into a segment.+conduit :: (a ~ ByteString,ToChunk b,m ~ IO) => ConduitM a b m r -> Segment r+conduit f = SegmentConduit (f `fuseUpstream` CL.map toChunk)++-- | Lift a process into a segment.+liftProcess :: CreateProcess -> Segment ()+liftProcess cp =+ SegmentProcess+ (\(Handles inh outh errh) ->+ let config =+ cp {std_in = UseHandle inh+ ,std_out = UseHandle outh+ ,std_err = UseHandle errh+ ,close_fds = True}+ in do (Nothing,Nothing,Nothing,ph) <- createProcess_ "liftProcess" config+ ec <- waitForProcess ph case ec of ExitSuccess -> return ()- ExitFailure i ->- monadThrow (ShellExitFailure i)- proxyInterleaved =- do proxy cout Right- proxy cerr Left- ended <- liftIO (getProcessExitCode ph)- case ended of- Just{} -> return ended- Nothing ->- do minp <- await- case minp of- Nothing -> return Nothing- Just chunk ->- do case chunk of- Left{} -> yield chunk- Right bytes ->- liftIO (do S.hPut cin bytes- hFlush cin)- proxyInterleaved-startProxy _ = error "startProxy: unexpected arguments"+ _ ->+ throwIO (ProcessException cp ec)) --- | Concurrently yield eithers downstream.-remainders :: MonadIO m- => Handle -> Handle -> ConduitM i (Either ByteString ByteString) m ()-remainders cout cerr =- do chan <- liftIO newChan- void (liftIO (forkIO (remainder chan cout Right)))- void (liftIO (forkIO (remainder chan cerr Left)))- fix (\loop done ->- case done of- Just (These () ()) -> return ()- _ ->- do chunk <- liftIO (readChan chan)- case chunk of- Left mchunk ->- case mchunk of- Nothing ->- loop (done <>- Just (This ()))- Just chunk ->- do yield (Left chunk)- loop done- Right mchunk ->- case mchunk of- Nothing ->- loop (done <>- Just (That ()))- Just chunk ->- do yield (Right chunk)- loop done)- (Nothing :: Maybe (These () ()))+-- | Convert a conduit to a process.+conduitToProcess :: ConduitM ByteString (Either ByteString ByteString) IO r+ -> (Handles -> IO r)+conduitToProcess c (Handles inh outh errh) =+ sourceHandle inh $$ c `fuseUpstream`+ sinkHandles outh errh --- | Proxy final results from the handle and yield them.-remainder :: Chan (Either (Maybe ByteString) (Maybe ByteString))- -> Handle- -> (Maybe ByteString -> Either (Maybe ByteString) (Maybe ByteString))- -> IO ()-remainder chan h cons =- do bytes <- S.hGetSome h bufSize- if S.null bytes- then writeChan chan (cons Nothing)- else do writeChan chan (cons (Just bytes))- remainder chan h cons+-- | Sink everything into the two handles.+sinkHandles :: Handle -> Handle -> Consumer (Either ByteString ByteString) IO ()+sinkHandles out err =+ CL.mapM_ (\ebs ->+ case ebs of+ Left bs -> S.hPut out bs+ Right bs -> S.hPut err bs) --- | Proxy live results from the given handle and yield them.-proxy :: MonadIO m- => Handle -> (ByteString -> o) -> ConduitM i o m ()-proxy h cons =- do ready <- liftIO (hReady' h)- if not ready- then return ()- else do bytes <- liftIO (S.hGetSome h bufSize)- yield (cons bytes)- proxy h cons+-- | Create a pipe.+createHandles :: IO (Handle, Handle)+createHandles =+ mask_ (do (inFD,outFD) <- createPipe+ x <- fdToHandle inFD+ y <- fdToHandle outFD+ hSetBuffering x NoBuffering+ hSetBuffering y NoBuffering+ return (x,y)) --- | Is the handle ready? Catches any exceptions.-hReady' :: Handle -> IO Bool-hReady' h =- E.catch (hReady h)- (\(E.SomeException _) -> return False)+-- | Fuse two processes.+fuseProcess :: (Handles -> IO ()) -> (Handles -> IO r) -> (Handles -> IO r)+fuseProcess left right (Handles in1 out2 err) =+ do (in2,out1) <- createHandles+ runConcurrently+ (Concurrently+ (left (Handles in1 out1 err) `finally`+ hClose out1) *>+ Concurrently+ (right (Handles in2 out2 err) `finally`+ hClose in2)) --- | A safer 'waitForProcess'.-waitForProcess' :: ProcessHandle -> IO ExitCode-waitForProcess' ph =- E.catch (waitForProcess ph)- (\(E.SomeException _) ->- return ExitSuccess)+-- | Fuse two conduits.+fuseConduit :: Monad m+ => ConduitM ByteString (Either ByteString ByteString) m ()+ -> ConduitM ByteString (Either ByteString ByteString) m r+ -> ConduitM ByteString (Either ByteString ByteString) m r+fuseConduit left right = left =$= getZipConduit right'+ where right' =+ ZipConduit (CL.filter isRight) *>+ ZipConduit+ (CL.mapMaybe (either (const Nothing) Just) =$=+ right)+ isRight Right{} = True+ isRight Left{} = False --- | Polyfill for base < 4.7-isLeft :: Either a b -> Bool-isLeft (Left _) = True-isLeft _ = False+-- | Fuse a conduit with a process.+fuseConduitProcess :: ConduitM ByteString (Either ByteString ByteString) IO ()+ -> (Handles -> IO r)+ -> (Handles -> IO r)+fuseConduitProcess left right (Handles in1 out2 err) =+ do (in2,out1) <- createHandles+ runConcurrently+ (Concurrently+ ((sourceHandle in1 $$ left =$+ sinkHandles out1 err) `finally`+ hClose out1) *>+ Concurrently+ (right (Handles in2 out2 err) `finally`+ hClose in2))++-- | Fuse a process with a conduit.+fuseProcessConduit :: (Handles -> IO ())+ -> ConduitM ByteString (Either ByteString ByteString) IO r+ -> (Handles -> IO r)+fuseProcessConduit left right (Handles in1 out2 err) =+ do (in2,out1) <- createHandles+ runConcurrently+ (Concurrently+ (left (Handles in1 out1 err) `finally`+ hClose out1) *>+ Concurrently+ ((sourceHandle in2 $$ right `fuseUpstream`+ sinkHandles out2 err) `finally`+ hClose in2))++-- | Fuse one segment with another.+fuseSegment :: Segment () -> Segment r -> Segment r+SegmentConduit x `fuseSegment` SegmentConduit y =+ SegmentConduit (fuseConduit x y)+SegmentConduit x `fuseSegment` SegmentProcess y =+ SegmentProcess (fuseConduitProcess x y)+SegmentProcess x `fuseSegment` SegmentConduit y =+ SegmentProcess (fuseProcessConduit x y)+SegmentProcess x `fuseSegment` SegmentProcess y =+ SegmentProcess (fuseProcess x y)
src/Data/Conduit/Shell/Types.hs view
@@ -21,19 +21,8 @@ import Control.Monad.Trans.Class import Control.Monad.Trans.Control import Control.Monad.Trans.Resource-import Data.ByteString (ByteString) import Data.Conduit import Data.Typeable---- | A chunk, either stdout/stdin or stderr. Used for both input and--- output.-type Chunk = Either ByteString ByteString---- | Either stdout or stderr.-data ChunkType- = Stderr -- ^ Stderr.- | Stdout -- ^ Stdin or stdout.- deriving (Eq,Enum,Bounded) -- | Shell transformer. newtype ShellT m a =
src/Data/Conduit/Shell/Variadic.hs view
@@ -9,12 +9,9 @@ ,CmdArg(..)) where -import Control.Monad.Trans.Resource import qualified Data.ByteString as SB import qualified Data.ByteString.Lazy as LB-import Data.Conduit import Data.Conduit.Shell.Process-import Data.Conduit.Shell.Types import qualified Data.Text as ST import qualified Data.Text.Encoding as ST import qualified Data.Text.Lazy as LT@@ -26,22 +23,14 @@ variadicProcess name = spr name [] -- | Make the final conduit.-makeProcessLauncher :: (MonadResource m)- => String -> [ST.Text] -> Conduit Chunk m Chunk+makeProcessLauncher :: String -> [ST.Text] -> Segment () makeProcessLauncher name args = proc name (map ST.unpack args) -- | Process return type. class ProcessType t where spr :: String -> [ST.Text] -> t --- | The real type should be:------ @ConduitM Chunk Chunk m ()@------ But with this more liberal instance head we catch all cases in the--- instance resolver, and then apply the equality restrictions later.----instance (MonadResource m, c ~ Chunk, c' ~ Chunk, r ~ ()) => ProcessType (ConduitM c c' m r) where+instance (r ~ ()) => ProcessType (Segment r) where spr name args = makeProcessLauncher name (reverse args) -- | Accept strings as arguments.