amqp-utils 0.5.0.0 → 0.6.0.0
raw patch · 8 files changed
+130/−51 lines, 8 files
Files
- ChangeLog.md +8/−0
- Network/AMQP/Utils/Connection.hs +2/−0
- Network/AMQP/Utils/Helpers.hs +13/−8
- Network/AMQP/Utils/Options.hs +9/−3
- README.md +91/−31
- amqp-utils.cabal +1/−1
- konsum.hs +4/−6
- plane.hs +2/−2
ChangeLog.md view
@@ -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)
Network/AMQP/Utils/Connection.hs view
@@ -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
Network/AMQP/Utils/Helpers.hs view
@@ -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
Network/AMQP/Utils/Options.hs view
@@ -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"]
README.md view
@@ -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
amqp-utils.cabal view
@@ -1,6 +1,6 @@ name: amqp-utils -version: 0.5.0.0+version: 0.6.0.0 synopsis: AMQP toolset for the command line
konsum.hs view
@@ -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
plane.hs view
@@ -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