moleculer-rabbitmq
v1.2.0
Published
Moleculer RabbitMQ queue plugin
Downloads
298
Readme
moleculer-rabbitmq
Task queue mixin for RabbitMQ AMQP.
Install
$ npm install moleculer-rabbitmq --save
Usage
Enable async queue for exist action
const QueueMixin = require("moleculer-rabbitmq");
const queueMixin = QueueMixin({
connection: "amqp://localhost",
asyncActions: true, // Enable auto generate .async version for actions
});
broker.createService({
name: "consumer",
mixins: [queueMixin],
settings: {
amqp: {
connection: "amqp://localhost", // You can also override setting from service setting
},
},
actions: {
hello: {
// Enable queue for this action
queue: {
// Options for AMQP queue
amqp: {
queueAssert: {
durable: true,
},
consume: {
noAck: false,
},
prefetch: 0,
},
},
params: {
name: "string|convert:true|empty:false",
},
async handler(ctx) {
this.logger.info(
`[CONSUMER] PID: ${process.pid} Received job with name=${ctx.params.name}`
);
return new Promise((resolve) => {
setTimeout(() => {
this.logger.info(
`[CONSUMER] PID: ${process.pid} Processed job with name=${ctx.params.name}`
);
return resolve(`hello ${ctx.params.name}`);
}, 1000);
});
},
},
},
});
Call async action to queue jobs
const QueueMixin = require("moleculer-rabbitmq");
const queueMixin = QueueMixin({
connection: "amqp://localhost",
asyncActions: true, // Enable auto generate .async version for actions
});
broker.createService({
name: "publisher",
mixins: [queueMixin],
settings: {
amqp: {
connection: "amqp://localhost", // You can also override setting from service setting
},
},
async started() {
await broker.waitForServices("consumer");
let name = 1;
setInterval(async () => {
const response = await broker.call("consumer.hello.async", {
// `params` is the real param will be passed to original action
params: {
name,
},
// `options` is the real options will be passed to original action
options: {
timeout: 2000,
},
});
this.logger.info(
`[PUBLISHER] Called job with name=${name} response=${response}`
);
name++;
}, 2000);
},
});
Retry failed jobs
By default, this plugin will not retry failed job. There are two option to enable retry logic: RabbitMQ requeue and using rabbitmq-delayed-message-exchange plugin
RabbitMQ requeue
Set retry option to true when declare queue to enable Rabbitmq requeue. Please note that the message will re requeue forever because max_retry is not available
actions: {
hello: {
queue: {
retry: true, // Using rabbitmq default requeue logic
},
// ...
},
},
Retry using rabbitmq-delayed-message-exchange plugins
REQUIRE rabbitmq-delayed-message-exchange plugin to be install and enabled in RabbitMQ
Example:
actions: {
hello: {
queue: {
retryExchangeAssert: { // RabbitMQ advance exchange option
durable: true, // (boolean) if true, the exchange will survive broker restarts. Defaults to true.
autoDelete: false, // (boolean) if true, the exchange will be destroyed once the number of bindings for which it is the source drop to zero. Defaults to false.
alternateExchange: null, // (string) an exchange to send messages to if this exchange can’t route them to any queues.
arguments: { // additional arguments, usually parameters for some kind of broker-specific extension e.g., high availability, TTL
},
},
retry: {
max_retry: 3, // Max retry count, 3 mean if the first time failed, it will try 3 more times
delay: (retry_count) => { // Number of miliseconds delay between each retry, could be a number or a function(retry_count) that return a number
return retry_count * 1000;
},
},
},
// ...
},
},
Deduplicate jobs
This plugin allow you to avoid dupplicate task using dedupHash
option that can be number, string or function(ctx) that return a number or string
To get this feature to work, you have to install rabbitmq-message-deduplication plugin.
And please aware that using this feature may cause decrease in queue performance
Example RabbitMQ dockerfile Example plugin usage with deduplication
Plugin Configuration
Mixin configuration
connection: "amqp://localhost", // (String|Object) Required. connection string or object, passed to amqplib.connect (You can also set this on broker.createService settings.amqp.connection parameter)
asyncActions: true, // (Boolean) Optional, default: false. Enable auto generate .async version for actions
Action configuration
queue: {
amqp: {
queueAssert: { // Options for job queue (Ref: http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertQueue)
exclusive: false, // (boolean) if true, scopes the queue to the connection (defaults to false)
durable: true, // (boolean) if true, the queue will survive broker restarts, modulo the effects of exclusive and autoDelete; this defaults to true if not supplied, unlike the others
autoDelete: false, // (boolean) if true, the queue will be deleted when the number of consumers drops to zero (defaults to false)
arguments: { // additional arguments, usually parameters for some kind of broker-specific extension e.g., high availability, TTL
"x-message-deduplication": true, // Preserve for deduplication feature
},
},
retryExchangeAssert: { // Options for retry exchange (Ref: http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange)
durable: true, // (boolean) if true, the exchange will survive broker restarts. Defaults to true.
autoDelete: false, // (boolean) if true, the exchange will be destroyed once the number of bindings for which it is the source drop to zero. Defaults to false.
alternateExchange: null, // (string) an exchange to send messages to if this exchange can’t route them to any queues.
arguments: { // additional arguments, usually parameters for some kind of broker-specific extension e.g., high availability, TTL
"arguments.x-delayed-type": "direct", // Set by this plugin
"x-message-deduplication": true, // Preserve for deduplication feature
},
},
consume: { // Options for consumer (Ref: http://www.squaremobius.net/amqp.node/channel_api.html#channel_consume)
noAck: false,
},
prefetch: 0, // Set the prefetch count for this channel
},
retry: { // (Boolean|Object) : Enable or disable retry option
max_retry: 0, // Max retry count
delay: 0, // Delay in ms each retry
},
dedupHash: null, // (Number|String|Function(ctx)) Hash or function to calculate hash to deduplicate task
}
Examples
Take a look at examples folder for more examples
- Simple example : Example for basic usage
- Retry example : Example with retry logic
- Deduplication example : Example with deduplicate message feature
- Graceful shutdown example : Example with moleculer graceful shutdown feature
Checklist
- [x] Implement retry logic for rabbitmq queue
- [x] Allow deduplicate message
- [X] Graceful shutdown queue
- [ ] Test & Coverage
License
The project is available under the MIT license.