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

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.
 

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']);
     (<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 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 = (<any>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']);
+      (<any>await promise).content.toString().should.eql(JSON.stringify(content));
+    });
+  });
 });

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;

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(

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<typeof getLogger>;
   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) {