Merge pull request #70 from Workable/classic-queue-args

Enforce classic queue type for non-durable and exclusive queues

Author: nikosd23

README.md

@@ -250,6 +250,18 @@ new DemoHandler('demoQueue', rabbit, {
 rabbit.publish('demoQueue', { test: 'data' }, { correlationId: '4' });
 ```
 
+### Queue Type Behavior
+
+You can control the type of queues created with the `defaultQueueType` option in your Rabbit configuration.
+```javascript
+const rabbit = new Rabbit(process.env.RABBIT_URL, { defaultQueueType: 'quorum' });
+```
+
+When declaring queues, the following rules apply:
+- **Non-durable, exclusive, or auto-delete queues** always use **classic** queue type, since **quorum** queues and **streams** are always durable and not designed for temporary or connection-bound queues. 
+
+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. 
+
 ### Changelog
 
 ### New in v5.4.x

test/queue.test.ts

@@ -47,7 +47,7 @@ describe('Test Queue class', function () {
         deadLetterExchange: undefined,
         deadLetterRoutingKey: undefined,
         maxLength: undefined,
-        arguments: {}
+        arguments: { 'x-queue-type': 'classic' }
       })
       .should.be.true();
   });

test/rabbit.test.ts

@@ -194,6 +194,51 @@ describe('Test rabbit class', function() {
     options!.arguments.should.not.have.property('x-max-priority');
   });
 
+  it('should set queue type to classic when creating a non durable 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, { durable: false }, 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['x-queue-type'].should.equals('classic');
+  });
+
+  it('should set queue type to classic when creating an exclusive 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, { exclusive: true }, 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['x-queue-type'].should.equals('classic');
+  });
+
+  it('should set queue type to classic when creating an auto delete 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, { autoDelete: true }, 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['x-queue-type'].should.equals('classic');
+  });
+
   it('should unsubscribe', async function() {
     const stub = sandbox.stub(Queue.prototype, 'unsubscribe');
     rabbit = new Rabbit(this.url);

ts/queue.ts

@@ -54,6 +54,10 @@ export default class Queue {
     if (priority !== undefined) {
       queueOptions.arguments = { ...queueOptions.arguments, 'x-max-priority': priority };
     }
+    if (!durable || exclusive || autoDelete) {
+      queueOptions.arguments ??= {};
+      queueOptions.arguments['x-queue-type'] = 'classic';
+    }
     if (queueOptions.arguments?.['x-max-priority'] &&
       queueOptions.arguments?.['x-queue-type'] === 'quorum') {
       logger.warn(`Invalid x-max-priority argument for quorum queue '${this.name}'`);