Merge pull request #69 from Workable/introduce-default-queue-type

Introduce default queue type

Author: nikosd23

test/rabbit.test.ts

@@ -119,6 +119,81 @@ describe('Test rabbit class', function() {
     stub.calledWith(handler).should.be.true();
   });
 
+  it('should create queue with max priority when priority is specified', async function() {
+    const subscription = sandbox.stub(Queue.prototype, 'subscribe');
+    rabbit = new Rabbit(this.url);
+    await rabbit.connected;
+    const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue')
+      .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 });
+    const priority = 23;
+    const handler = () => {};
+    await rabbit.createQueue(this.name, { priority: priority }, handler);
+    subscription.calledWith(handler).should.be.true();
+    const [name, options] = stub.firstCall.args;
+    name.should.equal(this.name);
+    options!.arguments['x-max-priority'].should.equals(priority);
+  });
+
+  it('should apply queue type when defaultQueueType is provided', async function() {
+    const subscription = sandbox.stub(Queue.prototype, 'subscribe');
+    const queueType = 'classic';
+    rabbit = new Rabbit(this.url, { defaultQueueType: queueType });
+    await rabbit.connected;
+    const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue')
+      .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 });
+    const handler = () => {};
+    await rabbit.createQueue(this.name, {}, handler);
+    subscription.calledWith(handler).should.be.true();
+    const [name, options] = stub.firstCall.args;
+    name.should.equal(this.name);
+    options!.arguments['x-queue-type'].should.equals(queueType);
+  });
+
+  it('should not set x-queue-type when neither default nor explicit queue type is provided', async function() {
+    const subscription = sandbox.stub(Queue.prototype, 'subscribe');
+    rabbit = new Rabbit(this.url);
+    await rabbit.connected;
+    const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue')
+      .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 });
+    const handler = () => {};
+    await rabbit.createQueue(this.name, { }, handler);
+    subscription.calledWith(handler).should.be.true();
+    stub.calledOnce.should.be.true();
+    const [name, options] = stub.firstCall.args;
+    name.should.equal(this.name);
+    options!.arguments.should.not.have.property('x-queue-type');
+  });
+
+  it('should use the provided queue type instead of the default', async function() {
+    const subscription = sandbox.stub(Queue.prototype, 'subscribe');
+    rabbit = new Rabbit(this.url, { defaultQueueType: 'classic' });
+    await rabbit.connected;
+    const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue')
+      .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 });
+    const handler = () => {};
+    const queueType = 'quorum';
+    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);
+    options!.arguments['x-queue-type'].should.equals(queueType);
+  });
+
+  it('should ignore priority when creating a quorum queue', async function() {
+    const subscription = sandbox.stub(Queue.prototype, 'subscribe');
+    rabbit = new Rabbit(this.url, { defaultQueueType: 'quorum' });
+    await rabbit.connected;
+    const stub = sandbox.stub(rabbit.consumeChannel, 'assertQueue')
+      .resolves({ queue: this.name, messageCount: 0, consumerCount: 0 });
+    const handler = () => {};
+    await rabbit.createQueue(this.name, { priority: 23 }, handler);
+    subscription.calledWith(handler).should.be.true();
+    stub.calledOnce.should.be.true();
+    const [name, options] = stub.firstCall.args;
+    name.should.equal(this.name);
+    options!.arguments.should.not.have.property('x-max-priority');
+  });
+
   it('should unsubscribe', async function() {
     const stub = sandbox.stub(Queue.prototype, 'unsubscribe');
     rabbit = new Rabbit(this.url);

ts/queue.ts

@@ -54,6 +54,11 @@ export default class Queue {
     if (priority !== undefined) {
       queueOptions.arguments = { ...queueOptions.arguments, 'x-max-priority': priority };
     }
+    if (queueOptions.arguments?.['x-max-priority'] &&
+      queueOptions.arguments?.['x-queue-type'] === 'quorum') {
+      logger.warn(`Invalid x-max-priority argument for quorum queue '${this.name}'`);
+      delete queueOptions.arguments['x-max-priority'];
+    }
     await this.channel.assertQueue(this.name, queueOptions);
   }
 

ts/rabbit.ts

@@ -25,11 +25,26 @@ export default class Rabbit extends EventEmitter {
   public prefix: string;
   public scheduledPublish: boolean;
   public socketOptions;
+  private defaultQueueType: string | undefined;
   public logger: ReturnType<typeof getLogger>;
 
   constructor(
     public url: string,
-    { prefetch = 1, replyPattern = true, prefix = '', scheduledPublish = false, socketOptions = {} } = {}
+    {
+      prefetch = 1,
+      replyPattern = true,
+      prefix = '',
+      scheduledPublish = false,
+      socketOptions = {},
+      defaultQueueType = undefined,
+    }: {
+      prefetch?: number
+      replyPattern?: boolean
+      prefix?: string
+      scheduledPublish?: boolean
+      socketOptions?: any
+      defaultQueueType?: string
+    } = {}
   ) {
     super();
     if (!Rabbit.INSTANCE) {
@@ -42,6 +57,7 @@ export default class Rabbit extends EventEmitter {
     this.prefix = prefix;
     this.scheduledPublish = scheduledPublish;
     this.socketOptions = socketOptions;
+    this.defaultQueueType = defaultQueueType;
     this.reconnect();
   }
 
@@ -101,9 +117,12 @@ export default class Rabbit extends EventEmitter {
 
   async createQueue(
     name: string,
-    options: amqp.Options.AssertQueue & { 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']) {
+      options.arguments = { ...options.arguments, 'x-queue-type': this.defaultQueueType };
+    }
     options.prefetch = options.prefetch || this.prefetch;
     name = this.updateName(name, options.prefix);
     await this.connected;