Change: Disable DLQ setup for RabbitMqEventBus by default

Author: snatalenko

src/rabbitmq/RabbitMqCommandBus.ts

@@ -5,7 +5,7 @@ import { RabbitMqGateway, type Subscription } from './RabbitMqGateway.ts';
 import { type ConfigProvider, resolveProvider } from './utils/index.ts';
 
 export type RabbitMqCommandBusConfig = Partial<Pick<Subscription,
-	'exchange' | 'queueName' | 'ignoreOwn' | 'concurrentLimit' | 'handlerProcessTimeout' | 'queueExpires'>>;
+	'exchange' | 'queueName' | 'ignoreOwn' | 'concurrentLimit' | 'handlerProcessTimeout' | 'queueExpires' | 'deadLetterQueue'>>;
 
 type ResolvedRabbitMqCommandBusConfig = RabbitMqCommandBusConfig
 	& Required<Pick<RabbitMqCommandBusConfig, 'exchange' | 'queueName' | 'ignoreOwn'>>;
@@ -18,7 +18,8 @@ async function resolveConfig(provider?: ConfigProvider<RabbitMqCommandBusConfig>
 		concurrentLimit,
 		handlerProcessTimeout,
 		queueName,
-		queueExpires
+		queueExpires,
+		deadLetterQueue
 	} = await resolveProvider(provider) ?? {};
 
 	assertString(exchange, 'rabbitMqCommandConfig.exchange');
@@ -30,8 +31,10 @@ async function resolveConfig(provider?: ConfigProvider<RabbitMqCommandBusConfig>
 		assertNonNegativeInteger(handlerProcessTimeout, 'rabbitMqCommandConfig.handlerProcessTimeout');
 	if (queueExpires !== undefined)
 		assertNonNegativeInteger(queueExpires, 'rabbitMqCommandConfig.queueExpires');
+	if (deadLetterQueue !== undefined)
+		assertBoolean(deadLetterQueue, 'rabbitMqCommandConfig.deadLetterQueue');
 
-	return { exchange, queueName, ignoreOwn, concurrentLimit, handlerProcessTimeout, queueExpires };
+	return { exchange, queueName, ignoreOwn, concurrentLimit, handlerProcessTimeout, queueExpires, deadLetterQueue };
 }
 
 /**
@@ -99,18 +102,12 @@ export class RabbitMqCommandBus implements ICommandBus {
 	 * Only one consumer receives each message.
 	 */
 	async on(messageType: string, handler: IMessageHandler): Promise<void> {
-		const { exchange, queueName, ignoreOwn, concurrentLimit, handlerProcessTimeout, queueExpires } =
-			await this.#resolveConfig();
+		const subscriptionParams = await this.#resolveConfig();
 
 		await this.#gateway.subscribe({
-			exchange,
-			queueName,
+			...subscriptionParams,
 			eventType: messageType,
-			handler,
-			ignoreOwn,
-			concurrentLimit,
-			handlerProcessTimeout,
-			queueExpires
+			handler
 		});
 	}
 

src/rabbitmq/RabbitMqEventBus.ts

@@ -6,7 +6,7 @@ import { RabbitMqGateway, type Subscription, type SubscribeResult } from './Rabb
 import { type ConfigProvider, resolveProvider } from './utils/index.ts';
 
 export type RabbitMqEventBusConfig = Partial<Pick<Subscription,
-	'exchange' | 'queueName' | 'ignoreOwn' | 'concurrentLimit' | 'handlerProcessTimeout' | 'queueExpires'>>;
+	'exchange' | 'queueName' | 'ignoreOwn' | 'concurrentLimit' | 'handlerProcessTimeout' | 'queueExpires' | 'deadLetterQueue'>>;
 
 type ResolvedRabbitMqEventBusConfig = RabbitMqEventBusConfig
 	& Required<Pick<RabbitMqEventBusConfig, 'exchange' | 'ignoreOwn'>>;
@@ -19,7 +19,8 @@ async function resolveConfig(provider?: ConfigProvider<RabbitMqEventBusConfig>)
 		concurrentLimit,
 		handlerProcessTimeout,
 		queueName,
-		queueExpires
+		queueExpires,
+		deadLetterQueue = false
 	} = await resolveProvider(provider) ?? {};
 
 	assertString(exchange, 'rabbitMqEventConfig.exchange');
@@ -32,8 +33,9 @@ async function resolveConfig(provider?: ConfigProvider<RabbitMqEventBusConfig>)
 		assertString(queueName, 'rabbitMqEventConfig.queueName');
 	if (queueExpires !== undefined)
 		assertNonNegativeInteger(queueExpires, 'rabbitMqEventConfig.queueExpires');
