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.6.0.0 -- 2020-03-10 + +* update and amend README +* add --arg as alias for --args +* konsum: remove default binding to # +* plane: use -r (--routingkey) instead of -Q (--qname) (but still accept legacy -Q) +* print server properties + ## 0.5.0.0 -- 2020-03-03 * add option -f (--prefetch) for arbeite, default 1 (was unset before) 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 @@ -52,6 +52,8 @@ , coHeartbeatDelay = heartBeat args , coName = fmap T.pack $ connectionName args } + getServerProperties conn >>= return . (formatheaders fieldshow) >>= + printparam "server properties" Just chan <- timeout to $ openChannel conn return (conn, chan) where 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 @@ -106,6 +106,9 @@ fieldshow :: (T.Text, FieldValue) -> String fieldshow (k, v) = "\n " ++ T.unpack k ++ ": " ++ valueshow v +fieldshow' :: (T.Text, FieldValue) -> String +fieldshow' (k, v) = "\n " ++ T.unpack k ++ ": " ++ valueshow v + -- | callback cmdline formatting fieldshowOpt :: (T.Text, FieldValue) -> [String] fieldshowOpt (k, v) = ["-h", T.unpack k ++ "=" ++ valueshow v] @@ -121,14 +124,16 @@ -- | 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 (FVDouble value) = show value -valueshow value = show value +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 (FVDouble value) = show value +valueshow (FVBool value) = show value +valueshow (FVFieldTable value) = (formatheaders fieldshow') value +valueshow value = show value -- | skip showing body head if binary type isimage :: Maybe String -> Bool 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 @@ -135,7 +135,7 @@ (ReqArg (\s o -> o {bindings = (currentExchange o, s) : (bindings o)}) "BINDINGKEY") - ("AMQP binding key (default: #)")) + ("AMQP binding key")) , ( "kr" , Option ['X'] @@ -152,7 +152,7 @@ , ( "kr" , Option ['a'] - ["args"] + ["args", "arg"] (ReqArg (\s o -> o {additionalArgs = s : (additionalArgs o)}) "ARG") "additional argument for -X callback") , ( "kr" @@ -187,6 +187,12 @@ ["routingkey"] (ReqArg (\s o -> o {rKey = s}) "ROUTINGKEY") "AMQP routing key") + , ( "p" + , Option + ['r','Q'] + ["routingkey","qname"] + (ReqArg (\s o -> o {rKey = s}) "ROUTINGKEY") + "AMQP routing key") , ( "ap" , Option ['f'] @@ -362,7 +368,7 @@ ["cleanup"] (NoArg (\o -> o {cleanupTmpFile = not (cleanupTmpFile o)})) "Toggle remove tempfile after script call. Default False, but default True if --simple (-i)") - , ( "krp" + , ( "kr" , Option ['Q'] ["qname"] diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,27 +1,19 @@ # haskell-amqp-utils -generic Haskell AMQP commandline tools for use with RabbitMQ - -## Repository access - -- git://woffs.de/git/fd/haskell-amqp-utils.git -- https://woffs.de/git/fd/haskell-amqp-utils.git -- https://git.sr.ht/~woffs/haskell-amqp-utils +generic AMQP commandline tools for use with RabbitMQ -## Overview -The package contains 4 binaries for commandline use. +- consumer (`konsum`) +- publisher (`agitprop`) +- rpc client and worker (`plane` / `arbeite`) -- konsum, a generic consumer -- agitprop, a generic publisher -- arbeite, an rpc server -- plane, an rpc client +## usage and examples -## konsum -### usage +### konsum +A consumer. konsum [options] Options: - -r BINDINGKEY --bindingkey=BINDINGKEY AMQP binding key (default: #) + -r BINDINGKEY --bindingkey=BINDINGKEY AMQP binding key -X[EXE] --execute[=EXE] Callback Script File (implies -t) (-X without arg: /usr/lib/haskell-amqp-utils/callback) -a ARG --args=ARG additional argument for -X callback -t[DIR] --tempdir[=DIR], --target[=DIR] tempdir (default: no file creation, -t without arg: /tmp) @@ -46,11 +38,10 @@ -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) -### examples +#### examples connect to localhost with default credentials and attach to a new temp -queue on the default exchange (which will usually fail because attaching -a queue to the default exchange is not allowed): +queue on the default exchange: konsum @@ -60,7 +51,7 @@ 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 + 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 binding.key.# -l 500 Authenticate with user and pass. Attach to an existing queue. Generate a file for every message: @@ -70,12 +61,18 @@ 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 -r# +You can have no bindings, or you can bind the queue to one or multiple exchanges + binding keys: + + konsum -x x1 -r b1 -r b2 -x x2 -r b3 + +would bind the queue to exchange x1 with binding keys b1 and b2 and to exchange x2 with binding key b3. + Stop with ^C -## agitprop -### usage +### agitprop +A publisher. agitprop [options] @@ -126,19 +123,20 @@ Line-by-line mode sends one message per INPUTFILE line. -### examples +#### examples Send a message containing a file and put the filename into a fileName message header: agitprop -x amq.topic -r test -F fileName -f agitprop.hs -## plane -### usage +### plane +An RPC client. plane [options] Options: + -r ROUTINGKEY --routingkey=ROUTINGKEY AMQP routing key -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) @@ -146,7 +144,6 @@ --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: /) @@ -160,14 +157,14 @@ -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) -### examples +#### 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 +### arbeite +An RPC worker. arbeite [options] @@ -175,6 +172,7 @@ -X[EXE] --execute[=EXE] Callback Script File (implies -t) (-X without arg: /usr/lib/haskell-amqp-utils/callback) -a ARG --args=ARG additional argument for -X callback -t[DIR] --tempdir[=DIR], --target[=DIR] tempdir (default: no file creation, -t without arg: /tmp) + -f INT --prefetch=INT Prefetch count. (0=unlimited, 1=off, default: 1) -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) @@ -195,9 +193,71 @@ -n NAME --name=NAME connection name, will be shown in RabbitMQ web interface -w SECONDS --connect_timeout=SECONDS timeout for establishing initial connection (default: 600) -### examples +#### 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 +## callbacks + +`konsum` and `arbeite` can execute programs (`-Xprogram`) on reception of a message. + +### normal (old) mode + +The callback is called with the following arguments: + + -f filename + -n messagenumber + +Optionally are added: + + -r routingkey + -m contenttype + -e encoding + -i messageid + -t timestamp + -p priority + -R redelivered + +Message headers are added through several -h options: + + -h HEADER=VALUE + +Any additional arguments provided by `-a arg` options are appended. + +### simple (new) mode + +Wenn called with `-i` (`--simple`), the callback is called with only the filename +as argument. Any additional arguments provided by `-a arg` options are prepended. + +### environment + +The arguments provided in normal (old) mode are also available as environment +variables in both modes in the callback: + + AMQP_FILE + AMQP_NUMBER + + AMQP_ROUTINGKEY + AMQP_CONTENTTYPE + AMQP_ENCODING + AMQP_MSGID + AMQP_TIMESTAMP + AMQP_PRIORITY + AMQP_REDELIVERED + +Message headers, if any, are available as + + AMQP_HEADER_KEY_0 + AMQP_HEADER_VALUE_0 + AMQP_HEADER_KEY_1 + AMQP_HEADER_VALUE_1 + +etc. + +## get the source! + +- git://woffs.de/git/fd/haskell-amqp-utils.git +- https://woffs.de/git/fd/haskell-amqp-utils.git +- https://git.sr.ht/~woffs/haskell-amqp-utils 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.5.0.0 +version: 0.6.0.0 synopsis: AMQP toolset for the command line diff --git a/konsum.hs b/konsum.hs --- a/konsum.hs +++ b/konsum.hs @@ -26,7 +26,7 @@ -- attach to given queue? or build exclusive queue and bind it? queue <- maybe - (tempQueue chan (tmpQName args) (bindings args) (currentExchange args)) + (tempQueue chan (tmpQName args) (bindings args)) (return) (fmap T.pack (qName args)) printparam "queue name" queue @@ -54,8 +54,8 @@ hr "connection closed" -- | exclusive temp queue -tempQueue :: Channel -> String -> [(String, String)] -> String -> IO T.Text -tempQueue chan tmpqname bindlist x = do +tempQueue :: Channel -> String -> [(String, String)] -> IO T.Text +tempQueue chan tmpqname bindlist = do (q, _, _) <- declareQueue chan @@ -64,9 +64,7 @@ (\(xchange, bkey) -> bindQueue chan q (T.pack xchange) (T.pack bkey) >> printparam "binding" [xchange, bkey]) - (if null bindlist - then [(x, "#")] - else bindlist) + bindlist return q -- | process received message diff --git a/plane.hs b/plane.hs --- a/plane.hs +++ b/plane.hs @@ -27,7 +27,7 @@ (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 "routing key" $ rKey args (conn, chan) <- connect args addChannelExceptionHandler chan (X.throwTo tid) (q, _, _) <- declareQueue chan newQueue {queueExclusive = True} @@ -54,7 +54,7 @@ publishMsg chan (T.pack $ currentExchange args) - (T.pack $ tmpQName args) + (T.pack $ rKey args) newMsg { msgBody = message , msgReplyTo = Just q