diff --git a/README.md b/README.md index 6ceaceb..c68e707 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ 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) socketOptions: {} // socketOptions will be passed as a second param to amqp.connect and from ther to the socket library (net or tls) @@ -122,7 +123,7 @@ new DemoHandler('demoQueue', rabbit, { 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ÏÏ + createAndSubscribeToQueue: true // used internally no need to overwrite }); rabbit.publish('demoQueue', { test: 'data' }, { correlationId: '4' }); @@ -268,10 +269,22 @@ While queues move between classic and quorum, old and new declares disagree on t const rabbit = new Rabbit(process.env.RABBIT_URL, { tolerateQueueTypeMismatch: true }); ``` +### Scheduled publish queues as quorum + +`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. +```javascript +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`. + ### Changelog ### New in v5.10.x +- `scheduledPublishQueuesAsQuorum` option. When enabled, the internal queues used by `publishWithDelay` are 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. - `tolerateQueueTypeMismatch` option. When enabled, a queue that already exists with a different type or priority (classic vs quorum, with or without `x-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. diff --git a/test/delay-queue.test.ts b/test/delay-queue.test.ts index 67d99cc..bc68633 100644 --- a/test/delay-queue.test.ts +++ b/test/delay-queue.test.ts @@ -22,9 +22,8 @@ describe('Test DelayQueue', function() { await rabbit.destroyQueue('delay_3000'); await rabbit.destroyQueue('delay_10'); await rabbit.destroyQueue('delay_reply'); - }); - - afterEach(async function() { + await rabbit.destroyQueue('delay_quorum_10'); + await rabbit.destroyQueue('delay_quorum_reply'); await rabbit.close(); }); @@ -38,7 +37,7 @@ describe('Test DelayQueue', function() { 'delay', { deadLetterExchange: '', - deadLetterRoutingKey: 'delay_reply' + deadLetterRoutingKey: 'delay_reply', } ]); }); @@ -85,4 +84,89 @@ describe('Test DelayQueue', function() { spy.args[0].should.containDeep([{ queueName: 'queue', obj: content }, { expiration: '10' }, 'delay_10']); (await promise).content.toString().should.eql(JSON.stringify(content)); }); + + describe('when option createAsQuorum is true', function() { + const createQueueAsQuorum = true; + + it('should createDelayQueue as quorum type', async function() { + const delayQueueName = 'delay'; + await DelayQueue.createDelayQueueReply(rabbit.consumeChannel, delayQueueName, true); + const queueInstance = sinon.createStubInstance(Queue.default); + const stub = sandbox.stub(Queue, 'default').returns(queueInstance); + + await DelayQueue.createDelayQueue(rabbit.consumeChannel, delayQueueName, createQueueAsQuorum); + + stub.args[0].should.eql([ + rabbit.consumeChannel, + delayQueueName, + { + deadLetterExchange: '', + deadLetterRoutingKey: 'delay_quorum_reply', + arguments: { + 'x-queue-type': 'quorum', + 'x-overflow': 'reject-publish', + 'x-dead-letter-strategy': 'at-least-once' + } + } + ]); + }); + + it('should createDelayQueueReply as quorum with relevant name', async function() { + const queueInstance = sinon.createStubInstance(Queue.default); + const stub = sandbox.stub(Queue, 'default').returns(queueInstance); + + await DelayQueue.createDelayQueueReply(rabbit.consumeChannel, 'delay', createQueueAsQuorum); + + stub.args.should.eql([[rabbit.consumeChannel, 'delay_quorum_reply', { + arguments: { + 'x-queue-type': 'quorum', + 'x-overflow': 'reject-publish', + 'x-dead-letter-strategy': 'at-least-once' + } + }]]); + }); + + it('should publishWithDelay and create not existing quorum queue', async function() { + await DelayQueue.createDelayQueueReply(rabbit.consumeChannel, 'delay', createQueueAsQuorum); + const queueInstance = sinon.createStubInstance(Queue.default); + const stub = sandbox.stub(Queue, 'default').returns(queueInstance); + const publish = (stub).publish = sandbox.stub().returns(null); + + // An expiration no other test uses, so the delay queue isn't already cached + await DelayQueue.publishWithDelay('delay_quorum', {}, { expiration: '20' }, rabbit.consumeChannel, 'test', + createQueueAsQuorum); + + stub.args.should.eql([[rabbit.consumeChannel, 'delay_quorum_20', { + deadLetterExchange: '', + deadLetterRoutingKey: 'delay_quorum_reply', + arguments: { + 'x-queue-type': 'quorum', + 'x-overflow': 'reject-publish', + 'x-dead-letter-strategy': 'at-least-once' + } + }]]); + publish.calledOnce.should.be.true(); + publish.args[0].should.containDeep([{ queueName: 'test', obj: {} }, { expiration: '20' }]); + }); + + it('should deliver the message through the quorum queues', async function() { + await DelayQueue.createDelayQueueReply(rabbit.consumeChannel, 'delay', createQueueAsQuorum); + const queue = new Queue.default(rabbit.consumeChannel, 'queue', { exclusive: true }); + const content = { content: true }; + let resolve; + const promise = new Promise(r => { + resolve = r; + }); + await queue.subscribe(function(msg, ack) { + ack(); + resolve(msg); + }); + + const spy = sandbox.spy(Queue.default, 'publish'); + await DelayQueue.publishWithDelay('delay_quorum', content, { expiration: '10' }, rabbit.consumeChannel, 'queue', + createQueueAsQuorum); + spy.args[0].should.containDeep([{ queueName: 'queue', obj: content }, { expiration: '10' }, 'delay_quorum_10']); + (await promise).content.toString().should.eql(JSON.stringify(content)); + }); + }); }); diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index cbe3e98..72a446e 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -162,7 +162,7 @@ describe('Test rabbit class', function() { const stub = sandbox.stub(AmqpChannel.prototype, 'assertQueue') .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; - await rabbit.createQueue(this.name, { }, handler); + await rabbit.createQueue(this.name, {}, handler); subscription.calledWith(handler).should.be.true(); stub.calledOnce.should.be.true(); const [name, options] = stub.firstCall.args; @@ -178,7 +178,7 @@ describe('Test rabbit class', function() { .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 }); const handler = () => {}; const queueType = 'quorum'; - await rabbit.createQueue(this.name, { arguments: { 'x-queue-type': queueType }}, handler); + await rabbit.createQueue(this.name, { arguments: { 'x-queue-type': queueType } }, handler); subscription.calledWith(handler).should.be.true(); const [name, options] = stub.firstCall.args; name.should.equal(this.name); @@ -289,7 +289,7 @@ describe('Test rabbit class', function() { const headers = { headers: { test: 1 } }; await rabbit.publishWithDelay(`test_${this.name}`, content, headers); stub.calledOnce.should.be.true(); - stub.args.should.eql([['test_delay', content, headers, rabbit.consumeChannel, `test_${this.name}`]]); + stub.args.should.eql([['test_delay', content, headers, rabbit.consumeChannel, `test_${this.name}`, false]]); }); it('should publish to queue with getReply', async function() { @@ -399,6 +399,18 @@ describe('Test rabbit class', function() { stub.calledTwice.should.be.true(); }); + describe('when scheduledPublishQueuesAsQuorum option is true', function() { + it('should publish to queue with Delay, and use quorum delay queue', async function() { + const stub = sandbox.stub(DelayQueue, 'publishWithDelay'); + rabbit = new Rabbit(this.url, { prefix: 'test', scheduledPublish: true, scheduledPublishQueuesAsQuorum: true }); + const content = { content: true }; + const headers = { headers: { test: 1 } }; + await rabbit.publishWithDelay(`test_${this.name}`, content, headers); + stub.calledOnce.should.be.true(); + stub.args.should.eql([['test_delay_quorum', content, headers, rabbit.consumeChannel, `test_${this.name}`, true]]); + }); + }); + describe('when the queue exists with different arguments', function() { let admin: amqp.ChannelModel; let queueName: string; diff --git a/ts/delay-queue.ts b/ts/delay-queue.ts index 8c4401a..bc627ab 100644 --- a/ts/delay-queue.ts +++ b/ts/delay-queue.ts @@ -9,17 +9,24 @@ let delayedQueue: { [key: string]: Queue } = {}; let delayedQueueReply: Queue; let delayedQueueNameReply: string; -export async function createDelayQueueReply(channel: Channel, delayedQueueName: string) { - delayedQueueNameReply = `${delayedQueueName}_reply`; - delayedQueueReply = new Queue(channel, delayedQueueNameReply, {}); +const QUORUM_ARGUMENTS = { + 'x-queue-type': 'quorum', + 'x-overflow': 'reject-publish', + 'x-dead-letter-strategy': 'at-least-once' +}; + +export async function createDelayQueueReply(channel: Channel, delayedQueueName: string, createAsQuorum = false) { + delayedQueueNameReply = createAsQuorum ? `${delayedQueueName}_quorum_reply` : `${delayedQueueName}_reply`; + delayedQueueReply = new Queue(channel, delayedQueueNameReply, createAsQuorum ? { arguments: QUORUM_ARGUMENTS } : {}); await delayedQueueReply.created; delayedQueueReply.subscribe(onMessage(channel)); } -export async function createDelayQueue(channel: Channel, delayedQueueName: string) { +export async function createDelayQueue(channel: Channel, delayedQueueName: string, createAsQuorum = false) { delayedQueue[delayedQueueName] = new Queue(channel, delayedQueueName, { deadLetterExchange: '', - deadLetterRoutingKey: delayedQueueNameReply + deadLetterRoutingKey: delayedQueueNameReply, + ...(createAsQuorum && { arguments: QUORUM_ARGUMENTS }) }); await delayedQueue[delayedQueueName].created; } @@ -29,13 +36,14 @@ export async function publishWithDelay( obj, headers: amqp.Options.Publish = {}, channel: Channel, - queueName: string + queueName: string, + createAsQuorum: boolean = false ) { const { expiration = '10000' } = headers || {}; name = `${name}_${expiration}`; if (!delayedQueue[name]) { - await createDelayQueue(channel, name); + await createDelayQueue(channel, name, createAsQuorum); } const timestamp = new Date().getTime(); Queue.publish( diff --git a/ts/rabbit.ts b/ts/rabbit.ts index 0043bb6..83104d7 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -26,6 +26,7 @@ export default class Rabbit extends EventEmitter { public scheduledPublish: boolean; public socketOptions; private defaultQueueType: string | undefined; + private scheduledPublishQueuesAsQuorum: boolean; private tolerateQueueTypeMismatch: boolean; public logger: ReturnType; private readonly sigtermHandler: () => void; @@ -39,6 +40,7 @@ export default class Rabbit extends EventEmitter { scheduledPublish = false, socketOptions = {}, defaultQueueType = undefined, + scheduledPublishQueuesAsQuorum = false, tolerateQueueTypeMismatch = false, }: { prefetch?: number @@ -47,6 +49,7 @@ export default class Rabbit extends EventEmitter { scheduledPublish?: boolean socketOptions?: any defaultQueueType?: string + scheduledPublishQueuesAsQuorum?: boolean tolerateQueueTypeMismatch?: boolean } = {} ) { @@ -62,6 +65,7 @@ export default class Rabbit extends EventEmitter { this.scheduledPublish = scheduledPublish; this.socketOptions = socketOptions; this.defaultQueueType = defaultQueueType; + this.scheduledPublishQueuesAsQuorum = scheduledPublishQueuesAsQuorum; this.tolerateQueueTypeMismatch = tolerateQueueTypeMismatch; this.sigtermHandler = this.handleSigterm.bind(this); this.reconnect(); @@ -110,7 +114,7 @@ export default class Rabbit extends EventEmitter { await createReplyQueue(this.consumeChannel); } if (!publish && this.scheduledPublish) { - await createDelayQueueReply(this.consumeChannel, this.updateName('delay')); + await createDelayQueueReply(this.consumeChannel, this.updateName('delay'), this.scheduledPublishQueuesAsQuorum); } } @@ -127,7 +131,7 @@ export default class Rabbit extends EventEmitter { async createQueue( name: string, - options: amqp.Options.AssertQueue & amqp.Options.Consume & { prefix?: string; prefetch? } = {}, + options: amqp.Options.AssertQueue & amqp.Options.Consume & { prefix?: string; prefetch?} = {}, handler?: (msg: any, ack: (error?, reply?) => any) => any ) { if (this.defaultQueueType && !options.arguments?.['x-queue-type']) { @@ -196,7 +200,8 @@ export default class Rabbit extends EventEmitter { } name = this.updateName(name, prefix); await this.connected; - await publishWithDelay(this.updateName('delay'), obj, properties, this.consumeChannel, name); + const queueName = this.scheduledPublishQueuesAsQuorum ? 'delay_quorum' : 'delay'; + await publishWithDelay(this.updateName(queueName), obj, properties, this.consumeChannel, name, this.scheduledPublishQueuesAsQuorum); } async getReply(name: string, obj, properties: amqp.Options.Publish, prefix?: string, timeout?: number) {