diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,13 @@ # Revision history for haskell-amqp-utils +## 0.4.5.0 -- 2020-02-10 + +* format with stylish-haskell +* agitprop: rename variables in code +* new options -d (--dirscan) and -u (--remove) for agitprop hotfolder mode +* cwd to hotfolder directory in agitprop hotfolder mode +* change logging of boolean parameters + ## 0.4.4.1 -- 2020-08-19 * change default connection timeout from 60 to 600s diff --git a/Network/AMQP/Utils/Connection.hs b/Network/AMQP/Utils/Connection.hs --- a/Network/AMQP/Utils/Connection.hs +++ b/Network/AMQP/Utils/Connection.hs @@ -2,17 +2,17 @@ module Network.AMQP.Utils.Connection where -import qualified Data.ByteString as B -import Data.Default.Class -import qualified Data.Text as T -import Network.AMQP -import Network.AMQP.Utils.Helpers -import Network.AMQP.Utils.Options -import qualified Network.Connection as N -import Network.TLS -import Network.TLS.Extra -import System.Timeout -import System.X509 +import qualified Data.ByteString as B +import Data.Default.Class +import qualified Data.Text as T +import Network.AMQP +import Network.AMQP.Utils.Helpers +import Network.AMQP.Utils.Options +import qualified Network.Connection as N +import Network.TLS +import Network.TLS.Extra +import System.Timeout +import System.X509 -- | opens a connection and a channel connect :: Args -> IO (Connection, Channel) @@ -74,6 +74,6 @@ myCert (Just cert') (Just key') _ = do result <- credentialLoadX509 cert' key' case result of - Left x -> printparam "ERROR" x >> return Nothing + Left x -> printparam "ERROR" x >> return Nothing Right x -> return $ Just x myCert _ _ _ = return Nothing diff --git a/Network/AMQP/Utils/Helpers.hs b/Network/AMQP/Utils/Helpers.hs --- a/Network/AMQP/Utils/Helpers.hs +++ b/Network/AMQP/Utils/Helpers.hs @@ -2,28 +2,28 @@ module Network.AMQP.Utils.Helpers where -import Control.Concurrent -import qualified Control.Exception as X -import Control.Monad +import Control.Concurrent +import qualified Control.Exception as X +import Control.Monad import qualified Data.ByteString.Lazy.Char8 as BL -import qualified Data.ByteString.UTF8 as BS -import Data.Int (Int64) -import Data.List -import qualified Data.Map as M -import Data.Maybe -import qualified Data.Text as T -import Data.Time -import Data.Time.Clock.POSIX -import Data.Word (Word16) -import Network.AMQP -import Network.AMQP.Types -import Network.AMQP.Utils.Options -import Network.Socket (PortNumber) -import System.Directory (removeFile) -import System.Environment (getEnvironment) -import System.Exit -import System.IO -import System.Process +import qualified Data.ByteString.UTF8 as BS +import Data.Int (Int64) +import Data.List +import qualified Data.Map as M +import Data.Maybe +import qualified Data.Text as T +import Data.Time +import Data.Time.Clock.POSIX +import Data.Word (Word16) +import Network.AMQP +import Network.AMQP.Types +import Network.AMQP.Utils.Options +import Network.Socket (PortNumber) +import System.Directory (removeFile) +import System.Environment (getEnvironment) +import System.Exit +import System.IO +import System.Process -- | print config parameters class (Show a) => @@ -44,7 +44,7 @@ instance (Flexprint a) => Flexprint (Maybe a) where empty = isNothing - printparam _ Nothing = return () + printparam _ Nothing = return () printparam x (Just y) = printparam x y instance Flexprint String where @@ -63,7 +63,8 @@ flexprint x = hPutStrLn stderr "" >> BL.hPut stderr x >> hPutStrLn stderr "" empty = BL.null -instance Flexprint Bool +instance Flexprint Bool where + empty = not instance Flexprint Int @@ -119,13 +120,13 @@ -- | showing a FieldValue valueshow :: FieldValue -> String valueshow (FVString value) = BS.toString value -valueshow (FVInt8 value) = show value -valueshow (FVInt16 value) = show value -valueshow (FVInt32 value) = show value -valueshow (FVInt64 value) = show value -valueshow (FVFloat value) = show value +valueshow (FVInt8 value) = show value +valueshow (FVInt16 value) = show value +valueshow (FVInt32 value) = show value +valueshow (FVInt64 value) = show value +valueshow (FVFloat value) = show value valueshow (FVDouble value) = show value -valueshow value = show value +valueshow value = show value -- | skip showing body head if binary type isimage :: Maybe String -> Bool @@ -140,11 +141,11 @@ anriss' x = case x of Nothing -> id - Just y -> BL.take y + Just y -> BL.take y -- | callback cmdline with optional parameters printopt :: (String, Maybe String) -> [String] -printopt (_, Nothing) = [] +printopt (_, Nothing) = [] printopt (opt, Just s) = [opt, s] -- | prints header and head on stderr and returns @@ -196,7 +197,7 @@ return (cmdline, environment) where step (_, Nothing) xs = xs - step (k, Just v) xs = ("AMQP_" ++ k, v) : xs + step (k, Just v) xs = ("AMQP_" ++ k, v) : xs cmdline = concat (map @@ -250,7 +251,7 @@ now' = case timediff of Just "now" -> Nothing - _ -> showtime tz $ Just nowutc + _ -> showtime tz $ Just nowutc timestamp' = showtime tz msgtimeutc timestamp'' = liftM3 @@ -349,7 +350,7 @@ if isJust action && isJust sout then ((fromJust action $ exitcode) (fromJust sout)) >> acke envi a else case exitcode of - ExitSuccess -> acke envi a + ExitSuccess -> acke envi a ExitFailure _ -> reje envi a if (cleanupTmpFile a) then X.catch diff --git a/Network/AMQP/Utils/Options.hs b/Network/AMQP/Utils/Options.hs --- a/Network/AMQP/Utils/Options.hs +++ b/Network/AMQP/Utils/Options.hs @@ -1,18 +1,18 @@ module Network.AMQP.Utils.Options where import qualified Data.ByteString.Char8 as BS -import Data.Default.Class -import Data.Int (Int64) -import qualified Data.Map as M -import Data.Maybe -import Data.Text (Text, pack) -import Data.Version (showVersion) -import Data.Word (Word16) -import Network.AMQP -import Network.AMQP.Types -import Network.Socket (PortNumber) -import Paths_amqp_utils (version) -import System.Console.GetOpt +import Data.Default.Class +import Data.Int (Int64) +import qualified Data.Map as M +import Data.Maybe +import Data.Text (Text, pack) +import Data.Version (showVersion) +import Data.Word (Word16) +import Network.AMQP +import Network.AMQP.Types +import Network.Socket (PortNumber) +import Paths_amqp_utils (version) +import System.Console.GetOpt portnumber :: Args -> PortNumber portnumber a @@ -23,52 +23,54 @@ -- | A data type for our options data Args = Args - { server :: String - , port :: Maybe PortNumber - , tls :: Bool - , vHost :: String + { server :: String + , port :: Maybe PortNumber + , tls :: Bool + , vHost :: String , currentExchange :: String - , bindings :: [(String, String)] - , rKey :: String - , anRiss :: Maybe Int64 - , fileProcess :: Maybe String - , qName :: Maybe String - , cert :: Maybe String - , key :: Maybe String - , user :: String - , pass :: String - , preFetch :: Word16 - , heartBeat :: Maybe Word16 - , tempDir :: Maybe String - , additionalArgs :: [String] - , connectionName :: Maybe String - , tmpQName :: String - , inputFile :: String - , outputFile :: String - , lineMode :: Bool - , confirm :: Bool - , msgid :: Maybe Text - , msgtype :: Maybe Text - , userid :: Maybe Text - , appid :: Maybe Text - , clusterid :: Maybe Text - , contenttype :: Maybe Text + , bindings :: [(String, String)] + , rKey :: String + , anRiss :: Maybe Int64 + , fileProcess :: Maybe String + , qName :: Maybe String + , cert :: Maybe String + , key :: Maybe String + , user :: String + , pass :: String + , preFetch :: Word16 + , heartBeat :: Maybe Word16 + , tempDir :: Maybe String + , additionalArgs :: [String] + , connectionName :: Maybe String + , tmpQName :: String + , inputFile :: String + , outputFile :: String + , lineMode :: Bool + , confirm :: Bool + , msgid :: Maybe Text + , msgtype :: Maybe Text + , userid :: Maybe Text + , appid :: Maybe Text + , clusterid :: Maybe Text + , contenttype :: Maybe Text , contentencoding :: Maybe Text - , replyto :: Maybe Text - , prio :: Maybe Octet - , corrid :: Maybe Text - , msgexp :: Maybe Text - , msgheader :: Maybe FieldTable - , fnheader :: [String] - , suffix :: [String] - , magic :: Bool - , persistent :: Maybe DeliveryMode - , ack :: Bool - , requeuenack :: Bool - , rpc_timeout :: Double + , replyto :: Maybe Text + , prio :: Maybe Octet + , corrid :: Maybe Text + , msgexp :: Maybe Text + , msgheader :: Maybe FieldTable + , fnheader :: [String] + , suffix :: [String] + , magic :: Bool + , persistent :: Maybe DeliveryMode + , ack :: Bool + , requeuenack :: Bool + , rpc_timeout :: Double , connect_timeout :: Int - , simple :: Bool - , cleanupTmpFile :: Bool + , simple :: Bool + , cleanupTmpFile :: Bool + , removeSentFile :: Bool + , initialScan :: Bool } instance Default Args where @@ -120,6 +122,8 @@ 600 False False + False + False -- | all options allOptions :: [(String, OptDescr (Args -> Args))] @@ -302,6 +306,18 @@ "Allowed file suffixes in hotfolder mode") , ( "a" , Option + ['u'] + ["remove"] + (NoArg (\o -> o {removeSentFile = not (removeSentFile o)})) + ("Toggle removal of sent file in hotfolder mode (default: " ++ show (removeSentFile def) ++ ")")) + , ( "a" + , Option + ['d'] + ["dirscan"] + (NoArg (\o -> o {initialScan = not (initialScan o)})) + ("Toggle initial directory scan in hotfolder mode (default: " ++ show (initialScan def) ++ ")")) + , ( "a" + , Option ['m'] ["magic"] (NoArg (\o -> o {magic = not (magic o)})) @@ -473,4 +489,4 @@ longname 'k' = "konsum" longname 'r' = "arbeite" longname 'p' = "plane" -longname _ = "command" +longname _ = "command" diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ -P PASSWORD --pass=PASSWORD Password for Auth -s INT --heartbeats=INT heartbeat interval (0=disable, default: set by server) -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface - -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 60) + -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) ### examples @@ -98,6 +98,8 @@ -h HEADER=VALUE --header=HEADER=VALUE Message Headers -F HEADERNAME --fnheader=HEADERNAME Put filename into this header -S SUFFIX --suffix=SUFFIX Allowed file suffixes in hotfolder mode + -u --remove Toggle removal of sent file in hotfolder mode (default: False) + -d --dirscan Toggle initial directory scan in hotfolder mode (default: False) -m --magic Toggle setting content-type and -encoding from file contents (default: False) -e --persistent Set persistent delivery -E --nonpersistent Set nonpersistent delivery @@ -112,15 +114,15 @@ -P PASSWORD --pass=PASSWORD Password for Auth -s INT --heartbeats=INT heartbeat interval (0=disable, default: set by server) -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface - -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 60) + -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) -If INPUTFILE is a file, the file is sent as a message and the program -exits. If INPUTFILE is a directory, the directory is watched via inotify -and every file, which is written and closed or moved in gets sent, -optionally only files which match one or several SUFFIXes. Optionally -the file name is written into a message header named HEADERNAME. -Optionally Content-Type and Content-Encoding headers are set via magic -retrieved from file contents. +If INPUTFILE is a file, the file is sent as a message and the program exits. If +INPUTFILE is a directory, the directory is watched via inotify ("hotfolder +mode") and every file, which is written and closed or moved in gets sent, +optionally only files which match one or several SUFFIXes. Optionally the file +name is written into a message header named HEADERNAME. Optionally +Content-Type and Content-Encoding headers are set via magic retrieved from file +contents. Line-by-line mode sends one message per INPUTFILE line. @@ -156,7 +158,7 @@ -P PASSWORD --pass=PASSWORD Password for Auth -s INT --heartbeats=INT heartbeat interval (0=disable, default: set by server) -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface - -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 60) + -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) ### examples @@ -191,7 +193,7 @@ -P PASSWORD --pass=PASSWORD Password for Auth -s INT --heartbeats=INT heartbeat interval (0=disable, default: set by server) -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface - -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 60) + -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) ### examples diff --git a/agitprop.hs b/agitprop.hs --- a/agitprop.hs +++ b/agitprop.hs @@ -1,31 +1,33 @@ -{-# LANGUAGE CPP #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE OverloadedStrings #-} -- generic AMQP publisher -import Control.Concurrent -import qualified Control.Exception as X -import Control.Monad (forever) -import qualified Data.ByteString.Lazy.Char8 as BL +import Control.Concurrent +import qualified Control.Exception as X +import Control.Monad (forever) +import qualified Data.ByteString.Lazy.Char8 as BL #if MIN_VERSION_hinotify(0,3,10) -import qualified Data.ByteString.Char8 as BS +import qualified Data.ByteString.Char8 as BS #endif -import Data.List (isSuffixOf) -import Data.Maybe -import qualified Data.Text as T -import Data.Time -import Data.Time.Clock.POSIX -import Data.Version (showVersion) -import Data.Word (Word64) -import Magic -import Network.AMQP -import Network.AMQP.Types -import Network.AMQP.Utils.Connection -import Network.AMQP.Utils.Helpers -import Network.AMQP.Utils.Options -import Paths_amqp_utils (version) -import System.Environment -import System.Exit -import System.INotify -import qualified System.Posix.Files as F +import Data.List (isSuffixOf) +import Data.Maybe +import qualified Data.Text as T +import Data.Time +import Data.Time.Clock.POSIX +import Data.Version (showVersion) +import Data.Word (Word64) +import Magic +import Network.AMQP +import Network.AMQP.Types +import Network.AMQP.Utils.Connection +import Network.AMQP.Utils.Helpers +import Network.AMQP.Utils.Options +import Paths_amqp_utils (version) +import System.Directory +import System.Environment +import System.Exit +import System.INotify +import qualified System.Posix.Files as F main :: IO () main = do @@ -40,7 +42,9 @@ then return False else F.getFileStatus (inputFile args) >>= return . F.isDirectory if isDir - then printparam "hotfolder" $ inputFile args + then + printparam "hotfolder" (inputFile args) >> + printparam "initial scan" (initialScan args) else printparam "input file" [ inputFile args @@ -48,6 +52,7 @@ then "(line-by-line)" else "" ] + printparam "remove sent file" (removeSentFile args && isDir) (conn, chan) <- connect args addChannelExceptionHandler chan (X.throwTo tid) printparam "confirm mode" $ confirm args @@ -56,21 +61,21 @@ confirmSelect chan False addConfirmationListener chan confirmCallback else return () - let publishOneMsg = publishOneMsg' chan args + let publishOneMsg = publishOneMsg' chan args {removeSentFile = removeSentFile args && isDir} X.catch (if isDir then do + setCurrentDirectory (inputFile args) + if (initialScan args) + then getDirectoryContents "." >>= mapM_ (\fn -> handleFile publishOneMsg (suffix args) fn) + else return() inotify <- initINotify wd <- addWatch inotify [CloseWrite, MoveIn] -#if MIN_VERSION_hinotify(0,3,10) - (BS.pack (inputFile args)) -#else - (inputFile args) -#endif - (handleEvent publishOneMsg (suffix args) (inputFile args)) + "." + (handleEvent publishOneMsg (suffix args)) hr $ "BEGIN watching " ++ (inputFile args) _ <- forever $ threadDelay 1000000 removeWatch wd @@ -116,37 +121,36 @@ handleEvent :: (Maybe String -> BL.ByteString -> IO ()) -> [String] - -> String -> Event -> IO () -- just handle closewrite and movedin events #if MIN_VERSION_hinotify(0,3,10) -handleEvent f s p (Closed False (Just x) True) = - handleFile f s (p ++ "/" ++ (BS.unpack x)) -handleEvent f s p (MovedIn False x _) = - handleFile f s (p ++ "/" ++ (BS.unpack x)) +handleEvent func suffixes (Closed False (Just fileName) True) = + handleFile func suffixes (BS.unpack fileName) +handleEvent func suffixes (MovedIn False fileName _) = + handleFile func suffixes (BS.unpack fileName) #else -handleEvent f s p (Closed False (Just x) True) = handleFile f s (p ++ "/" ++ x) -handleEvent f s p (MovedIn False x _) = handleFile f s (p ++ "/" ++ x) +handleEvent func suffixes (Closed False (Just fileName) True) = handleFile func suffixes fileName +handleEvent func suffixes (MovedIn False fileName _) = handleFile func suffixes fileName #endif -handleEvent _ _ _ _ = return () +handleEvent _ _ _ = return () -- | Hotfolder file handler handleFile :: (Maybe String -> BL.ByteString -> IO ()) -> [String] -> FilePath -> IO () handleFile _ _ ('.':_) = return () -- ignore hidden files -handleFile f s@(_:_) x = - if any (flip isSuffixOf x) s - then handleFile f [] x +handleFile func suffixes@(_:_) fileName = + if any (flip isSuffixOf fileName) suffixes + then handleFile func [] fileName else return () -handleFile f [] x = +handleFile func [] fileName = X.catch - (BL.readFile x >>= f (Just x)) + (BL.readFile fileName >>= func (Just fileName)) (\e -> printparam "exception in handleFile" (e :: X.SomeException)) -- | Publish one message with our settings -publishOneMsg' :: Channel -> Args -> Maybe String -> BL.ByteString -> IO () -publishOneMsg' c a fn f = do +publishOneMsg' :: Channel -> Args -> Maybe FilePath -> BL.ByteString -> IO () +publishOneMsg' chan a fn content = do printparam "sending" fn (mtype, mencoding) <- if (magic a) && isJust fn @@ -160,11 +164,11 @@ else return ((contenttype a), (contentencoding a)) now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds publishMsg - c + chan (T.pack $ currentExchange a) (T.pack $ rKey a) newMsg - { msgBody = f + { msgBody = content , msgDeliveryMode = persistent a , msgTimestamp = Just now , msgID = msgid a @@ -181,9 +185,13 @@ , msgHeaders = substheader (fnheader a) fn $ msgheader a } >>= printparam "sent" + removeSentFileIfRequested (removeSentFile a) fn where substheader :: [String] -> Maybe String -> Maybe FieldTable -> Maybe FieldTable substheader (s:r) (Just fname) old = substheader r (Just fname) (addheader old (s ++ "=" ++ fname)) substheader _ _ old = old + removeSentFileIfRequested False _ = return () + removeSentFileIfRequested True Nothing = return () + removeSentFileIfRequested True (Just fname) = printparam "removing" fname >> removeFile fname diff --git a/amqp-utils.cabal b/amqp-utils.cabal --- a/amqp-utils.cabal +++ b/amqp-utils.cabal @@ -1,6 +1,6 @@ name: amqp-utils -version: 0.4.4.1 +version: 0.4.5.0 synopsis: AMQP toolset for the command line diff --git a/arbeite.hs b/arbeite.hs --- a/arbeite.hs +++ b/arbeite.hs @@ -1,22 +1,22 @@ {-# LANGUAGE OverloadedStrings #-} -- generic AMQP rpc server -import Control.Concurrent -import qualified Control.Exception as X -import Control.Monad -import qualified Data.ByteString.Char8 as BS -import Data.Map (singleton) -import Data.Maybe -import qualified Data.Text as T -import Data.Time -import Data.Version (showVersion) -import Network.AMQP -import Network.AMQP.Types -import Network.AMQP.Utils.Connection -import Network.AMQP.Utils.Helpers -import Network.AMQP.Utils.Options -import Paths_amqp_utils (version) -import System.Environment +import Control.Concurrent +import qualified Control.Exception as X +import Control.Monad +import qualified Data.ByteString.Char8 as BS +import Data.Map (singleton) +import Data.Maybe +import qualified Data.Text as T +import Data.Time +import Data.Version (showVersion) +import Network.AMQP +import Network.AMQP.Types +import Network.AMQP.Utils.Connection +import Network.AMQP.Utils.Helpers +import Network.AMQP.Utils.Options +import Paths_amqp_utils (version) +import System.Environment main :: IO () main = do @@ -55,10 +55,7 @@ (rpcServerCallback tid args addiArgs chan) printparam "consumer tag" ctag printparam "send acks" $ ack args - printparam "requeue if rejected" $ - if (ack args) - then Just (requeuenack args) - else Nothing + printparam "requeue if rejected" $ (ack args) && (requeuenack args) hr "entering main loop" X.catch (forever $ threadDelay 5000000) diff --git a/konsum.hs b/konsum.hs --- a/konsum.hs +++ b/konsum.hs @@ -1,16 +1,16 @@ -- generic amqp consumer -import Control.Concurrent -import qualified Control.Exception as X -import Control.Monad -import qualified Data.Text as T -import Data.Time -import Data.Version (showVersion) -import Network.AMQP -import Network.AMQP.Utils.Connection -import Network.AMQP.Utils.Helpers -import Network.AMQP.Utils.Options -import Paths_amqp_utils (version) -import System.Environment +import Control.Concurrent +import qualified Control.Exception as X +import Control.Monad +import qualified Data.Text as T +import Data.Time +import Data.Version (showVersion) +import Network.AMQP +import Network.AMQP.Utils.Connection +import Network.AMQP.Utils.Helpers +import Network.AMQP.Utils.Options +import Paths_amqp_utils (version) +import System.Environment main :: IO () main = do @@ -48,10 +48,7 @@ (myCallback args addiArgs tid) printparam "consumer tag" ctag printparam "send acks" $ ack args - printparam "requeue if rejected" $ - if (ack args) - then Just (show (requeuenack args)) - else Nothing + printparam "requeue if rejected" $ (ack args) && (requeuenack args) hr "entering main loop" X.catch (forever $ threadDelay 5000000) diff --git a/plane.hs b/plane.hs --- a/plane.hs +++ b/plane.hs @@ -1,22 +1,22 @@ {-# LANGUAGE OverloadedStrings #-} -- generic AMQP rpc client -import Control.Concurrent -import qualified Control.Exception as X -import Control.Monad -import qualified Data.ByteString.Lazy.Char8 as BL -import qualified Data.Text as T -import Data.Time -import Data.Time.Clock.POSIX -import Data.Version (showVersion) -import Network.AMQP -import Network.AMQP.Utils.Connection -import Network.AMQP.Utils.Helpers -import Network.AMQP.Utils.Options -import Paths_amqp_utils (version) -import System.Environment -import System.Exit -import System.IO +import Control.Concurrent +import qualified Control.Exception as X +import Control.Monad +import qualified Data.ByteString.Lazy.Char8 as BL +import qualified Data.Text as T +import Data.Time +import Data.Time.Clock.POSIX +import Data.Version (showVersion) +import Network.AMQP +import Network.AMQP.Utils.Connection +import Network.AMQP.Utils.Helpers +import Network.AMQP.Utils.Options +import Paths_amqp_utils (version) +import System.Environment +import System.Exit +import System.IO main :: IO () main = do @@ -79,7 +79,7 @@ exceptionHandler :: RpcException -> IO (ExitCode) exceptionHandler ReceivedException = return ExitSuccess -exceptionHandler TimeoutException = return $ ExitFailure 1 +exceptionHandler TimeoutException = return $ ExitFailure 1 rpcClientCallback :: Handle -> ThreadId -> Args -> (Message, Envelope) -> IO () rpcClientCallback h tid a m@(_, env) = do