,---. | | o| ,---.
|---',---.|---.|---..|--- | |. .,---.. .,---.
| \ ,---|| || ||| | || ||---'| ||---'
` ``---^`---'`---'``---' `---\`---'`---'`---'`---'
This module is based on WRabbit and Jackrabbit.
It makes RabbitMQ management and integration easy. Provides an abstraction layer above amqplib.
It is written in typescript and requires node v10.0.0 or higher. Most features will work with nodejs v6.0.0 and higher but using older versions than v10.0.0 is not recommended. RPC streams will not work in older versions.
By default it works well with jsons. Override properties.contentType to work with strings or binary data.
const { Rabbit } = require('rabbit-queue');
const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost', {
prefetch: 1, //default prefetch from queue
replyPattern: true, //if reply pattern is enabled an exclusive queue is created
scheduledPublish: false,
scheduledPublishQueuesAsQuorum: false, //declare the publishWithDelay queues as quorum (see Scheduled publish queues as quorum)
prefix: '', //prefix all queues with an application name
tolerateQueueTypeMismatch: false, //consume an existing queue whose type or priority differs from the declared one (see Queue Type Behavior)
cancelConsumersOnSigterm: true, //cancel every consumer when the process receives SIGTERM (see Shutdown)
socketOptions: {} // socketOptions will be passed as a second param to amqp.connect and from ther to the socket library (net or tls)
});
rabbit.on('connected', () => {
//create queues, add halders etc.
//this will be called on every reconnection too
});
rabbit.on('disconnected', (err = new Error('Rabbitmq Disconnected')) => {
//handle disconnections and try to reconnect
console.error(err);
setTimeout(() => rabbit.reconnect(), 100);
});const { Rabbit } = require('rabbit-queue');
const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost', { scheduledPublish: true });
rabbit
.createQueue('queueName', { durable: false }, (msg, ack) => {
console.log(msg.content.toString());
ack(null, 'response');
})
.then(() => console.log('queue created'));
rabbit.publish('queueName', { test: 'data' }, { correlationId: '1' }).then(() => console.log('message published'));
// Please note that a queue will be created for each different expiration used: prefix_delay_expiration.
rabbit
.publishWithDelay('queueName', { test: 'data' }, { correlationId: '1', expiration: '10000' })
.then(() => console.log('message will be published'));
rabbit
.getReply('queueName', { test: 'data' }, { correlationId: '1' })
.then((reply) => console.log('received reply', reply));
rabbit
.getReply('queueName', { test: 'data' }, { correlationId: '1' }, '', 100)
.then((reply) => console.log('received reply', reply))
.catch((error) => console.log('Timed out after 100ms'));
rabbit.bindToTopic('queueName', 'routingKey');
rabbit
.getTopicReply('routingKey', { test: 'data' }, { correlationId: '1' }, '', 100)
.then((reply) => console.log('received reply', reply))
.catch((error) => console.log('Timed out after 100ms'));const { Rabbit } = require('rabbit-queue');
const rabbit = new Rabbit('amqp://localhost');
rabbit
.createQueue('queueName', { durable: false }, (msg, ack) => {
console.log(msg.content.toString());
ack(null, 'response');
})
.then(() => console.log('queue created'));
rabbit.bindToExchange('queueName', 'amq.topic', 'routingKey');
//shortcut rabbit.bindToTopic('queueName', 'routingKey');
rabbit
.publishExchange('amq.topic', 'routingKey', { test: 'data' }, { correlationId: '1' })
.then(() => console.log('message published'));
//shortcut rabbit.publishTopic( 'routingKey', { test: 'data' }, { correlationId: '1' });const rabbit = new Rabbit('amqp://localhost');
class DemoHandler extends BaseQueueHandler {
handle({ msg, event, correlationId, startTime }) {
console.log('Received: ', event);
console.log('With correlation id: ' + correlationId);
}
afterDlq({ msg, event }) {
// Something to do after added to dlq
}
}
new DemoHandler('demoQueue', rabbit, {
retries: 3,
retryDelay: 1000,
logEnabled: true, //log queue processing time
scope: 'SINGLETON', //can also be 'PROTOTYPE' to create a new instance every time
createAndSubscribeToQueue: true // used internally no need to overwrite
});
rabbit.publish('demoQueue', { test: 'data' }, { correlationId: '4' });const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost');
class DemoHandler extends BaseQueueHandler {
handle({ msg, event, correlationId, startTime }) {
console.log('Received: ', event);
console.log('With correlation id: ' + correlationId);
return Promise.resolve('reply'); // could be return 'reply';
}
afterDlq({ msg, event }) {
// Something to do after added to dlq
}
}
new DemoHandler('demoQueue', rabbit, {
retries: 3,
retryDelay: 1000,
logEnabled: true,
scope: 'SINGLETON'
});
rabbit
.getReply('demoQueue', { test: 'data' }, { correlationId: '5' })
.then((reply) => console.log('received reply', reply)); const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost');
class DemoHandler extends BaseQueueHandler {
handle({msg, event, correlationId, startTime}) {
return Promise.reject(new Error('test Error')); // could be synchronous: throw new Error('test error');
}
afterDlq({msg, event}) {
console.log('added to dlq');
}
}
new DemoHandler('demoQueue', rabbit,
{
retries: 3,
retryDelay: 1,
logEnabled: true
});
rabbit.getReply('demoQueue', { test: 'data' }, { correlationId: '5' })
.then(reply => console.log('received reply', reply));
.catch(error => console.log('error', error)); //error will be 'test Error';const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost');
class DemoHandler extends BaseQueueHandler {
handle({ msg, event, correlationId, startTime }) {
const stream = new Readable({ read() {} });
stream.push('streaming');
stream.push('events');
stream.push(null); //the end
return stream;
}
}
new DemoHandler('demoQueue', rabbit, {
retries: 3,
retryDelay: 1,
logEnabled: true
});
const reply = await rabbit.getReply('demoQueue', { test: 'data' }, { correlationId: '5' });
for await (const chunk of reply) {
console.log(`Received chunk: ${chunk.toString()}`);
}Rabbit-queue logs to console by default. It also emits events for each log so that you can use your own logger
eg:
const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost');
rabbit.on('log', (component, level, ...args) => console.log(`[${level}] ${component}`, ...args));Rabbit-queue supports migrating queues from one configuration to another.
RabbitMQ will fail and close the channel if you try to create a queue with the same name but different configuration.
To migrate a queue, you can use the migrateQueue option.
const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost');
class DemoHandler extends BaseQueueHandler {
handle({ msg, event, correlationId, startTime }) {
console.log('Received: ', event);
console.log('With correlation id: ' + correlationId);
}
afterDlq({ msg, event }) {
// Something to do after added to dlq
}
}
new DemoHandler('demoQueue', rabbit, {
retries: 3,
retryDelay: 1000,
logEnabled: true, //log queue processing time
scope: 'SINGLETON', //can also be 'PROTOTYPE' to create a new instance every time
createAndSubscribeToQueue: true, // used internally no need to overwrite
migrateQueue: true // if you want to migrate the queue from one configuration to another - otherwise it will throw an error if the queue already exists with different configuration
});
rabbit.publish('demoQueue', { test: 'data' }, { correlationId: '4' });You can control the type of queues created with the defaultQueueType option in your Rabbit configuration.
const rabbit = new Rabbit(process.env.RABBIT_URL, { defaultQueueType: 'quorum' });When declaring queues, the following rules apply:
- Non-durable, exclusive, or auto-delete queues always use classic queue type, since quorum queues and streams are always durable and not designed for temporary or connection-bound queues.
The type of a queue is immutable once it has been declared. When a queue already exists with different arguments, the broker rejects the declare with PRECONDITION_FAILED and createQueue rejects. Queue declares, binds and unbinds run on a short-lived channel, so a rejected one doesn't close the channel the other consumers are on.
While queues move between classic and quorum, old and new declares disagree on the queue type and on priority, which quorum queues don't support. With tolerateQueueTypeMismatch: true, a mismatch on x-queue-type or x-max-priority only logs a warning, and the queue is consumed as it exists. Any other mismatch (e.g. dead-letter exchange or routing key, TTL, max length) still rejects createQueue, even when the type differs too.
const rabbit = new Rabbit(process.env.RABBIT_URL, { tolerateQueueTypeMismatch: true });publishWithDelay uses internal queues: a delay_{expiration} queue for each expiration, whose expired messages are dead-lettered to delay_reply and republished from there (both prefixed with prefix_ when a prefix is set). They are declared without a queue type, so they get the virtual host's default type. With scheduledPublishQueuesAsQuorum: true they are declared as quorum queues, named delay_quorum_{expiration} and delay_quorum_reply, so they don't conflict with the existing classic ones. Off by default.
const rabbit = new Rabbit(process.env.RABBIT_URL, { scheduledPublish: true, scheduledPublishQueuesAsQuorum: true });Quorum queues need RabbitMQ 3.10 or later for per-message TTL and at-least-once dead-lettering.
Messages already in a classic delay_{expiration} queue are still dead-lettered to delay_reply. Keep at least one consumer running without the option until the classic delay queues are empty, then delete them along with delay_reply.
On SIGTERM, rabbit-queue cancels the consumers of every queue it created, so the process stops taking new messages while it shuts down. Set cancelConsumersOnSigterm: false when something else in the process has its own SIGTERM hook that waits on messages from these queues, e.g. eip-redis's PubSub waiting for throttled events that only complete when an eip-rabbit Timer queue delivers. With the consumers cancelled, that wait never ends and the process hangs until it is killed. On by default.
const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: false });cancelConsumersOnSigtermoption. Set it tofalseto keep consuming afterSIGTERM, for processes that drain work through their ownSIGTERMhook. On by default.
scheduledPublishQueuesAsQuorumoption. When enabled, the internal queues used bypublishWithDelayare declared as quorum queues, under new names (delay_quorum_{expiration},delay_quorum_reply). Off by default.- Queue declares, binds and unbinds run on short-lived channels. One that fails still rejects, but no longer closes the shared consume channel and disconnects every consumer.
tolerateQueueTypeMismatchoption. When enabled, a queue that already exists with a different type or priority (classic vs quorum, with or withoutx-max-priority) is consumed as it is, with a warning. Old and new pods can then run side by side during a rolling deploy that changes a queue's type or priority, e.g. while migrating classic mirrored queues to quorum. Off by default.
- Support for stream RPC to pass options for the Readable stream created in the caller. e.g.
const reply = await rabbit.getReply(
'demoQueue',
{ test: 'data' },
{ headers: { test: 1, backpressure: true, options: { highWaterMark: 1 } }, correlationId: '1' }
);
for await (const chunk of reply) {
console.log(`Received chunk: ${chunk.toString()}`);
if ('sufficient_data_received') {
reply.emit(Queue.STOP_STREAM);
}
}- Support for Node LTS v6 and v8 was dropped. You should use Node v10 and higher.
- Two connections are created by default to rabbitMQ. One for consuming and one for producing messages.
- Dependency to log4js was dropped. Console is used by default but you can easily plug your own logger.
When declaring queues with handlers you can define a prefetch count different from the global one
rabbit.createQueue('queueName', { durable: false, prefetch: 100 }, (msg, ack) => {
console.log(msg.content.toString());
ack(null, 'response');
});
// or
class DemoHandler extends BaseQueueHandler {
// ...
}
new DemoHandler('demoQueue', rabbit, { prefetch: 100 });Note that the prefetch value is set to RabbitMQ and if you have other ways of creating consumers eg. by calling queue.subscribe directly you might end up with consumers being created with different prefetch from the global. The two ways mentioned above are handled by rabbit-queue and they are synchronized so that no other queue consumer might be affected.
Also note that if you use rabbit prior to 3.3.0 the behavior might be different. See https://www.rabbitmq.com/consumer-prefetch.html for more details
RPC stream enhancement: When backpressure is enabled, the consumer can stop communication, when data received is sufficient
e.g.
const reply = await rabbit.getReply(
'demoQueue',
{ test: 'data' },
{ headers: { test: 1, backpressure: true }, correlationId: '1' }
);
for await (const chunk of reply) {
console.log(`Received chunk: ${chunk.toString()}`);
if ('sufficient_data_received') {
reply.emit(Queue.STOP_STREAM);
}
}Get reply as a stream supports two more optional headers inside properties:
backpressure (by default false), timeout
eg:
await rabbit.getReply(
'demoQueue',
{ test: 'data' },
{ correlationId: '5', headers: { backpressure: true, timeout: 10000 } }
);If used the rpc that responds to this request will stop sending messages until the receiving stream has consumed those messages or has buffered them (By default nodejs stream buffer for json streams is 16 objects). If this does not happen within the timeout the process will stop.
Use this feature only if both requesting and receiving part have rabbit-queue > 4.2.0
Code was reorganized. RPC stream was introduced. Drops support for node older than v10.0.0 due to the usage of async generators Supports contentType. By default it is application/json for backwards compatibility.
Log4js was removed. You can no longer pass your own logger in Rabbit constructor. Instead log4js-api is used and your log4js configuration is used by default if present. Logger name is rabbit-queue.
Queue subscribe 2ond param ack was updated. Now it accepts as 1st param an error and as a second the response. Rpc calls will throw an error if the first param is defined.
The tests are set up with Docker + Docker-Compose, so you don't need to install rabbitmq (or even node) to run them:
npm run test-docker
npm version patch
git push --follow-tags
npm run gh:release