[messaging-rabbitmq-extension] fix: channel wrapper avoid override

packages/messaging-rabbitmq-extension/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nestjstools/messaging-rabbitmq-extension",
-  "version": "4.2.0",
+  "version": "4.2.1",
   "description": "Extension to handle messages and dispatch them over AMQP protocol",
   "author": "Sebastian Iwanczyszyn",
   "license": "MIT",

packages/messaging-rabbitmq-extension/src/consumer/rabbitmq-messaging.consumer.ts

@@ -19,8 +19,8 @@ import { MessageDeadLetterVisitor } from './message-dead-letter.visitor';
 export class RabbitmqMessagingConsumer
   implements IMessagingConsumer<AmqpChannel>, OnModuleDestroy
 {
-  private channel?: AmqpChannel = undefined;
-  private amqpChannel: ChannelWrapper;
+  private readonly channelWrappers = new WeakMap<AmqpChannel, ChannelWrapper>();
+  private readonly channels = new Set<AmqpChannel>();
 
   constructor(
     private readonly rabbitMqMigrator: RabbitmqMigrator,
@@ -32,7 +32,7 @@ export class RabbitmqMessagingConsumer
     dispatcher: ConsumerMessageBus,
     channel: AmqpChannel,
   ): Promise<void> {
-    this.channel = channel;
+    this.channels.add(channel);
     await this.rabbitMqMigrator.run(channel);
 
     if (!channel.connection) {
@@ -43,10 +43,10 @@ export class RabbitmqMessagingConsumer
 
     const channelWrapper = channel.createChannelWrapper();
     await channelWrapper.waitForConnect();
-    this.amqpChannel = channelWrapper;
+    this.channelWrappers.set(channel, channelWrapper);
 
     await channelWrapper.addSetup(async (rawChannel: Channel) => {
-      await rawChannel.prefetch(this.channel.config.qos, false);
+      await rawChannel.prefetch(channel.config.qos, false);
       return rawChannel.consume(
         channel.config.queue,
         async (msg: ConsumeMessage | null) => {
@@ -89,7 +89,9 @@ export class RabbitmqMessagingConsumer
     errored: ConsumerDispatchedMessageError,
     channel: AmqpChannel,
   ): Promise<void> {
-    if (!this.amqpChannel) {
+    const channelWrapper = this.channelWrappers.get(channel);
+
+    if (!channelWrapper) {
       return;
     }
 
@@ -104,25 +106,27 @@ export class RabbitmqMessagingConsumer
         return this.messageRetrier.retryMessage(
           errored,
           channel,
-          this.amqpChannel,
+          channelWrapper,
           currentRetryCount,
         );
       }
     }
 
-    if (channel.config.deadLetterQueueFeature && this.amqpChannel) {
+    if (channel.config.deadLetterQueueFeature) {
       return this.messageDeadLetter.sendToDeadLetter(
         errored,
         channel,
-        this.amqpChannel,
+        channelWrapper,
       );
     }
   }
 
   async onModuleDestroy(): Promise<void> {
-    if (this.channel?.connection) {
-      await this.channel.connection.close();
+    for (const channel of this.channels) {
+      if (channel.connection) {
+        await channel.connection.close();
+      }
     }
-    this.channel = undefined;
+    this.channels.clear();
   }
 }

packages/messaging-rabbitmq-extension/test/unit/consumer/rabbitmq-messaging.consumer.spec.ts

@@ -120,7 +120,7 @@ describe('RabbitmqMessagingConsumer', () => {
       } as unknown as jest.Mocked<Channel>;
       const setupFn = (wrapper.addSetup as jest.Mock).mock.calls[0][0];
       await setupFn(rawChannel);
-      expect(rawChannel.prefetch).toHaveBeenCalledWith(10);
+      expect(rawChannel.prefetch).toHaveBeenCalledWith(10, false);
 
       const consumeHandler = (rawChannel.consume as jest.Mock).mock
         .calls[0][1] as (msg: ConsumeMessage | null) => Promise<void>;
@@ -230,21 +230,24 @@ describe('RabbitmqMessagingConsumer', () => {
       } as any;
 
       amqpChannel = {} as any;
-
-      (consumer as any).channel = channel;
-      (consumer as any).amqpChannel = amqpChannel;
+      (consumer as any).channelWrappers.set(channel, amqpChannel);
     });
 
     it('should return when no amqpChannel is available', async () => {
-      (consumer as any).amqpChannel = undefined;
+      const unregisteredChannel = {
+        config: {
+          retryMessage: 3,
+          deadLetterQueueFeature: true,
+        },
+      } as any;
 
       const errored: ConsumerDispatchedMessageError = {
         dispatchedConsumerMessage: {
           metadata: { [RABBITMQ_HEADER_RETRY_COUNT]: 1 },
         },
       } as any;
 
-      const result = await consumer.onError(errored, channel);
+      const result = await consumer.onError(errored, unregisteredChannel);
 
       expect(result).toBeUndefined();
       expect(mockRetrier.retryMessage).not.toHaveBeenCalled();
@@ -311,6 +314,7 @@ describe('RabbitmqMessagingConsumer', () => {
           deadLetterQueueFeature: true,
         },
       } as any;
+      (consumer as any).channelWrappers.set(channelWithoutRetry, amqpChannel);
 
       const errored: ConsumerDispatchedMessageError = {
         dispatchedConsumerMessage: {
@@ -335,6 +339,10 @@ describe('RabbitmqMessagingConsumer', () => {
           deadLetterQueueFeature: false,
         },
       } as any;
+      (consumer as any).channelWrappers.set(
+        channelWithDisabledDeadLetter,
+        amqpChannel,
+      );
 
       const errored: ConsumerDispatchedMessageError = {
         dispatchedConsumerMessage: {
@@ -367,26 +375,27 @@ describe('RabbitmqMessagingConsumer', () => {
   });
 
   describe('onModuleDestroy', () => {
-    it('should close connection and clear channel reference', async () => {
+    it('should close all registered channel connections', async () => {
       const close = jest.fn().mockResolvedValue(undefined);
-      (consumer as any).channel = {
+      const channel = {
         connection: {
           close,
         },
       };
+      (consumer as any).channels.add(channel);
 
       await consumer.onModuleDestroy();
 
       expect(close).toHaveBeenCalledTimes(1);
-      expect((consumer as any).channel).toBeUndefined();
+      expect((consumer as any).channels.size).toBe(0);
     });
 
-    it('should only clear channel reference when connection does not exist', async () => {
-      (consumer as any).channel = { connection: undefined };
+    it('should ignore channels without connection and still clear registry', async () => {
+      (consumer as any).channels.add({ connection: undefined });
 
       await consumer.onModuleDestroy();
 
-      expect((consumer as any).channel).toBeUndefined();
+      expect((consumer as any).channels.size).toBe(0);
     });
   });
 });