diff --git a/README.md b/README.md index 6f24401..27e24ea 100644 --- a/README.md +++ b/README.md @@ -293,6 +293,7 @@ const rabbit = new Rabbit(process.env.RABBIT_URL, { cancelConsumersOnSigterm: fa ### 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. +- v5.11.1: when only one of the two connections (consume or publish) dropped, reconnecting left the other one open, so its consumers kept running next to the ones created again on `connected`. The reconnect now closes it first, and one outage emits `disconnected` once instead of once per connection and channel. ### New in v5.10.x diff --git a/test/rabbit.test.ts b/test/rabbit.test.ts index be4bd11..fe19c8f 100644 --- a/test/rabbit.test.ts +++ b/test/rabbit.test.ts @@ -55,6 +55,18 @@ describe('Test rabbit class', function() { }); }); + it('should not emit "connected" when closed while connecting', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + const connected = sandbox.spy(); + rabbit.on('connected', connected); + + await rabbit.close(); + await rabbit.connected.should.be.rejectedWith(/closed while connecting/); + + connected.called.should.be.false(); + rabbit = null; + }); + it('should call amqp connect with socketOptions', async function() { const connectStub = sandbox.spy(amqp, 'connect'); rabbit = new Rabbit(this.url, { socketOptions: 'socketOptions' }); @@ -558,4 +570,89 @@ describe('Test rabbit class', function() { disconnected.called.should.be.false(); }); }); + + describe('when one of the two connections drops', function() { + let admin: amqp.ChannelModel; + let queueName: string; + + beforeEach(async function() { + admin = await amqp.connect(this.url); + queueName = `reconnect_${randomUUID()}`; + const channel = await admin.createChannel(); + await channel.assertQueue(queueName, { durable: true }); + await channel.close(); + }); + + afterEach(async function() { + rabbit?.removeAllListeners('connected'); + rabbit?.removeAllListeners('disconnected'); + const channel = await admin.createChannel(); + await channel.deleteQueue(queueName); + await admin.close(); + }); + + async function consumerCount() { + const channel = await admin.createChannel(); + const { consumerCount: count } = await channel.checkQueue(queueName); + await channel.close(); + return count; + } + + async function waitForConsumers(expected: number) { + for (let attempt = 0; attempt < 50 && (await consumerCount()) !== expected; attempt++) { + await new Promise(r => setTimeout(r, 20)); + } + (await consumerCount()).should.equal(expected); + } + + // What orka does: reconnect on every "disconnected" and create the handlers' queues again on "connected". + function resubscribeOnReconnect() { + rabbit.on('disconnected', () => setTimeout(() => rabbit.reconnect(), 10)); + rabbit.on('connected', () => rabbit.createQueue(queueName, {}, () => ({}))); + } + + it('should keep one consumer per queue after the publish connection drops', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + resubscribeOnReconnect(); + await rabbit.connected; + await waitForConsumers(1); + + const reconnected = new Promise(resolve => rabbit.once('connected', resolve)); + await rabbit.publishConnection.close(); + await reconnected; + await new Promise(r => setTimeout(r, 200)); + + (await consumerCount()).should.equal(1); + }); + + it('should close the surviving publish connection after the consume connection drops', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + resubscribeOnReconnect(); + await rabbit.connected; + await waitForConsumers(1); + const survivorClose = sandbox.spy(rabbit.publishConnection, 'close'); + + const reconnected = new Promise(resolve => rabbit.once('connected', resolve)); + await rabbit.consumeConnection.close(); + await reconnected; + await new Promise(r => setTimeout(r, 200)); + + survivorClose.called.should.be.true(); + (await consumerCount()).should.equal(1); + }); + + it('should emit "disconnected" once when a connection and its channel close', async function() { + rabbit = new Rabbit(this.url, { replyPattern: false }); + await rabbit.connected; + const disconnected = sandbox.spy(); + rabbit.on('disconnected', disconnected); + + await rabbit.publishConnection.close(); + await new Promise(r => setTimeout(r, 100)); + + disconnected.callCount.should.equal(1); + await rabbit.consumeConnection.close(); + rabbit = null; + }); + }); }); diff --git a/ts/rabbit.ts b/ts/rabbit.ts index 96216fc..bbae754 100644 --- a/ts/rabbit.ts +++ b/ts/rabbit.ts @@ -20,6 +20,11 @@ export default class Rabbit extends EventEmitter { public lock: Promise; public queues: { [s: string]: Queue } = {}; public connecting = false; + private attempt: Promise; + private connections = new Set(); + private lossReported = false; + private attemptError: Error | null = null; + private closed = false; public prefetch: number; public replyPattern: boolean; public prefix: string; @@ -77,35 +82,92 @@ export default class Rabbit extends EventEmitter { } } - private async connect() { - if (this.connecting) { - return; + private connect() { + if (!this.connecting) { + this.connecting = true; + this.attempt = this.openConnections(); + } + return this.attempt; + } + + private async openConnections() { + // Losing either connection triggers a reconnect, so the one that survived is closed here: otherwise its + // consumers keep running next to the ones the handlers create again on "connected". Not awaited, so a + // half-open connection still waiting for its close-ok doesn't hold up the new pair. + this.closeAll(this.connections); + const connections = new Set(); + this.connections = connections; + this.lossReported = false; + this.attemptError = null; + + try { + const consumeConnection = await this.openConnection(connections); + const consumeChannel = await this.createChannel(consumeConnection); + await this.initChannel(consumeChannel); + + const publishConnection = await this.openConnection(connections); + const publishChannel = await this.createChannel(publishConnection); + await this.initChannel(publishChannel, true); + + if (this.attemptError) { + throw this.attemptError; + } + this.consumeConnection = consumeConnection; + this.consumeChannel = consumeChannel; + this.publishConnection = publishConnection; + this.publishChannel = publishChannel; + } catch (error) { + this.closeAll(connections); + this.connecting = false; + this.reportLoss(error); + throw error; } - this.connecting = true; - this.consumeConnection = await amqp.connect(this.url, this.socketOptions); - this.consumeChannel = await this.createChannel(this.consumeConnection); - await this.initChannel(this.consumeChannel); - - this.publishConnection = await amqp.connect(this.url, this.socketOptions); - this.publishChannel = await this.createChannel(this.publishConnection); - await this.initChannel(this.publishChannel, true); - this.emit('connected'); this.connecting = false; + this.emit('connected'); } async reconnect() { + this.closed = false; this.connected = this.connect(); - await this.connected.catch(error => this.emitDisconnected(error)); + await this.connected.catch(() => undefined); } - private emitDisconnected(error) { - this.connecting = false; + private async openConnection(connections: Set) { + const connection = await amqp.connect(this.url, this.socketOptions); + connections.add(connection); + if (this.closed) { + throw new Error('Rabbit was closed while connecting'); + } + return connection; + } + + private closeAll(connections: Iterable) { + return Promise.all([...connections].map(connection => connection.close().catch(() => undefined))); + } + + private connectionLost(connection: amqp.ChannelModel, error) { + if (!this.connections.has(connection)) { + return; + } + if (this.connecting) { + this.attemptError = this.attemptError || error || new Error('Connection closed while connecting'); + return; + } + this.reportLoss(error); + } + + // One outage closes a connection and its channel, and may raise errors too: report it once. + private reportLoss(error) { + if (this.lossReported) { + return; + } + this.lossReported = true; this.emit('disconnected', error); } async createChannel(connection: amqp.ChannelModel) { - connection.once('close', error => this.emitDisconnected(error)); - connection.on('error', error => this.emitDisconnected(error)); + connection.once('close', error => this.connectionLost(connection, error)); + connection.on('error', error => this.connectionLost(connection, error)); const channel: Channel = await connection.createConfirmChannel(); channel.channelModel = connection; return channel; @@ -113,12 +175,12 @@ export default class Rabbit extends EventEmitter { async initChannel(channel: Channel, publish = false) { channel.prefetch(this.prefetch); - channel.on('close', error => this.emitDisconnected(error)); + channel.on('close', error => this.connectionLost(channel.channelModel, error)); if (!publish && this.replyPattern) { - await createReplyQueue(this.consumeChannel); + await createReplyQueue(channel); } if (!publish && this.scheduledPublish) { - await createDelayQueueReply(this.consumeChannel, this.updateName('delay'), this.scheduledPublishQueuesAsQuorum); + await createDelayQueueReply(channel, this.updateName('delay'), this.scheduledPublishQueuesAsQuorum); } } @@ -262,9 +324,9 @@ export default class Rabbit extends EventEmitter { async close() { process.off('SIGTERM', this.sigtermHandler); - - await this.consumeConnection.close(); - await this.publishConnection.close(); + this.closed = true; + // Covers a connect still in flight, whose connections aren't in consumeConnection/publishConnection yet. + await this.closeAll(this.connections); } private handleSigterm() {