Skip to content

Commit b670d48

Browse files
snatalenkosnatalenko
committed
Merge commit 'cd5773ae233076ae20eaab264c0a2c9aa14c29b6' into beta
2 parents 945aaa6 + cd5773a commit b670d48

3 files changed

Lines changed: 192 additions & 2 deletions

File tree

‎src/rabbitmq/RabbitMqGateway.ts‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -441,6 +441,7 @@ export class RabbitMqGateway {
441441
queueName,
442442
eventType,
443443
queueExpires,
444+
handlerProcessTimeout,
444445
singleActiveConsumer
445446
} = subscription;
446447

@@ -479,7 +480,8 @@ export class RabbitMqGateway {
479480
const reply = await this.#assetQueue(channel, exchange, queueGivenName, eventType, {
480481
deadLetterExchangeName,
481482
...singleActiveConsumer && { singleActiveConsumer },
482-
...queueExpires && { queueExpires }
483+
...queueExpires && { queueExpires },
484+
handlerProcessTimeout
483485
});
484486
messageCount = reply.messageCount;
485487
consumerCount = reply.consumerCount;
@@ -557,13 +559,17 @@ export class RabbitMqGateway {
557559

558560
/** Auto-delete the queue after this many ms of idleness (no consumers, no new messages) */
559561
queueExpires?: number,
562+
563+
/** Handler process timeout used to derive the per-queue `x-consumer-timeout` argument */
564+
handlerProcessTimeout?: number,
560565
}) {
561566
const {
562567
durable = true,
563568
exclusive = false,
564569
singleActiveConsumer = false,
565570
deadLetterExchangeName,
566-
queueExpires
571+
queueExpires,
572+
handlerProcessTimeout = RabbitMqGateway.HANDLER_PROCESS_TIMEOUT
567573
} = options ?? {};
568574

569575
await channel.assertExchange(exchange, 'topic', { durable: true });
@@ -583,6 +589,9 @@ export class RabbitMqGateway {
583589
},
584590
...singleActiveConsumer && {
585591
'x-single-active-consumer': true
592+
},
593+
...handlerProcessTimeout && {
594+
'x-consumer-timeout': handlerProcessTimeout + 1_000
586595
}
587596
}
588597
});

‎tests/integration/rabbitmq/RabbitMqGateway.test.ts‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -370,6 +370,63 @@ describe('RabbitMqGateway', () => {
370370

371371
expect(exists).toBe(false);
372372
}, 20_000);
373+
374+
it('sets x-consumer-timeout on durable queue when handlerProcessTimeout is specified', async () => {
375+
const customTimeout = 30_000;
376+
const handler = jest.fn();
377+
378+
await gateway1.subscribe({
379+
exchange,
380+
queueName,
381+
eventType: 'timeout.test',
382+
handler,
383+
handlerProcessTimeout: customTimeout
384+
});
385+
386+
const res = await fetch(`http://localhost:15672/api/queues/%2F/${encodeURIComponent(queueName)}`, {
387+
headers: { Authorization: `Basic ${btoa('guest:guest')}` }
388+
});
389+
const queue: any = await res.json();
390+
391+
expect(queue.arguments['x-consumer-timeout']).toBe(customTimeout + 1_000);
392+
});
393+
394+
it('sets default x-consumer-timeout on durable queue when handlerProcessTimeout is not specified', async () => {
395+
const handler = jest.fn();
396+
397+
await gateway1.subscribe({
398+
exchange,
399+
queueName,
400+
eventType: 'timeout.default',
401+
handler
402+
});
403+
404+
const res = await fetch(`http://localhost:15672/api/queues/%2F/${encodeURIComponent(queueName)}`, {
405+
headers: { Authorization: `Basic ${btoa('guest:guest')}` }
406+
});
407+
const queue: any = await res.json();
408+
409+
expect(queue.arguments['x-consumer-timeout']).toBe(RabbitMqGateway.HANDLER_PROCESS_TIMEOUT + 1_000);
410+
});
411+
412+
it('does not set x-consumer-timeout when handlerProcessTimeout is 0', async () => {
413+
const handler = jest.fn();
414+
415+
await gateway1.subscribe({
416+
exchange,
417+
queueName,
418+
eventType: 'timeout.disabled',
419+
handler,
420+
handlerProcessTimeout: 0
421+
});
422+
423+
const res = await fetch(`http://localhost:15672/api/queues/%2F/${encodeURIComponent(queueName)}`, {
424+
headers: { Authorization: `Basic ${btoa('guest:guest')}` }
425+
});
426+
const queue: any = await res.json();
427+
428+
expect(queue.arguments['x-consumer-timeout']).toBeUndefined();
429+
});
373430
});
374431

