amqp-tools-cli

Receive messages from amqp, and spawn processes to handle them on cli - a node.js package.

This package was created as a replacement for librabbitmq-tools, with added support for heartbeats, logging and (hopefully) less crashing.

Installation

# Install with npm:
npm install --global --no-dev amqp-tools-cli

# Run with npx:
npx amqp --help

Overview

This package adds a npx-available amqp command, which allows publishing messages to amqp and consuming amqp messages with external programs.

Consumer for each received message a child process running specified program will be launched and fed the message as stdin. The consumer will periodically check if the connection with server is intact (heartbeat). If processing of any of the messages fails (child process exits with a non-zero status code), further messages won't be received, and the consumer will exit after last child process exits.

Usage

Common options

Option name | type | required? | default | description ------------ | ------ | --------- | --------- | --- help | flag | no | | Show help and quit version | flag | no | | Show version number and quit hostname | string | no | localhost | amqp server listening hostname port | number | no | 5672 | amqp server listening port user | string | no | | amqp user password | string | no | | amqp password vhost | string | no | / | amqp vhost heartbeat | number | no | 60 | heartbeat in seconds, if server does not respond within 2 heartbeats, the connection will be closed log-level | string | no | warn | valid values are: fatal, error, warn, info, trace, debug

All options can be read from environmental variables with prefix "AMQP_TOOLS", e.g. password can be given as:

npx amqp consume-queue --queue="queue" --password="mysecretpassword" -- ./on-message.sh # password on command line, a bad idea

or:

env AMQP_TOOLS_PASSWORD="mysecretpassword" npx amqp consume-queue --queue="queue" -- ./on-message.sh

amqp consume-queue

Consumes messages from an existing queue.

Option name | type | required? | default | description -------------------------- | ------ | --------- | ------- | --- -n, --prefetch, --parallel | number | no | 5 | How many messages can be processed in parallel --ignore-consumer-errors | flag | no | false | Continue operations if a command run for message exits with a non-zero status --reject-after-tries | number | no | 0 | If message caused consumer to fail this many times, reject it. Values equal or less than 0 disable this check. If reject-after-seconds is also specified, both have to be satisfied for rejection to happen. --reject-after-seconds | number | no | 0 | If message caused consumer to fail and was first published more than this many seconds, reject it. Values equal or less than 0 disable this check. If reject-after-tries is also specified, both have to be satisfied for rejection to happen. --queue | string | yes | | Name of the queue to consume commands from [--] command | string | yes | | Command/program to invoke for each message with the message content on stdin ...commandArguments | any | no | | Arguments for the command/program

examples

Print messages from amqp queue:

npx amqp consume-queue --queue "queueName" -- cat

Use messages as arguments for a program:

npx amqp consume-queue --queue "queueName" -- xargs your-program

amqp consume-exchange

Creates an exclusive queue, binds it to an existing exchange using provided routing-key, and consumes messages.

Option name | type | required? | default | description -------------------------- | ------ | --------- | ------- | --- -n, --prefetch, --parallel | number | no | 5 | How many messages can be processed in parallel --ignore-consumer-errors | flag | no | false | Continue operations if a command run for message exits with a non-zero status --reject-after-tries | number | no | 0 | If message caused consumer to fail this many times, reject it. Values equal or less than 0 disable this check. If reject-after-seconds is also specified, both have to be satisfied for rejection to happen. --reject-after-seconds | number | no | 0 | If message caused consumer to fail and was first published more than this many seconds, reject it. Values equal or less than 0 disable this check. If reject-after-tries is also specified, both have to be satisfied for rejection to happen. --exchange | string | yes | | Name of the exchange to bind to --routing-key | string | yes | | Routing key pattern to bind with [--] command | string | yes | | Command/program to invoke for each message with the message content on stdin ...commandArguments | any | no | | Arguments for the command/program

examples

Listen for all messages passing through topic exchange

npx amqp consume-exchange --exchange "amq.topic" --routing-key "#"  -- cat

amqp publish-message

Publish a message to exchange with given routing key

Option name | type | required? | description -------------- | ------ | --------- | ------- --exchange | string | yes | Name of exchange to which the message will be published --routing-key | string | yes | Routing key set on the message --content-type | string | no | Content-type of the message [--] content | string | no | Content of the message, if missing, content will be read from stdin

examples

Publish a message to default exchange from command line arguments:

npx amqp publish-message --exchange "" --routing-key "queue" -- "Hello I love you, won't you tell me your name?"

Publish a JSON message from stdin:

echo '["Hello I love you, let me jump into your game."]' \
  | npx amqp publish-message --exchange "" --routing-key "queue" --content-type="application/json"

Testing

Simplest way to test this library with docker:

docker run -d --hostname my-rabbit --name some-rabbit -p 5672:5672 rabbitmq:3
uild/bin/amqp.js consume-exchange --exchange="amq.direct" --routing-key=""  --log-level=info -- cat

#in different terminal
echo "message" | build/bin/amqp.js publish-message --exchange="" --routing-key=""

Real tests might get even written at some point.

Notes

If different message published is used, to make sure --reject-after-seconds measures from the time the message was first published, set header x-first-published-timestamp: <unix timestamp> when publishing the message