+	assertBoolean(deadLetterQueue, 'rabbitMqEventConfig.deadLetterQueue');
 
-	return { exchange, ignoreOwn, concurrentLimit, handlerProcessTimeout, queueName, queueExpires };
+	return { exchange, ignoreOwn, concurrentLimit, handlerProcessTimeout, queueName, queueExpires, deadLetterQueue };
 }
 
 /**
@@ -43,6 +45,10 @@ async function resolveConfig(provider?: ConfigProvider<RabbitMqEventBusConfig>)
  * By default uses an exclusive (non-durable) queue per connection.
  * Set `queueName` in config for a durable queue that survives restarts.
  *
+ * Dead letter queue is disabled by default (`deadLetterQueue: false`).
+ * When a handler fails or times out, the message is discarded.
+ * Set `deadLetterQueue: true` to route rejected messages to a `.failed` queue.
+ *
  * Optionally ignores messages published by this instance (default: true).
  *
  * Supports named durable queues via {@link queue} for single-consumer delivery.
@@ -92,18 +98,12 @@ export class RabbitMqEventBus implements IEventBus, IObservableQueueProvider {
 	 * (non-durable) queue that is deleted on disconnect.
 	 */
 	async on(eventType: string, handler: IMessageHandler): Promise<SubscribeResult> {
-		const { exchange, queueName, ignoreOwn, concurrentLimit, handlerProcessTimeout, queueExpires } =
-			await this.#resolveConfig();
+		const subscriptionParams = await this.#resolveConfig();
 
 		return this.#gateway.subscribe({
-			exchange,
-			queueName,
+			...subscriptionParams,
 			eventType,
 			handler,
-			ignoreOwn,
-			concurrentLimit,
-			handlerProcessTimeout,
-			queueExpires,
 			singleActiveConsumer: true
 		});
 	}
@@ -133,15 +133,16 @@ export class RabbitMqEventBus implements IEventBus, IObservableQueueProvider {
 			queue = new RabbitMqCommandBus({
 				rabbitMqGateway: this.#gateway,
 				rabbitMqCommandBusConfig: async () => {
-					const { exchange, concurrentLimit, handlerProcessTimeout, queueExpires } =
+					const { exchange, concurrentLimit, handlerProcessTimeout, queueExpires, deadLetterQueue } =
 						await this.#resolveConfig();
 
 					return {
 						exchange,
 						queueName,
 						concurrentLimit,
 						handlerProcessTimeout,
-						queueExpires
+						queueExpires,
+						deadLetterQueue
 					};
 				}
 			});

src/rabbitmq/RabbitMqGateway.ts

@@ -65,6 +65,13 @@ export type Subscription = {
 	 * Enables RabbitMQ single active consumer mode (`x-single-active-consumer`).
 	 */
 	singleActiveConsumer?: boolean;
+
+	/**
+	 * Whether to create a dead letter queue for rejected or timed out messages.
+	 * Only applicable to durable queues (when `queueName` is set).
+	 * Defaults to `true` when `queueName` is set, `false` otherwise.
+	 */
+	deadLetterQueue?: boolean;
 };
 
 export type SubscribeResult = {
@@ -442,9 +449,13 @@ export class RabbitMqGateway {
 			eventType,
 			queueExpires,
 			handlerProcessTimeout,
-			singleActiveConsumer
+			singleActiveConsumer,
+			deadLetterQueue = !!queueName
 		} = subscription;
 
+		if (deadLetterQueue && !queueName)
+			throw new Error('deadLetterQueue requires a durable queue (queueName must be set)');
+
 		const channel = await this.#assertQueueChannel(queueName);
 
 		let queueGivenName = queueName;
@@ -471,14 +482,17 @@ export class RabbitMqGateway {
 		}
 		else {
 			// Handle durable queue case
-			const deadLetterExchangeName = `${exchange}.failed`;
+			let deadLetterExchangeName: string | undefined;
+			if (deadLetterQueue) {
+				deadLetterExchangeName = `${exchange}.failed`;
 
-			// Assert dead letter queue for rejected or timed out messages
-			await this.#assetQueue(channel, deadLetterExchangeName, `${queueGivenName}.failed`);
+				// Assert dead letter queue for rejected or timed out messages
+				await this.#assetQueue(channel, deadLetterExchangeName, `${queueGivenName}.failed`);
+			}
 
 			// Assert durable queue that will survive broker restart
 			const reply = await this.#assetQueue(channel, exchange, queueGivenName, eventType, {
-				deadLetterExchangeName,
+				...deadLetterExchangeName && { deadLetterExchangeName },
 				...singleActiveConsumer && { singleActiveConsumer },
 				...queueExpires && { queueExpires },
 				handlerProcessTimeout

tests/integration/rabbitmq/RabbitMqGateway.test.ts

@@ -227,6 +227,33 @@ describe('RabbitMqGateway', () => {
 			await cn2.close();
 		});
 
+		it('does not create DLQ when deadLetterQueue is false', async () => {
+			await gateway1.subscribe({
+				exchange,
+				queueName,
+				eventType: 'dlq.disabled',
+				handler: _msg => {
+					throw new Error('intentional failure');
+				},
+				deadLetterQueue: false
+			});
+
+			const message: IMessage = {
+				type: 'dlq.disabled',
+				payload: { shouldFail: true }
+			};
+
+			await gateway1.publish(exchange, message);
+			await delay(50);
+
+			// Verify the .failed queue does not exist
+			const res = await fetch(`http://localhost:15672/api/queues/%2F/${encodeURIComponent(`${queueName}.failed`)}`, {
+				headers: { Authorization: `Basic ${btoa('guest:guest')}` }
+			});
+
+			expect(res.status).toBe(404);
+		});
+
 		it('does not ack a message after timeout and logs it while consumer keeps processing', async () => {
 			const errorLogs: Array<{ message: string, meta?: { [key: string]: any } }> = [];
 			const logger: ILogger = {

tests/unit/rabbitmq/RabbitMqBuses.test.ts

@@ -37,6 +37,7 @@ describe('RabbitMq bus config forwarding', () => {
 			concurrentLimit: 2,
 			handlerProcessTimeout: 1234,
 			queueExpires: 5678,
+			deadLetterQueue: false,
 			singleActiveConsumer: true
 		});
 	});
@@ -61,6 +62,7 @@ describe('RabbitMq bus config forwarding', () => {
 			concurrentLimit: undefined,
 			handlerProcessTimeout: undefined,
 			queueExpires: undefined,
+			deadLetterQueue: false,
 			singleActiveConsumer: true
 		});
 	});
@@ -88,7 +90,8 @@ describe('RabbitMq bus config forwarding', () => {
 			ignoreOwn: false,
 			concurrentLimit: 3,
 			handlerProcessTimeout: 2345,
-			queueExpires: 6789
+			queueExpires: 6789,
+			deadLetterQueue: undefined
 		});
 	});
 
