feat: add cancelConsumersOnSigterm option (DS-3369)

README.md

@@ -31,6 +31,7 @@ const rabbit = new Rabbit(process.env.RABBIT_URL || 'amqp://localhost', {
   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)
+  cancelConsumersOnSigterm: true, //cancel every consumer when the process receives SIGTERM (see Shutdown)
   socketOptions: {} // socketOptions will be passed as a second param to amqp.connect and from ther to the socket library (net or tls)
 });
 
@@ -280,8 +281,19 @@ Quorum queues need RabbitMQ 3.10 or later for per-message TTL and at-least-once
 
 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`.
 
+### Shutdown
+
+On `SIGTERM`, rabbit-queue cancels the consumers of every queue it created, so the process stops taking new messages while it shuts down. Set `cancelConsumersOnSigterm: false` when something else in the process has its own `SIGTERM` hook that waits on messages from these queues, e.g. eip-redis's PubSub waiting for throttled events that only complete when an eip-rabbit Timer queue delivers. With the consumers cancelled, that wait never ends and the process hangs until it is killed. On by default.
+```javascript
+const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: false });
+```
+
 ### Changelog
 
+### New in v5.11.x
+
+- `cancelConsumersOnSigterm` option. Set it to `false` to keep consuming after `SIGTERM`, for processes that drain work through their own `SIGTERM` hook. On by default.
+
 ### 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.

test/rabbit.test.ts

@@ -385,6 +385,20 @@ describe('Test rabbit class', function() {
     process.listeners('SIGTERM').should.not.containEql(handler);
   });
 
+  it('should keep consumers on SIGTERM when cancelConsumersOnSigterm is false', async function() {
+    const stub = sandbox.stub(Queue.prototype, 'unsubscribe').resolves();
+    const listenersBefore = process.listeners('SIGTERM');
+
+    rabbit = new Rabbit(this.url, { cancelConsumersOnSigterm: false });
+    await rabbit.connected;
+    await rabbit.createQueue(randomUUID().toString(), { exclusive: true }, () => ({}));
+
+    const added = process.listeners('SIGTERM').filter(l => !listenersBefore.includes(l));
+    added.should.be.empty();
+    added.forEach(l => (l as () => void)());
+    stub.called.should.be.false();
+  });
+
   it('should call unsubscribe for all queues on SIGTERM', async function() {
     const stub = sandbox.stub(Queue.prototype, 'unsubscribe').resolves();
 

ts/rabbit.ts

@@ -42,6 +42,7 @@ export default class Rabbit extends EventEmitter {
       defaultQueueType = undefined,
       scheduledPublishQueuesAsQuorum = false,
       tolerateQueueTypeMismatch = false,
+      cancelConsumersOnSigterm = true,
     }: {
       prefetch?: number
       replyPattern?: boolean
@@ -51,6 +52,7 @@ export default class Rabbit extends EventEmitter {
       defaultQueueType?: string
       scheduledPublishQueuesAsQuorum?: boolean
       tolerateQueueTypeMismatch?: boolean
+      cancelConsumersOnSigterm?: boolean
     } = {}
   ) {
     super();
@@ -70,7 +72,9 @@ export default class Rabbit extends EventEmitter {
     this.sigtermHandler = this.handleSigterm.bind(this);
     this.reconnect();
 
-    process.once('SIGTERM', this.sigtermHandler);
+    if (cancelConsumersOnSigterm) {
+      process.once('SIGTERM', this.sigtermHandler);
+    }
   }
 
   private async connect() {