375432
describe('subscribe', () => {

‎tests/unit/rabbitmq/RabbitMqGateway.test.ts‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
import { RabbitMqGateway } from '../../../src/rabbitmq/RabbitMqGateway.ts';
2+
import { EventEmitter } from 'stream';
3+
4+
function createMockChannel() {
5+
const assertQueueCalls: any[] = [];
6+
return {
7+
assertQueueCalls,
8+
assertExchange: jest.fn().mockResolvedValue(undefined),
9+
assertQueue: jest.fn().mockImplementation((_name, options) => {
10+
assertQueueCalls.push(options);
11+
return Promise.resolve({ queue: 'test-queue', messageCount: 0, consumerCount: 0 });
12+
}),
13+
bindQueue: jest.fn().mockResolvedValue(undefined),
14+
prefetch: jest.fn().mockResolvedValue(undefined),
15+
consume: jest.fn().mockResolvedValue({ consumerTag: 'tag-1' }),
16+
on: jest.fn()
17+
};
18+
}
19+
20+
function createMockConnection(channel: ReturnType<typeof createMockChannel>) {
21+
return {
22+
createChannel: jest.fn().mockResolvedValue(channel),
23+
createConfirmChannel: jest.fn().mockResolvedValue({
24+
on: jest.fn(),
25+
publish: jest.fn()
26+
}),
27+
on: jest.fn(),
28+
close: jest.fn().mockResolvedValue(undefined)
29+
};
30+
}
31+
32+
describe('RabbitMqGateway', () => {
33+
34+
describe('x-consumer-timeout', () => {
35+
36+
let channel: ReturnType<typeof createMockChannel>;
37+
let connection: ReturnType<typeof createMockConnection>;
38+
let gateway: RabbitMqGateway;
39+
40+
beforeEach(() => {
41+
channel = createMockChannel();
42+
connection = createMockConnection(channel);
43+
44+
gateway = new RabbitMqGateway({
45+
rabbitMqConnectionFactory: () => Promise.resolve(connection as any),
46+
process: new EventEmitter() as any
47+
});
48+
});
49+
50+
afterEach(async () => {
51+
await gateway.disconnect();
52+
});
53+
54+
it('sets x-consumer-timeout to handlerProcessTimeout + 1000 on durable queue', async () => {
55+
await gateway.subscribe({
56+
exchange: 'test-exchange',
57+
queueName: 'test-queue',
58+
eventType: 'test.event',
59+
handler: jest.fn(),
60+
handlerProcessTimeout: 30_000
61+
});
62+
63+
// assertQueue is called twice for durable queues: first for the dead letter queue, then for the main queue.
64+
// The main queue call has x-queue-type: 'quorum' and x-dead-letter-exchange.
65+
const durableQueueCall = channel.assertQueueCalls.find(
66+
(opts: any) => opts.arguments?.['x-dead-letter-exchange']
67+
);
68+
69+
expect(durableQueueCall).toBeDefined();
70+
expect(durableQueueCall.arguments['x-consumer-timeout']).toBe(31_000);
71+
});
72+
73+
it('sets x-consumer-timeout to default HANDLER_PROCESS_TIMEOUT + 1000 when not specified', async () => {
74+
await gateway.subscribe({
75+
exchange: 'test-exchange',
76+
queueName: 'test-queue',
77+
eventType: 'test.event',
78+
handler: jest.fn()
79+
});
80+
81+
const durableQueueCall = channel.assertQueueCalls.find(
82+
(opts: any) => opts.arguments?.['x-dead-letter-exchange']
83+
);
84+
85+
expect(durableQueueCall).toBeDefined();
86+
expect(durableQueueCall.arguments['x-consumer-timeout']).toBe(
87+
RabbitMqGateway.HANDLER_PROCESS_TIMEOUT + 1_000
88+
);
89+
});
90+
91+
it('does not set x-consumer-timeout when handlerProcessTimeout is 0', async () => {
92+
await gateway.subscribe({
93+
exchange: 'test-exchange',
94+
queueName: 'test-queue',
95+
eventType: 'test.event',
96+
handler: jest.fn(),
97+
handlerProcessTimeout: 0
98+
});
99+
100+
const durableQueueCall = channel.assertQueueCalls.find(
101+
(opts: any) => opts.arguments?.['x-dead-letter-exchange']
102+
);
103+
104+
expect(durableQueueCall).toBeDefined();
105+
expect(durableQueueCall.arguments['x-consumer-timeout']).toBeUndefined();
106+
});
107+
108+
it('sets default x-consumer-timeout on exclusive (fanout) queues', async () => {
109+
await gateway.subscribe({
110+
exchange: 'test-exchange',
111+
handler: jest.fn()
112+
});
113+
114+
const exclusiveQueueCall = channel.assertQueueCalls.find(
115+
(opts: any) => opts.exclusive === true
116+
);
117+
118+
expect(exclusiveQueueCall).toBeDefined();
119+
expect(exclusiveQueueCall.arguments['x-consumer-timeout']).toBe(
120+
RabbitMqGateway.HANDLER_PROCESS_TIMEOUT + 1_000
121+
);
122+
});
123+
});
124+
});

0 commit comments

Comments
 (0)