@@ -114,7 +117,61 @@ describe('RabbitMq bus config forwarding', () => {
 			ignoreOwn: false,
 			concurrentLimit: 4,
 			handlerProcessTimeout: 3456,
-			queueExpires: 7890
+			queueExpires: 7890,
+			deadLetterQueue: false
 		});
 	});
+
+	it('forwards deadLetterQueue: false from event bus config to subscribe()', async () => {
+		const rabbitMqGateway = createGateway();
+		const bus = new RabbitMqEventBus({
+			rabbitMqGateway: rabbitMqGateway as any,
+			rabbitMqEventBusConfig: {
+				exchange: 'events',
+				queueName: 'event-queue',
+				deadLetterQueue: false
+			}
+		});
+
+		await bus.on('test.event', handler);
+
+		expect(rabbitMqGateway.subscribe).toHaveBeenCalledWith(
+			expect.objectContaining({ deadLetterQueue: false })
+		);
+	});
+
+	it('forwards deadLetterQueue: false from command bus config to subscribe()', async () => {
+		const rabbitMqGateway = createGateway();
+		const bus = new RabbitMqCommandBus({
+			rabbitMqGateway: rabbitMqGateway as any,
+			rabbitMqCommandBusConfig: {
+				exchange: 'commands',
+				queueName: 'command-queue',
+				deadLetterQueue: false
+			}
+		});
+
+		await bus.on('test.command', handler);
+
+		expect(rabbitMqGateway.subscribe).toHaveBeenCalledWith(
+			expect.objectContaining({ deadLetterQueue: false })
+		);
+	});
+
+	it('forwards deadLetterQueue from event bus queue() config to subscribe()', async () => {
+		const rabbitMqGateway = createGateway();
+		const bus = new RabbitMqEventBus({
+			rabbitMqGateway: rabbitMqGateway as any,
+			rabbitMqEventBusConfig: {
+				exchange: 'events',
+				deadLetterQueue: false
+			}
+		});
+
+		await bus.queue('shared-queue').on('test.event', handler);
+
+		expect(rabbitMqGateway.subscribe).toHaveBeenCalledWith(
+			expect.objectContaining({ deadLetterQueue: false })
+		);
+	});
 });