fix: improve channels

Author: Sebastian-Iwanczyszyn

src/channel/amqp.channel.ts

@@ -4,7 +4,6 @@ import * as amqp from 'amqplib';
 
 export class AmqpChannel extends Channel<ExtensionAmqpChannelConfig> {
   public connection?: any;
-  public channel?: any;
   public readonly config: ExtensionAmqpChannelConfig;
 
   constructor(config: ExtensionAmqpChannelConfig) {
@@ -13,26 +12,16 @@ export class AmqpChannel extends Channel<ExtensionAmqpChannelConfig> {
   }
 
   async init(): Promise<void> {
-    if (this.connection && this.channel) {
+    if (this.connection) {
       return Promise.resolve();
     }
 
     this.connection = undefined;
-    this.channel = undefined;
 
     this.connection = await amqp.connect(this.config.connectionUri);
-    this.channel = await this.connection.createChannel();
-
-    if (this.config.queue) {
-      await this.channel.assertQueue(this.config.queue, { durable: true });
-    }
   }
 
   async onChannelDestroy(): Promise<void> {
-    if (this.channel) {
-      await this.channel.close();
-      this.channel = undefined;
-    }
     if (this.connection) {
       await this.connection.close();
       this.connection = undefined;

src/consumer/rabbitmq-messaging.consumer.ts

@@ -15,6 +15,7 @@ export class RabbitmqMessagingConsumer
   implements IMessagingConsumer<AmqpChannel>, OnModuleDestroy
 {
   private channel?: AmqpChannel = undefined;
+  private amqpChannel: any;
 
   constructor(private readonly rabbitMqMigrator: RabbitmqMigrator) {}
 
@@ -26,7 +27,8 @@ export class RabbitmqMessagingConsumer
     this.channel = channel;
     await this.rabbitMqMigrator.run(channel);
 
-    const amqpChannel = channel.channel;
+    const amqpChannel = await this.channel.connection.createChannel();
+    this.amqpChannel = amqpChannel;
 
     if (!amqpChannel) {
       throw new Error('AMQP channel not initialized');
@@ -63,15 +65,11 @@ export class RabbitmqMessagingConsumer
     errored: ConsumerDispatchedMessageError,
     channel: AmqpChannel,
   ): Promise<void> {
-    if (channel.config.deadLetterQueueFeature && channel.channel) {
+    if (channel.config.deadLetterQueueFeature && this.amqpChannel) {
       const exchange = 'dead_letter.exchange';
       const routingKey = `${channel.config.queue}_dead_letter`;
 
-      await channel.channel.assertExchange(exchange, 'direct', {
-        durable: true,
-      });
-
-      channel.channel.publish(
+      this.amqpChannel.publish(
         exchange,
         routingKey,
         Buffer.from(JSON.stringify(errored.dispatchedConsumerMessage.message)),

src/message-bus/amqp-message.bus.ts

@@ -9,10 +9,14 @@ import { ExchangeType } from '../channel/rmq-channel.config';
 
 @Injectable()
 export class AmqpMessageBus implements IMessageBus {
+  public publisherChannel?: any;
+
   constructor(private readonly amqpChannel: AmqpChannel) {}
 
   async dispatch(message: RoutingMessage): Promise<object | void> {
     await this.amqpChannel.init();
+    await this.initPublisherChannel();
+
     if (
       message.messageOptions !== undefined &&
       !(message.messageOptions instanceof AmqpMessageOptions)
@@ -34,11 +38,7 @@ export class AmqpMessageBus implements IMessageBus {
 
     const amqpMessage = messageBuilder.buildMessage();
 
-    if (!this.amqpChannel.channel) {
-      throw new Error('AMQP channel not initialized. Did you call init()?');
-    }
-
-    await this.amqpChannel.channel.publish(
+    await this.publisherChannel.publish(
       amqpMessage.envelope.exchange,
       amqpMessage.envelope.routingKey,
       Buffer.from(JSON.stringify(amqpMessage.message)),
@@ -48,6 +48,12 @@ export class AmqpMessageBus implements IMessageBus {
     );
   }
 
+  async initPublisherChannel() {
+    if (!this.publisherChannel && this.amqpChannel.connection) {
+      this.publisherChannel = await this.amqpChannel.connection.createChannel();
+    }
+  }
+
   private createMessageBuilderWhenUndefined(
     message: RoutingMessage,
   ): AmqpMessageBuilder {

src/migrator/rabbitmq.migrator.ts

@@ -8,36 +8,34 @@ export class RabbitmqMigrator {
       return;
     }
 
-    if (!channel.channel) {
-      throw new Error('AMQP channel not initialized. Did you call init()?');
-    }
+    const amqpChannel = await channel.connection.createChannel();
 
-    await channel.channel.assertExchange(
+    await amqpChannel.assertExchange(
       channel.config.exchangeName,
       channel.config.exchangeType,
       { durable: true },
     );
 
-    await channel.channel.assertQueue(channel.config.queue, {
+    await amqpChannel.assertQueue(channel.config.queue, {
       durable: true,
     });
 
     if (channel.config.deadLetterQueueFeature === true) {
       const dlxExchange = 'dead_letter.exchange';
       const dlq = `${channel.config.queue}_dead_letter`;
 
-      await channel.channel.assertExchange(dlxExchange, 'direct', {
+      await amqpChannel.assertExchange(dlxExchange, 'direct', {
         durable: true,
       });
 
-      await channel.channel.assertQueue(dlq, { durable: true });
+      await amqpChannel.assertQueue(dlq, { durable: true });
 
-      await channel.channel.bindQueue(dlq, dlxExchange, dlq);
+      await amqpChannel.bindQueue(dlq, dlxExchange, dlq);
     }
 
     // Bindings
     for (const bindingKey of channel.config.bindingKeys ?? []) {
-      await channel.channel.bindQueue(
+      await amqpChannel.bindQueue(
         channel.config.queue,
         channel.config.exchangeName,
         bindingKey,