amqp-utils 0.4.0.1 → 0.4.1.0
raw patch · 10 files changed
+325/−219 lines, 10 filesdep +network
Dependencies added: network
Files
- ChangeLog.md +7/−0
- Network/AMQP/Utils/Connection.hs +8/−8
- Network/AMQP/Utils/Helpers.hs +74/−35
- Network/AMQP/Utils/Options.hs +74/−53
- README.md +47/−23
- agitprop.hs +53/−48
- amqp-utils.cabal +9/−5
- arbeite.hs +13/−10
- konsum.hs +11/−13
- plane.hs +29/−24
ChangeLog.md view
@@ -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
Network/AMQP/Utils/Connection.hs view
@@ -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
Network/AMQP/Utils/Helpers.hs view
@@ -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
Network/AMQP/Utils/Options.hs view
@@ -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
README.md view
@@ -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+
agitprop.hs view
@@ -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
amqp-utils.cabal view
@@ -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,
arbeite.hs view
@@ -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 }
konsum.hs view
@@ -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
plane.hs view
@@ -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)