amqp-utils
AMQP toolset for the command line
Version on this page: | 0.6.3.2 |
LTS Haskell 22.43: | 0.6.4.0 |
Stackage Nightly 2023-12-26: | 0.6.4.0 |
Latest on Hackage: | 0.6.6.0 |
amqp-utils-0.6.3.2@sha256:eb15b14c5f1b4f14eb207efb63e84d245d46a4c7b68924df441331f164b8ed93,4695
Module documentation for 0.6.3.2
There are no documented modules for this package.
haskell-amqp-utils
generic AMQP commandline tools for use with RabbitMQ
- consumer (
konsum
) - publisher (
agitprop
) - rpc client and worker (
plane
/arbeite
)
usage and examples
konsum
A consumer.
konsum [options]
Options:
-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, --arg=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)
-q QUEUENAME --queue=QUEUENAME Ignore Exchange and bind to existing Queue
-i --simple call callback with one arg (filename) only
-j --cleanup Toggle remove tempfile after script call. Default False, but default True if --simple (-i)
-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: 600)
examples
connect to localhost with default credentials and attach to a new temp queue on the default exchange:
konsum
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 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 binding.key.# -l 500
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 -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
A publisher.
agitprop [options]
Options:
-r ROUTINGKEY --routingkey=ROUTINGKEY AMQP routing key
-f INPUTFILE --inputfile=INPUTFILE Message input file (default: <stdin>)
-l --linemode Toggle line-by-line mode (default: False)
-C --confirm Toggle confirms (default: False)
--msgid=ID Message ID
--type=TYPE Message Type
--userid=USERID Message User-ID
--appid=APPID Message App-ID
--clusterid=CLUSTERID Message Cluster-ID
--contenttype=CONTENTTYPE Message Content-Type
--contentencoding=CONTENTENCODING Message Content-Encoding
--replyto=REPLYTO Message Reply-To
--prio=PRIO Message Priority
--corrid=CORRID Message CorrelationID
--exp=EXP Message Expiration
-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[DIR] --remove[=DIR], --move[=DIR] Remove (or move to DIR) sent file in hotfolder mode
-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
-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: 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 (“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.
In “hotfolder mode” multiple directories can be watched and sent with distinct routingkeys.
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
Watch two directories dir1 and dir2 and send files from dir1 with routingkey r1 and files from dir2 with routingkey r2, ignoring everything which is not *.txt or *.zip, and set Content-Type and Content-Encoding automatically, and move all sent files to subdir “done” (dir1/done and dir2/done, respectively):
agitprop -x amq.topic -F fileName -m -S .txt -S .zip -d -udone -r r1 -f dir1 -r r2 -f dir2
plane
An RPC client.
plane [options]
Options:
-r ROUTINGKEY, -Q ROUTINGKEY --routingkey=ROUTINGKEY, --qname=ROUTINGKEY AMQP routing key
-f INPUTFILE --inputfile=INPUTFILE Message input file (default: <stdin>)
-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)
-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: 600)
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
An RPC worker.
arbeite [options]
Options:
-X[EXE] --execute[=EXE] Callback Script File (implies -t) (-X without arg: /usr/lib/haskell-amqp-utils/callback)
-a ARG --args=ARG, --arg=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)
-q QUEUENAME --queue=QUEUENAME Ignore Exchange and bind to existing Queue
-i --simple call callback with one arg (filename) only
-j --cleanup Toggle remove tempfile after script call. Default False, but default True if --simple (-i)
-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: 600)
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
And the remaining message properties, too:
AMQP_SIZE
AMQP_TYPE
AMQP_USERID
AMQP_APPID
AMQP_CLUSTERID
AMQP_REPLYTO
AMQP_CORRID
AMQP_EXPIRATION
AMQP_DELIVERYMODE
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 it!
- 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
- https://repology.org/project/haskell:amqp-utils/packages
Changes
Revision history for haskell-amqp-utils
0.6.3.2 – 2021-10-25
- update README
0.6.3.1 – 2021-09-08
- avoid setCurrentDirectory which is not thread-safe
- start watch before initial scan
- put filename without directory into filename header
0.6.3.0 – 2021-09-06
- watch multiple dirs in hotfolder mode
- drop hinotify < 0.3.10 compatibility
- use RawFilePath for file names
- fix handling of non-ascii filenames in agitprop hotfolder
0.6.2.4 – 2021-08-30
- depend on tls >= 1.5.0 (thread safety for writes)
0.6.2.2 – 2021-08-25
- agitprop: option to move sent file to another dir instead of just removing
0.6.1.1 – 2021-05-11
- disable inotify on non-linux
0.6.1.0 – 2020-03-11
- add remaining message properties as callback environment variables, too
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)
- print a line after (not) receiving rpc answer
- catch IOException instead of SomeException in callback
- remove –enable-tests from cabal in Makefile
- reformat with hindent + stylish-haskell
0.4.5.1 – 2020-02-19
- unify threadDelay in mainloop (reduce system load when sleeping)
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
- update description
0.4.4.0 – 2020-02-18
- push callback options into environment variables
0.4.3.0 – 2020-02-10
- specify -R YES option for callback in case of a redelivered message
0.4.2.0 – 2019-12-19
- introduce –cleanup (-j) to remove temp file
- change temp file name from konsum-* to amqp-utils-*
- handle all numeric types in amqp headers same
- update to amqp-0.19
- use utf8-string for header string values
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
0.4.0.0 – 2019-11-11
- agitprop: show exchange arg
- agitprop does not need -q or -Q
- allow RPC with dedicated exchange
- add missing hFlush
- stdin / stdout handling without using /dev/
- cleanup plane answer file handling
- duplicate plane body to outfile and stderr
0.3.7.1 – 2019-09-10
- connect timeout
0.3.6.0 – 2019-02-15
- bug-fix: avoid deadlock in arbeite
- plane: add –header option
- rpc: log to stderr, result to stdout
0.3.4.0 – 2018-07-21
- bug fix: re-add exception handler
- plane + arbeite: rpc client + server
0.3.3.1 – 2018-07-10
- fix debian builds
- konsum options ack and requeuenack
- fix hotfolder mode
- enable parallel build
- tls, hinotify, lts-12.0 compat
0.3.2.0 – 2018-07-04
- agitprop, a publisher
- optional publisher confirms
- hotfolder mode, file magic
- several options
0.3.0.2 – 2018-04-24
- use ciphersuite_default
0.3.0.1 – 2018-03-04
- don’t let the thread sleep too long
0.3.0.0 – 2017-11-21
- add nix with
amqp_0_18_1
- multiple bindings per queue
- remove cool smart special options
0.2.2.0 – 2017-11-20
- option -Q (set queue name for temporary exclusive queue)
0.2.1.5 – 2017-09-25
- travis
- debian
- repair debian jessie
- callback: pass-through timestamp, ignore non-existing sha
- reduce load
- amqp 0.17, use coName
0.2.1.4 – 2017-06-01
- First version. Released on an unsuspecting world.