diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,12 @@ # Revision history for haskell-amqp-utils +## 0.4.1.0 -- 2019-12-09 +* printparam -> Flexprint +* introduce --simple / -i +* review data types +* reformat with hindent +* update doc + ## 0.4.0.1 -- 2019-12-04 * fix exit codes 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 @@ -17,11 +17,11 @@ -- | opens a connection and a channel connect :: Args -> IO (Connection, Channel) connect args = do - printparam' "server" $ server args - printparam' "port" $ show $ port args - printparam' "vhost" $ vHost args + printparam "server" $ server args + printparam "port" $ portnumber args + printparam "vhost" $ vHost args printparam "connection_name" $ connectionName args - printparam' "connect timeout" $ (show (connect_timeout args)) ++ "s" + printparam "connect timeout" $ [show (connect_timeout args), "s"] globalCertificateStore <- getSystemCertificateStore let myTLS = N.TLSSettings @@ -48,8 +48,8 @@ if (tls args) then Just (TLSCustom myTLS) else Nothing - , coServers = [(server args, fromIntegral $ port args)] - , coHeartbeatDelay = fmap fromIntegral $ heartBeat args + , coServers = [(server args, portnumber args)] + , coHeartbeatDelay = heartBeat args , coName = fmap T.pack $ connectionName args } Just chan <- timeout to $ openChannel conn @@ -59,7 +59,7 @@ -- addChannelExceptionHandler chan -- (\exception -> closeConnection conn >> --- printparam' "exiting" (show exception) >> +-- printparam "exiting" (show exception) >> -- killThread tid) -- -- -- noop sharedValidationCache, handy when debugging @@ -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 @@ -1,51 +1,83 @@ +{-# LANGUAGE FlexibleInstances #-} + module Network.AMQP.Utils.Helpers where import Control.Concurrent +import qualified Control.Exception as X import Control.Monad import qualified Data.ByteString.Lazy.Char8 as BL +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.Exit import System.IO import System.Process --- | log cmdline options -listToMaybeUnwords :: [String] -> Maybe String -listToMaybeUnwords [] = Nothing -listToMaybeUnwords x = Just $ unwords x +-- | print config parameters +class (Show a) => + Flexprint a + where + flexprint :: a -> IO () + flexprint = (hPutStrLn stderr) . show + empty :: a -> Bool + empty _ = False + printparam :: String -> a -> IO () + printparam label x = + if empty x + then return () + else do + mapM_ (hPutStr stderr) [" --- ", label, ": "] + flexprint x + hFlush stderr --- | Strings or ByteStrings with label, oder nothing at all -printwithlabel :: String -> Maybe (IO ()) -> IO () -printwithlabel _ Nothing = return () -printwithlabel labl (Just i) = do - mapM_ (hPutStr stderr) [" --- ", labl, ": "] - i - hFlush stderr +instance (Flexprint a) => Flexprint (Maybe a) where + empty = isNothing + printparam _ Nothing = return () + printparam x (Just y) = printparam x y --- | optional parameters -printparam :: String -> Maybe String -> IO () -printparam labl ms = printwithlabel labl $ fmap (hPutStrLn stderr) ms +instance Flexprint String where + flexprint = hPutStrLn stderr + empty = null --- | required parameters -printparam' :: String -> String -> IO () -printparam' d s = printparam d (Just s) +instance Flexprint [String] where + flexprint = flexprint . unwords + empty = null --- | head chars of body -printbody :: String -> Maybe BL.ByteString -> IO () -printbody labl ms = do - printwithlabel labl $ - fmap - (\s -> hPutStrLn stderr "" >> BL.hPut stderr s >> hPutStrLn stderr "") - ms - hFlush stderr +instance Flexprint T.Text where + flexprint = flexprint . T.unpack + empty = T.null +instance Flexprint BL.ByteString where + flexprint x = hPutStrLn stderr "" >> BL.hPut stderr x >> hPutStrLn stderr "" + empty = BL.null + +instance Flexprint Bool + +instance Flexprint Int + +instance Flexprint Int64 + +instance Flexprint Word16 + +instance Flexprint ExitCode + +instance Flexprint X.SomeException + +instance Flexprint AMQPException + +instance Flexprint ConfirmationResult + +instance Flexprint PortNumber + -- | log marker hr :: String -> IO () hr x = hPutStrLn stderr hr' >> hFlush stderr @@ -79,11 +111,11 @@ | otherwise = any (flip isPrefixOf ctype) ["application", "image"] -- | show the first bytes of message body -anriss' :: Maybe Int -> BL.ByteString -> BL.ByteString +anriss' :: Maybe Int64 -> BL.ByteString -> BL.ByteString anriss' x = case x of Nothing -> id - Just y -> BL.take (fromIntegral y) + Just y -> BL.take y -- | callback cmdline with optional parameters printopt :: (String, Maybe String) -> [String] @@ -91,7 +123,12 @@ printopt (opt, Just s) = [opt, s] -- | prints header and head on stderr and returns cmdline options to callback -printmsg :: Maybe Handle -> (Message, Envelope) -> Maybe Int -> ZonedTime -> IO [String] +printmsg :: + Maybe Handle + -> (Message, Envelope) + -> Maybe Int64 + -> ZonedTime + -> IO [String] printmsg h (msg, envi) anR now = do mapM_ (uncurry printparam) @@ -114,7 +151,7 @@ , ("expiration", mexp) , ("delivery mode", mdelivmode) ] - printbody label anriss + printparam label anriss mapM_ (\hdl -> BL.hPut hdl body >> hFlush hdl) h return $ concat @@ -209,10 +246,10 @@ printparam "saved to" path let callbackcmdline = liftM2 - (constructCallbackCmdLine callbackoptions addi numstring) + (constructCallbackCmdLine (simple a) callbackoptions addi numstring) (fileProcess a) path - printparam "calling" $ fmap unwords callbackcmdline + printparam "calling" callbackcmdline maybe (acke envi a) (\c -> @@ -236,8 +273,9 @@ -- | construct cmdline for callback script constructCallbackCmdLine :: - [String] -> [String] -> String -> String -> String -> [String] -constructCallbackCmdLine opts addi num exe path = + Bool -> [String] -> [String] -> String -> String -> String -> [String] +constructCallbackCmdLine True _ addi _ exe path = exe : addi ++ path : [] +constructCallbackCmdLine False opts addi num exe path = exe : "-f" : path : "-n" : num : opts ++ addi -- | call callback script @@ -252,8 +290,9 @@ (_, h, _, processhandle) <- createProcess (proc exe args) {std_out = out, std_err = Inherit} sout <- mapM BL.hGetContents h - exitcode <- maybe 0 id (fmap BL.length sout) `seq` waitForProcess processhandle - printparam' (numstring ++ " call returned") $ show exitcode + exitcode <- + maybe 0 id (fmap BL.length sout) `seq` waitForProcess processhandle + printparam (numstring ++ " call returned") exitcode if isJust action && isJust sout then ((fromJust action $ exitcode) (fromJust sout)) >> acke envi a else case exitcode of 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,68 +1,79 @@ module Network.AMQP.Utils.Options where 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 + | (port a) == Nothing && (tls a) = 5671 + | (port a) == Nothing = 5672 + | otherwise = fromJust (port a) + -- | A data type for our options -data Args = Args - { server :: String - , port :: Int - , tls :: Bool - , vHost :: String - , currentExchange :: String - , bindings :: [(String, String)] - , rKey :: String - , anRiss :: Maybe Int - , fileProcess :: Maybe String - , qName :: Maybe String - , cert :: Maybe String - , key :: Maybe String - , user :: String - , pass :: String - , preFetch :: Int - , heartBeat :: Maybe Int - , 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 - , connect_timeout :: Int - } +data Args = + Args + { 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 + , 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 + , connect_timeout :: Int + , simple :: Bool + } instance Default Args where def = Args "localhost" - 5672 + Nothing False "/" "" @@ -105,6 +116,7 @@ True 5 60 + False -- | all options allOptions :: [(String, OptDescr (Args -> Args))] @@ -245,7 +257,8 @@ ['t'] ["rpc_timeout"] (ReqArg (\s o -> o {rpc_timeout = read s}) "SECONDS") - ("How long to wait for reply (default: " ++ show (rpc_timeout def) ++ ")")) + ("How long to wait for reply (default: " ++ + show (rpc_timeout def) ++ ")")) , ( "a" , Option [] @@ -307,7 +320,7 @@ , Option ['l'] ["charlimit"] - (ReqArg (\s o -> o {anRiss = Just (read s :: Int)}) "INT") + (ReqArg (\s o -> o {anRiss = Just (read s)}) "INT") "limit number of shown body chars (default: unlimited)") , ( "kr" , Option @@ -315,6 +328,12 @@ ["queue"] (ReqArg (\s o -> o {qName = Just s}) "QUEUENAME") "Ignore Exchange and bind to existing Queue") + , ( "kr" + , Option + ['i'] + ["simple"] + (NoArg (\o -> o {simple = True})) + "call callback with one arg (filename) only") , ( "krp" , Option ['Q'] @@ -343,8 +362,8 @@ , Option ['p'] ["port"] - (ReqArg (\s o -> o {port = read s}) "PORT") - ("Server Port Number (default: " ++ show (port def) ++ ")")) + (ReqArg (\s o -> o {port = Just (read s)}) "PORT") + ("Server Port Number (default: " ++ show (portnumber def) ++ ")")) , ( "akrp" , Option ['T'] @@ -392,7 +411,8 @@ ['w'] ["connect_timeout"] (ReqArg (\s o -> o {connect_timeout = read s}) "SECONDS") - ("timeout for establishing initial connection (default: " ++ show (connect_timeout def) ++ ")")) + ("timeout for establishing initial connection (default: " ++ + show (connect_timeout def) ++ ")")) ] -- | Options for the executables @@ -401,7 +421,8 @@ -- | Add a header with a String value addheader :: Maybe FieldTable -> String -> Maybe FieldTable -addheader Nothing string = Just $ FieldTable $ M.singleton (getkey string) (getval string) +addheader Nothing string = + Just $ FieldTable $ M.singleton (getkey string) (getval string) addheader (Just (FieldTable oldheader)) string = Just $ FieldTable $ M.insert (getkey string) (getval string) oldheader diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -23,6 +23,9 @@ -A --ack Toggle ack messages (default: True) -R --requeuenack Toggle requeue when rejected (default: True) -l INT --charlimit=INT limit number of shown body chars (default: unlimited) + -q QUEUENAME --queue=QUEUENAME Ignore Exchange and bind to existing Queue + -i --simple call callback with one arg (filename) only + -Q TEMPQNAME --qname=TEMPQNAME Name for temporary exclusive Queue -x EXCHANGE --exchange=EXCHANGE AMQP Exchange (default: "") -o SERVER --server=SERVER AMQP Server (default: localhost) -y VHOST --vhost=VHOST AMQP Virtual Host (default: /) @@ -34,6 +37,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) ### examples @@ -46,19 +50,20 @@ Connect to a host with TLS on a custom port, authenticating with SSL client certificate. On every received message a callback is spawned. The message will be ACKed when the callback exits successfully. First -500 bytes of the message body are printed to stdout. Header infos are -always printed to stdout: +500 bytes of the message body are printed to stderr. Header infos are +always printed to stderr: konsum -o amqp.example.com -p 5673 -T -k amqp-key.pem -c amqp-crt.pem -y vhost -x exchange -X./callback.sh -a -c -a callback.config.sh -f 2 -r routing.key.# -l 500 -Authenticate with user and pass. Generate a file for every message: +Authenticate with user and pass. Attach to an existing queue. Generate a file +for every message: konsum -o amqp.example.com -U user -P pass -q queue -t Provide a custom CA cert for proving the server's identity via enviroment: - $ env SYSTEM_CERTIFICATE_PATH=/etc/amqp/cacert.crt ./konsum -T -y vhost -x exchange + $ env SYSTEM_CERTIFICATE_PATH=/etc/amqp/cacert.crt konsum -T -y vhost -x exchange Stop with ^C @@ -69,7 +74,7 @@ Options: -r ROUTINGKEY --routingkey=ROUTINGKEY AMQP routing key - -f INPUTFILE --inputfile=INPUTFILE Message input file (default: /dev/stdin) + -f INPUTFILE --inputfile=INPUTFILE Message input file (default: -) -l --linemode Toggle line-by-line mode (default: False) -C --confirm Toggle confirms (default: False) --msgid=ID Message ID @@ -89,8 +94,6 @@ -m --magic Toggle setting content-type and -encoding from file contents (default: False) -e --persistent Set persistent delivery -E --nonpersistent Set nonpersistent delivery - -q QUEUENAME --queue=QUEUENAME Ignore Exchange and bind to existing Queue - -Q TEMPQNAME --qname=TEMPQNAME Name for temporary exclusive Queue -x EXCHANGE --exchange=EXCHANGE AMQP Exchange (default: "") -o SERVER --server=SERVER AMQP Server (default: localhost) -y VHOST --vhost=VHOST AMQP Virtual Host (default: /) @@ -102,6 +105,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) 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 @@ -126,23 +130,33 @@ plane [options] Options: - -t SECONDS --timeout=SECONDS How long to wait for reply (default: 5.0) - --corrid=CORRID Message CorrelationID - --exp=EXP Message Expiration - -l INT --charlimit=INT limit number of shown body chars (default: unlimited) - -Q TEMPQNAME --qname=TEMPQNAME Name for temporary exclusive Queue - -x EXCHANGE --exchange=EXCHANGE AMQP Exchange (default: "") - -o SERVER --server=SERVER AMQP Server (default: localhost) - -y VHOST --vhost=VHOST AMQP Virtual Host (default: /) - -p PORT --port=PORT Server Port Number (default: 5672) - -T --tls Toggle TLS (default: False) - -c CERTFILE --cert=CERTFILE TLS Client Certificate File - -k KEYFILE --key=KEYFILE TLS Client Private Key File - -U USERNAME --user=USERNAME Username for Auth - -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 + -f INPUTFILE --inputfile=INPUTFILE Message input file (default: -) + -O OUTPUTFILE --outputfile=OUTPUTFILE Message output file (default: -) + -t SECONDS --rpc_timeout=SECONDS How long to wait for reply (default: 5.0) + --corrid=CORRID Message CorrelationID + --exp=EXP Message Expiration + -h HEADER=VALUE --header=HEADER=VALUE Message Headers + -l INT --charlimit=INT limit number of shown body chars (default: unlimited) + -Q TEMPQNAME --qname=TEMPQNAME Name for temporary exclusive Queue + -x EXCHANGE --exchange=EXCHANGE AMQP Exchange (default: "") + -o SERVER --server=SERVER AMQP Server (default: localhost) + -y VHOST --vhost=VHOST AMQP Virtual Host (default: /) + -p PORT --port=PORT Server Port Number (default: 5672) + -T --tls Toggle TLS (default: False) + -c CERTFILE --cert=CERTFILE TLS Client Certificate File + -k KEYFILE --key=KEYFILE TLS Client Private Key File + -U USERNAME --user=USERNAME Username for Auth + -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) +### examples + +send "ls" to a remote worker and get the result: + + echo ls | plane -o amqp.example.com -T -k amqp.pem -c amqp.pem -y myexchange -Q rpctest + ## arbeite ### usage @@ -156,7 +170,9 @@ -R --requeuenack Toggle requeue when rejected (default: True) -l INT --charlimit=INT limit number of shown body chars (default: unlimited) -q QUEUENAME --queue=QUEUENAME Ignore Exchange and bind to existing Queue + -i --simple call callback with one arg (filename) only -Q TEMPQNAME --qname=TEMPQNAME Name for temporary exclusive Queue + -x EXCHANGE --exchange=EXCHANGE AMQP Exchange (default: "") -o SERVER --server=SERVER AMQP Server (default: localhost) -y VHOST --vhost=VHOST AMQP Virtual Host (default: /) -p PORT --port=PORT Server Port Number (default: 5672) @@ -167,3 +183,11 @@ -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) + +### examples + +provide shell access to a remote user. Very insecure! :-) + + arbeite -o amqp.example.com -T -k amqp.pem -c amqp.pem -y myexchange -Q rpctest -i -Xsh + diff --git a/agitprop.hs b/agitprop.hs --- a/agitprop.hs +++ b/agitprop.hs @@ -1,4 +1,5 @@ {-# LANGUAGE CPP #-} + -- generic AMQP publisher import Control.Concurrent import qualified Control.Exception as X @@ -31,23 +32,25 @@ hr "starting" tid <- myThreadId args <- getArgs >>= parseargs 'a' - printparam' "client version" $ "amqp-utils " ++ (showVersion version) - printparam' "routing key" $ rKey args - printparam' "exchange" $ currentExchange args + printparam "client version" ["amqp-utils", showVersion version] + printparam "routing key" $ rKey args + printparam "exchange" $ currentExchange args isDir <- if inputFile args == "-" then return False else F.getFileStatus (inputFile args) >>= return . F.isDirectory if isDir - then printparam' "hotfolder" $ inputFile args - else printparam' "input file" $ - (inputFile args) ++ - if (lineMode args) - then " (line-by-line)" - else "" + then printparam "hotfolder" $ inputFile args + else printparam + "input file" + [ inputFile args + , if (lineMode args) + then "(line-by-line)" + else "" + ] (conn, chan) <- connect args addChannelExceptionHandler chan (X.throwTo tid) - printparam' "confirm mode" $ show $ confirm args + printparam "confirm mode" $ confirm args if (confirm args) then do confirmSelect chan False @@ -85,26 +88,29 @@ exceptionHandler -- all done. wait and close. if (confirm args) - then waitForConfirms chan >>= (printparam' "confirmed") . show + then waitForConfirms chan >>= printparam "confirmed" else return () X.catch (closeConnection conn) exceptionHandler -- | A handler for clean exit exceptionHandler :: AMQPException -> IO () -exceptionHandler (ChannelClosedException Normal txt) = printparam' "exit" txt >> exitWith ExitSuccess -exceptionHandler (ConnectionClosedException Normal txt) = printparam' "exit" txt >> exitWith ExitSuccess -exceptionHandler x = printparam' "exception" (show x) >> exitWith (ExitFailure 1) +exceptionHandler (ChannelClosedException Normal txt) = + printparam "exit" txt >> exitWith ExitSuccess +exceptionHandler (ConnectionClosedException Normal txt) = + printparam "exit" txt >> exitWith ExitSuccess +exceptionHandler x = printparam "exception" x >> exitWith (ExitFailure 1) -- | The handler for publisher confirms confirmCallback :: (Word64, Bool, AckType) -> IO () confirmCallback (deliveryTag, isAll, ackType) = - printparam' + printparam "confirmed" - ((show deliveryTag) ++ - (if isAll - then " all " - else " this ") ++ - (show ackType)) + [ show deliveryTag + , if isAll + then "all" + else "this" + , show ackType + ] -- | Hotfolder event handler handleEvent :: @@ -115,8 +121,10 @@ -> 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 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)) #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) @@ -134,9 +142,7 @@ handleFile f [] x = X.catch (BL.readFile x >>= f (Just x)) - (\exception -> - printparam' "exception in handleFile" $ - show (exception :: X.SomeException)) + (\e -> printparam "exception in handleFile" (e :: X.SomeException)) -- | Publish one message with our settings publishOneMsg' :: Channel -> Args -> Maybe String -> BL.ByteString -> IO () @@ -153,29 +159,28 @@ return (Just (T.pack t), Just (T.pack e)) else return ((contenttype a), (contentencoding a)) now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds - r <- - publishMsg - c - (T.pack $ currentExchange a) - (T.pack $ rKey a) - newMsg - { msgBody = f - , msgDeliveryMode = persistent a - , msgTimestamp = Just now - , msgID = msgid a - , msgType = msgtype a - , msgUserID = userid a - , msgApplicationID = appid a - , msgClusterID = clusterid a - , msgContentType = mtype - , msgContentEncoding = mencoding - , msgReplyTo = replyto a - , msgPriority = prio a - , msgCorrelationID = corrid a - , msgExpiration = msgexp a - , msgHeaders = substheader (fnheader a) fn $ msgheader a - } - printparam "sent" $ fmap show r + publishMsg + c + (T.pack $ currentExchange a) + (T.pack $ rKey a) + newMsg + { msgBody = f + , msgDeliveryMode = persistent a + , msgTimestamp = Just now + , msgID = msgid a + , msgType = msgtype a + , msgUserID = userid a + , msgApplicationID = appid a + , msgClusterID = clusterid a + , msgContentType = mtype + , msgContentEncoding = mencoding + , msgReplyTo = replyto a + , msgPriority = prio a + , msgCorrelationID = corrid a + , msgExpiration = msgexp a + , msgHeaders = substheader (fnheader a) fn $ msgheader a + } >>= + printparam "sent" where substheader :: [String] -> Maybe String -> Maybe FieldTable -> Maybe FieldTable 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.0.1 +version: 0.4.1.0 synopsis: Generic Haskell AMQP tools @@ -44,12 +44,13 @@ process, bytestring, x509-system, + network > 2.6, tls >= 1.3.9, amqp >=0.17 ghc-options: -threaded -Wall - default-language: Haskell98 + default-language: Haskell2010 other-modules: Network.AMQP.Utils.Options, Network.AMQP.Utils.Helpers, @@ -67,6 +68,7 @@ process, bytestring, x509-system, + network > 2.6, tls >= 1.3.9, amqp >=0.17, unix >= 2.7, @@ -75,7 +77,7 @@ ghc-options: -threaded -Wall - default-language: Haskell98 + default-language: Haskell2010 other-modules: Network.AMQP.Utils.Options, Network.AMQP.Utils.Helpers, @@ -93,13 +95,14 @@ process, bytestring, x509-system, + network > 2.6, tls >= 1.3.9, amqp >=0.17, unix >= 2.7 ghc-options: -threaded -Wall - default-language: Haskell98 + default-language: Haskell2010 other-modules: Network.AMQP.Utils.Options, Network.AMQP.Utils.Helpers, @@ -117,13 +120,14 @@ process, bytestring, x509-system, + network > 2.6, tls >= 1.3.9, amqp >=0.17, unix >= 2.7 ghc-options: -threaded -Wall - default-language: Haskell98 + default-language: Haskell2010 other-modules: Network.AMQP.Utils.Options, Network.AMQP.Utils.Helpers, diff --git a/arbeite.hs b/arbeite.hs --- a/arbeite.hs +++ b/arbeite.hs @@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings #-} + -- generic AMQP rpc server import Control.Concurrent import qualified Control.Exception as X @@ -22,10 +23,10 @@ tid <- myThreadId args <- getArgs >>= parseargs 'r' X.onException - (printparam' "worker" $ fromJust $ fileProcess args) + (printparam "worker" $ fromJust $ fileProcess args) (error "-X option required") let addiArgs = reverse $ additionalArgs args - printparam' "client version" $ "amqp-utils " ++ (showVersion version) + printparam "client version" ["amqp-utils", showVersion version] (conn, chan) <- connect args addChannelExceptionHandler chan (X.throwTo tid) queue <- @@ -36,10 +37,10 @@ (\(x, _, _) -> return x)) (return) (fmap T.pack (qName args)) - printparam' "queue name" $ T.unpack queue + printparam "queue name" queue if (currentExchange args /= "") then do - printparam' "exchange" $ currentExchange args + printparam "exchange" $ currentExchange args bindQueue chan queue (T.pack $ currentExchange args) queue else return () ctag <- @@ -50,16 +51,16 @@ then Ack else NoAck) (rpcServerCallback tid args addiArgs chan) - printparam' "consumer tag" $ T.unpack ctag - printparam' "send acks" $ show (ack args) + printparam "consumer tag" ctag + printparam "send acks" $ ack args printparam "requeue if rejected" $ if (ack args) - then Just (show (requeuenack args)) + then Just (requeuenack args) else Nothing hr "entering main loop" X.catch (forever $ threadDelay 5000000) - (\exception -> printparam' "exception" $ show (exception :: X.SomeException)) + (\e -> printparam "exception" (e :: X.SomeException)) closeConnection conn hr "connection closed" @@ -73,7 +74,7 @@ X.catch (printmsg Nothing m (anRiss a) now) (\x -> X.throwTo tid (x :: X.SomeException) >> return []) - either (\e -> printparam' "ERROR" (show (e :: X.SomeException))) return =<< + either (\e -> printparam "ERROR" (e :: X.SomeException)) return =<< X.try (optionalFileStuff m callbackoptions addi numstring a tid (Just reply)) hr $ "END " ++ numstring @@ -89,5 +90,7 @@ , msgCorrelationID = msgCorrelationID msg , msgTimestamp = msgTimestamp msg , msgExpiration = msgExpiration msg - , msgHeaders = Just $ FieldTable $ singleton "exitcode" $ FVString $ T.pack $ show e + , msgHeaders = + Just $ + FieldTable $ singleton "exitcode" $ FVString $ T.pack $ show e } diff --git a/konsum.hs b/konsum.hs --- a/konsum.hs +++ b/konsum.hs @@ -18,23 +18,23 @@ tid <- myThreadId args <- getArgs >>= parseargs 'k' let addiArgs = reverse $ additionalArgs args - printparam' "client version" $ "amqp-utils " ++ (showVersion version) + printparam "client version" ["amqp-utils", showVersion version] (conn, chan) <- connect args addChannelExceptionHandler chan (X.throwTo tid) -- set prefetch - printparam' "prefetch" $ show $ preFetch args - qos chan 0 (fromIntegral $ preFetch args) False + printparam "prefetch" $ preFetch args + qos chan 0 (preFetch args) False -- attach to given queue? or build exclusive queue and bind it? queue <- maybe (tempQueue chan (tmpQName args) (bindings args) (currentExchange args)) (return) (fmap T.pack (qName args)) - printparam' "queue name" $ T.unpack queue - printparam "shown body chars" $ fmap show $ anRiss args + printparam "queue name" queue + printparam "shown body chars" $ anRiss args printparam "temp dir" $ tempDir args printparam "callback" $ fileProcess args - printparam "callback args" $ listToMaybeUnwords addiArgs + printparam "callback args" $ addiArgs -- subscribe to the queue ctag <- consumeMsgs @@ -44,8 +44,8 @@ then Ack else NoAck) (myCallback args addiArgs tid) - printparam' "consumer tag" $ T.unpack ctag - printparam' "send acks" $ show (ack args) + printparam "consumer tag" ctag + printparam "send acks" $ ack args printparam "requeue if rejected" $ if (ack args) then Just (show (requeuenack args)) @@ -53,7 +53,7 @@ hr "entering main loop" X.catch (forever $ threadDelay 5000000) - (\exception -> printparam' "exception" $ show (exception :: X.SomeException)) + (\e -> printparam "exception" (e :: X.SomeException)) closeConnection conn hr "connection closed" @@ -67,7 +67,7 @@ mapM_ (\(xchange, bkey) -> bindQueue chan q (T.pack xchange) (T.pack bkey) >> - printparam' "binding" (xchange ++ ":" ++ bkey)) + printparam "binding" [xchange, bkey]) (if null bindlist then [(x, "#")] else bindlist) @@ -83,8 +83,6 @@ X.catch (printmsg Nothing m (anRiss a) now) (\x -> X.throwTo tid (x :: X.SomeException) >> return []) - either - (\e -> printparam' "ERROR" (show (e :: X.SomeException)) >> reje envi a) - return =<< + either (\e -> printparam "ERROR" (e :: X.SomeException) >> reje envi a) return =<< X.try (optionalFileStuff m callbackoptions addi numstring a tid Nothing) hr $ "END " ++ numstring diff --git a/plane.hs b/plane.hs --- a/plane.hs +++ b/plane.hs @@ -24,52 +24,57 @@ tid <- myThreadId args <- getArgs >>= parseargs 'p' X.onException - (printparam' "rpc_timeout" $ show (rpc_timeout args) ++ "s") + (printparam "rpc_timeout" [show (rpc_timeout args), "s"]) (error $ "invalid rpc_timeout") - printparam' "client version" $ "amqp-utils " ++ (showVersion version) - printparam' "destination queue" $ tmpQName args + printparam "client version" ["amqp-utils", showVersion version] + printparam "destination queue" $ tmpQName args (conn, chan) <- connect args addChannelExceptionHandler chan (X.throwTo tid) (q, _, _) <- declareQueue chan newQueue {queueExclusive = True} if (currentExchange args /= "") then do - printparam' "exchange" $ currentExchange args + printparam "exchange" $ currentExchange args bindQueue chan q (T.pack $ currentExchange args) q else return () - printparam' "input file" $ inputFile args + printparam "input file" $ inputFile args message <- if inputFile args == "-" then BL.getContents else BL.readFile (inputFile args) - printparam' "output file" $ outputFile args - h <- if outputFile args == "-" then return stdout else openBinaryFile (outputFile args) WriteMode + printparam "output file" $ outputFile args + h <- + if outputFile args == "-" + then return stdout + else openBinaryFile (outputFile args) WriteMode ctag <- consumeMsgs chan q NoAck (rpcClientCallback h tid args) - printparam' "consumer tag" $ T.unpack ctag + printparam "consumer tag" ctag now <- getCurrentTime >>= return . floor . utcTimeToPOSIXSeconds hr "publishing request" - _ <- publishMsg - chan - (T.pack $ currentExchange args) - (T.pack $ tmpQName args) - newMsg - { msgBody = message - , msgReplyTo = Just q - , msgCorrelationID = corrid args - , msgExpiration = msgexp args - , msgTimestamp = Just now - , msgHeaders = msgheader args - } + _ <- + publishMsg + chan + (T.pack $ currentExchange args) + (T.pack $ tmpQName args) + newMsg + { msgBody = message + , msgReplyTo = Just q + , msgCorrelationID = corrid args + , msgExpiration = msgexp args + , msgTimestamp = Just now + , msgHeaders = msgheader args + } hr "waiting for answer" - _ <- forkIO - (threadDelay (floor (1000000 * rpc_timeout args)) >> - throwTo tid TimeoutException) + _ <- + forkIO + (threadDelay (floor (1000000 * rpc_timeout args)) >> + throwTo tid TimeoutException) X.catch (forever $ threadDelay 200000) (\x -> do ec <- exceptionHandler x hr "closing connection" closeConnection conn - printparam' "exiting" $ show ec + printparam "exiting" ec exitWith ec) exceptionHandler :: RpcException -> IO (ExitCode)