Skip to content

Commit 957c355

Browse files
AkamaozuAkamaozu
authored
Merge pull request #8 from Akamaozu/improve-reconnection-tests
Improve Reconnection Tests
2 parents f653ad9 + 17295a5 commit 957c355

File tree

1 file changed

+79
-64
lines changed

1 file changed

+79
-64
lines changed

‎test/reconnection.test.js

+79-64
Original file line numberDiff line numberDiff line change
@@ -6,15 +6,20 @@ const Sinon = require('sinon');
66
const EventEmitter = require('events');
77

88
const { after, beforeEach, describe, it } = require('mocha');
9+
const {
10+
RABBIT_RECONNECTION_TIMEOUT: ENV_RABBIT_RECONNECTION_TIMEOUT,
11+
RABBIT_URL: ENV_RABBIT_URL,
12+
} = process.env
913

1014
const AMQP_PORT = 8080;
15+
const RABBIT_RECONNECTION_TIMEOUT = ENV_RABBIT_RECONNECTION_TIMEOUT && parseInt( ENV_RABBIT_RECONNECTION_TIMEOUT )
1116

1217
describe('reconnection', () => {
1318

1419
let RECONN_TIMEOUT;
1520
let RECONN_RETRIES;
1621
let EXACT_TIMEOUT;
17-
const STUB_RABBIT_URL = `amqp://localhost:${AMQP_PORT}`;
22+
const STUB_RABBIT_URL = ENV_RABBIT_URL ?? `amqp://localhost:${AMQP_PORT}`;
1823

1924
let AmqpStub;
2025
const Amqp = require('amqplib/callback_api');
@@ -25,7 +30,7 @@ describe('reconnection', () => {
2530
Sinon.stub(process, 'exit');
2631
AmqpStub = Sinon.stub(Amqp, 'connect');
2732

28-
RECONN_TIMEOUT = 4;
33+
RECONN_TIMEOUT = RABBIT_RECONNECTION_TIMEOUT ?? 4;
2934
RECONN_RETRIES = 5;
3035
EXACT_TIMEOUT = false;
3136

@@ -38,6 +43,34 @@ describe('reconnection', () => {
3843
done();
3944
});
4045

46+
const genQueueName = () => {
47+
48+
const chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-';
49+
let label = 'amq.gen-';
50+
for (let i = 22; i > 0; --i) {
51+
label += chars[Math.floor(Math.random() * chars.length)];
52+
}
53+
54+
return label;
55+
};
56+
57+
const verifyConnection = (conn, conn_args = {}) => {
58+
59+
const {
60+
exchangeCount = 0,
61+
queueCount = 0,
62+
consumerCount = conn_args?.queueCount ?? 0,
63+
uniqueKeys = conn_args?.queueCount ?? 0,
64+
} = conn_args
65+
66+
Assert.strictEqual(conn.createChannel.callCount, exchangeCount + queueCount);
67+
Assert.strictEqual(conn.assertExchange.callCount, exchangeCount);
68+
Assert.strictEqual(conn.prefetch.callCount, queueCount);
69+
Assert.strictEqual(conn.assertQueue.callCount, queueCount);
70+
Assert.strictEqual(conn.bindQueue.callCount, uniqueKeys);
71+
Assert.strictEqual(conn.consume.callCount, consumerCount);
72+
};
73+
4174
const mockLogger = () => {
4275

4376
let idx = 0;
@@ -74,6 +107,21 @@ describe('reconnection', () => {
74107
});
75108
};
76109

110+
const mockConnection = () => {
111+
112+
const conn = new EventEmitter();
113+
114+
conn.createChannel = Sinon.stub().callsFake((cb) => cb(null, conn));
115+
conn.assertExchange = Sinon.stub().callsFake((exchange, type, options, cb) => setTimeout(cb, 1));
116+
conn.prefetch = Sinon.stub();
117+
conn.assertQueue = Sinon.stub().callsFake((queue, options, cb) => setTimeout(() => cb(null, { queue: queue || genQueueName() }), 1));
118+
conn.bindQueue = Sinon.stub().callsFake((queue, exchange, routingKey, options, cb) => setTimeout(cb, 1));
119+
conn.consume = Sinon.stub();
120+
conn.publish = Sinon.stub().callsFake((exchange, routingKey, content, options) => setTimeout(() => {}, 1));
121+
122+
return conn;
123+
};
124+
77125
const mockRabbitServer = async ({ logger = undefined, stub, rabbit = undefined, addQueues = undefined, isReconnecting = undefined, exchangeCount = 0, queueCount = 0, consumerCount = queueCount, uniqueKeys = queueCount }) => {
78126

79127
if (!stub) {
@@ -90,43 +138,7 @@ describe('reconnection', () => {
90138
}
91139

92140
Assert.strictEqual(stub.callCount, 1);
93-
Assert.strictEqual(stub.args[0][0], `amqp://localhost:${AMQP_PORT}`);
94-
95-
const genQueueName = () => {
96-
97-
const chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-';
98-
let label = 'amq.gen-';
99-
for (let i = 22; i > 0; --i) {
100-
label += chars[Math.floor(Math.random() * chars.length)];
101-
}
102-
103-
return label;
104-
};
105-
106-
const mockConnection = () => {
107-
108-
const conn = new EventEmitter();
109-
110-
conn.createChannel = Sinon.stub().callsFake((cb) => cb(null, conn));
111-
conn.assertExchange = Sinon.stub().callsFake((exchange, type, options, cb) => setTimeout(cb, 1));
112-
conn.prefetch = Sinon.stub();
113-
conn.assertQueue = Sinon.stub().callsFake((queue, options, cb) => setTimeout(() => cb(null, { queue: queue || genQueueName() }), 1));
114-
conn.bindQueue = Sinon.stub().callsFake((queue, exchange, routingKey, options, cb) => setTimeout(cb, 1));
115-
conn.consume = Sinon.stub();
116-
conn.publish = Sinon.stub().callsFake((exchange, routingKey, content, options) => setTimeout(() => {}, 1));
117-
118-
return conn;
119-
};
120-
121-
const verifyConnection = (conn) => {
122-
123-
Assert.strictEqual(conn.createChannel.callCount, exchangeCount + queueCount);
124-
Assert.strictEqual(conn.assertExchange.callCount, exchangeCount);
125-
Assert.strictEqual(conn.prefetch.callCount, queueCount);
126-
Assert.strictEqual(conn.assertQueue.callCount, queueCount);
127-
Assert.strictEqual(conn.bindQueue.callCount, uniqueKeys);
128-
Assert.strictEqual(conn.consume.callCount, consumerCount);
129-
};
141+
Assert.strictEqual(stub.args[0][0], STUB_RABBIT_URL);
130142

131143
const onConnect = waitEvent(rabbit, reconnecting ? 'reconnected' : 'connected');
132144
const conn = mockConnection();
@@ -137,7 +149,12 @@ describe('reconnection', () => {
137149

138150
await onConnect;
139151

140-
verifyConnection(conn);
152+
verifyConnection( conn, {
153+
queueCount,
154+
exchangeCount,
155+
consumerCount,
156+
uniqueKeys,
157+
});
141158

142159
return { conn, rabbit, opts };
143160
};
@@ -290,34 +307,32 @@ describe('reconnection', () => {
290307
RECONN_RETRIES = 4;
291308

292309
const logger = mockLogger();
293-
const { conn, rabbit, opts } = await mockRabbitServer({ logger, stub: AmqpStub });
294-
const asyncReconn = waitEvent(rabbit, 'reconnecting', RECONN_TIMEOUT * .6); // immediate reconnection
310+
const { conn, rabbit, opts } = await mockRabbitServer({ stub: AmqpStub, logger });
295311

296-
const lostConnection = new Error('Connection to RabbitMQ lost');
297-
lostConnection.code = 320;
312+
let reconnect_attempts = 0
313+
let rabbit_error = 0
298314

299-
rabbitStartError(AmqpStub);
300-
conn.emit('close', lostConnection);
315+
rabbit.on('reconnecting', () => {
316+
reconnect_attempts += 1
301317

302-
await asyncReconn;
303-
await waitEvent(rabbit, 'reconnecting', Math.max(15, RECONN_TIMEOUT * 1.3)); // second attempt
304-
await waitEvent(rabbit, 'reconnecting', Math.max(15, RECONN_TIMEOUT * 1.3)); // third attempt
305-
await Promise.all([
306-
waitEvent(rabbit, 'reconnecting', Math.max(15, RECONN_TIMEOUT * 1.3)), // fourth attempt
307-
waitEvent(rabbit, 'error', Math.max(15, RECONN_TIMEOUT * 1.3)) // reconn error
308-
]);
318+
setTimeout(() => {
319+
const onConnectAttempt = AmqpStub.args[0][1]
320+
onConnectAttempt( new Error( 'whoops something went wrong while trying to connect' ) );
321+
}, 50)
322+
})
309323

310-
Assert.strictEqual(process.exit.callCount, 1);
311-
Assert.strictEqual(process.exit.args[0][0], 1);
324+
rabbit.on( 'error', () => {
325+
rabbit_error += 1
326+
})
312327

313-
// check logs
314-
logger.assert('warn', `Lost connection to RabbitMQ! Reconnecting in ${opts.reconnectionTimeout}ms...`);
315-
logger.assert('info', 'Reconnecting to RabbitMQ (1/4)...');
316-
logger.assert('info', 'Reconnecting to RabbitMQ (2/4)...');
317-
logger.assert('info', 'Reconnecting to RabbitMQ (3/4)...');
318-
logger.assert('info', 'Reconnecting to RabbitMQ (4/4)...');
319-
logger.assert('fatal', 'Rabbit connection error!');
328+
const waitForRabbitError = waitEvent(rabbit, 'error', 10_000);
329+
330+
conn.emit( 'close', new Error('Connection to RabbitMQ lost'), true);
331+
332+
await waitForRabbitError
320333

334+
Assert.strictEqual(reconnect_attempts, RECONN_RETRIES);
335+
Assert.strictEqual(rabbit_error, 1);
321336
});
322337

323338
it('Should reconnect once the connection is recovered', async () => {
@@ -381,7 +396,7 @@ describe('reconnection', () => {
381396

382397
rabbitStopError(AmqpStub);
383398

384-
await waitEvent(rabbit, 'reconnecting', Math.max(15, RECONN_TIMEOUT * 1.3)); // second attempt
399+
await waitEvent(rabbit, 'reconnecting', Math.max(5_000, RECONN_TIMEOUT * 1.3)); // second attempt
385400
const server = await mockRabbitServer({ logger, stub: AmqpStub, rabbit, exchangeCount: 1, queueCount: 1, uniqueKeys: 0, consumerCount: 0 });
386401
conn = server.conn;
387402

@@ -410,7 +425,7 @@ describe('reconnection', () => {
410425

411426
rabbitStopError(AmqpStub);
412427

413-
await waitEvent(rabbit, 'reconnecting', Math.max(15, RECONN_TIMEOUT * 1.3)); // second attempt
428+
await waitEvent(rabbit, 'reconnecting', Math.max(5_000, RECONN_TIMEOUT * 1.3)); // second attempt
414429

415430
let newServer = await mockRabbitServer({ logger, stub: AmqpStub, rabbit });
416431
conn = newServer.conn;

0 commit comments

Comments
 (0)