add the ability to create internal queues as quorum type instead of hard-coded classic

Author: Michalis-Apostolou

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

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,43 @@ describe('Test DelayQueue', function() {
     spy.args[0].should.containDeep([{ queueName: 'queue', obj: content }, { expiration: '10' }, 'delay_10']);
     (<any>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 queueInstance = sinon.createStubInstance(Queue.default);
+      const stub = sandbox.stub(Queue, 'default').returns(queueInstance);
+
+      await DelayQueue.createDelayQueue(rabbit.consumeChannel, 'delay', createQueueAsQuorum);
+
+      stub.args[0].should.eql([
+        rabbit.consumeChannel,
+        'delay',
+        {
+          deadLetterExchange: '',
+          deadLetterRoutingKey: 'delay_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']);
+    });
+  });
 });

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_quorum_delay', content, headers, rabbit.consumeChannel, `test_${this.name}`, true]]);
+    });
+  });
 });

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(

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 ? 'quorum_delay' : '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) {