diff --git a/README.md b/README.md index 5391356..f92fcd9 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,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' }); @@ -262,6 +262,9 @@ When declaring queues, the following rules apply: The type of a queue is **immutable** once it has been declared. Attempting to change it after creation will result in a **PRECONDITION_FAILED** error. +So far the default internal queue `({prefix}_)?delay_reply` was always created as classic. The same applies for queues created when publishing with delay which followed the format `({prefix}_)?delay_{expiration}`. In order to avoid conflicts with existing queues in the cluster , when opted-in to create queues as `quorum` default type , new internal queue will be created with the format `({prefix}_)?delay_quorum_reply`. That way you can have old deployments using the classic queues and newer deployments with quorum queues. + + ### Changelog ### New in v5.4.x diff --git a/test/delay-queue.test.ts b/test/delay-queue.test.ts index 67d99cc..58d9fcd 100644 --- a/test/delay-queue.test.ts +++ b/test/delay-queue.test.ts @@ -22,9 +22,7 @@ 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_reply'); await rabbit.close(); }); @@ -38,7 +36,7 @@ describe('Test DelayQueue', function() { 'delay', { deadLetterExchange: '', - deadLetterRoutingKey: 'delay_reply' + deadLetterRoutingKey: 'delay_reply', } ]); }); @@ -85,4 +83,45 @@ 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' } + } + ]); + }); + + 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' } }]]); + }); + + it('should publishWithDelay and create not existing queue', async function() { + const stub = sandbox.stub(Queue.default, 'publish').returns(null); + + await DelayQueue.publishWithDelay('delay', {}, {}, rabbit.consumeChannel, 'test', createQueueAsQuorum); + + stub.calledOnce.should.be.true(); + stub.args[0].should.containDeep([{ queueName: 'test', obj: {} }, { expiration: '10000' }, 'delay_10000']); + }); + }); }); diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index 85ac79a..95e51d3 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -157,7 +157,7 @@ describe('Test rabbit class', function() { const stub = sandbox.stub(rabbit.consumeChannel, '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; @@ -173,7 +173,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); @@ -284,7 +284,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() { @@ -393,4 +393,16 @@ describe('Test rabbit class', function() { (rabbit as any).sigtermHandler(); stub.calledTwice.should.be.true(); }); + + describe('when defaultQueueType is quorum', 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, defaultQueueType: 'quorum' }); + 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]]); + }); + }); }); diff --git a/ts/delay-queue.ts b/ts/delay-queue.ts index 8c4401a..dd7a186 100644 --- a/ts/delay-queue.ts +++ b/ts/delay-queue.ts @@ -9,17 +9,18 @@ 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, {}); +export async function createDelayQueueReply(channel: Channel, delayedQueueName: string, createAsQuorum: boolean = false) { + delayedQueueNameReply = createAsQuorum ? `${delayedQueueName}_quorum_reply` : `${delayedQueueName}_reply`; + delayedQueueReply = new Queue(channel, delayedQueueNameReply, { ...createAsQuorum && { arguments: { 'x-queue-type': 'quorum' } } }); await delayedQueueReply.created; delayedQueueReply.subscribe(onMessage(channel)); } -export async function createDelayQueue(channel: Channel, delayedQueueName: string) { +export async function createDelayQueue(channel: Channel, delayedQueueName: string, createAsQuorum: boolean = false) { delayedQueue[delayedQueueName] = new Queue(channel, delayedQueueName, { deadLetterExchange: '', - deadLetterRoutingKey: delayedQueueNameReply + deadLetterRoutingKey: delayedQueueNameReply, + ...createAsQuorum && { arguments: { 'x-queue-type': 'quorum' } } }); await delayedQueue[delayedQueueName].created; } @@ -29,13 +30,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 e3c5d73..775fed7 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -104,7 +104,8 @@ export default class Rabbit extends EventEmitter { await createReplyQueue(this.consumeChannel); } if (!publish && this.scheduledPublish) { - await createDelayQueueReply(this.consumeChannel, this.updateName('delay')); + const createAsQuorum = this.defaultQueueType === 'quorum'; + await createDelayQueueReply(this.consumeChannel, this.updateName('delay'), createAsQuorum); } } @@ -121,7 +122,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']) { @@ -187,7 +188,9 @@ 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 createQueueAsQuorum = this.defaultQueueType === 'quorum'; + const queueName = createQueueAsQuorum ? 'delay_quorum' : 'delay'; + await publishWithDelay(this.updateName(queueName), obj, properties, this.consumeChannel, name, createQueueAsQuorum); } async getReply(name: string, obj, properties: amqp.Options.Publish, prefix?: string, timeout?: number) {