diff --git a/README.md b/README.md index c68e707..6f24401 100644 --- a/README.md +++ b/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. diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index 72a446e..be4bd11 100644 --- a/test/rabbit.test.ts +++ b/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(); diff --git a/ts/rabbit.ts b/ts/rabbit.ts index 83104d7..96216fc 100644 --- a/ts/rabbit.ts +++ b/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() {