change library for supporting rabbitmq (#13)

* v3

* v3

* feat: BREAKING CHANGE: change library for supporting rabbitmq

* apply changes

* apply changes

Author: Sebastian-Iwanczyszyn

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nestjstools/messaging-rabbitmq-extension",
-  "version": "2.13.0",
+  "version": "3.0.0",
   "description": "Extension to handle messages and dispatch them over AMQP protocol",
   "author": "Sebastian Iwanczyszyn",
   "private": false,
@@ -51,12 +51,12 @@
     "test:e2e": "node_modules/.bin/jest --config ./test/jest-e2e.json"
   },
   "dependencies": {
-    "rabbitmq-client": "^5.0.4"
+    "amqplib": "^0.10.9"
   },
   "peerDependencies": {
     "@nestjs/common": "^10.x||^11.x",
     "@nestjs/core": "^10.x||^11.x",
-    "@nestjstools/messaging": "^2.X",
+    "@nestjstools/messaging": "^3.x",
     "reflect-metadata": "^0.2.0",
     "rxjs": "^7.x"
   },
@@ -66,7 +66,7 @@
     "@nestjs/core": "^11.0.0",
     "@nestjs/schematics": "^11.0.0",
     "@nestjs/testing": "^11.0.0",
-    "@nestjstools/messaging": "^2.21.0",
+    "@nestjstools/messaging": "^3.0.0",
     "@semantic-release/github": "^11.0.1",
     "@types/express": "^5.0.0",
     "@types/jest": "^29.5.2",

src/channel/amqp-channel.factory.ts

@@ -1,20 +1,41 @@
-import { AmqpChannelConfig, Channel } from '@nestjstools/messaging';
+import { Channel } from '@nestjstools/messaging';
 import { RmqChannelConfig as ExtensionAmqpChannelConfig } from './rmq-channel.config';
-import { Connection } from 'rabbitmq-client';
+import * as amqp from 'amqplib';
 
-export class AmqpChannel extends Channel<
-  AmqpChannelConfig | ExtensionAmqpChannelConfig
-> {
-  public readonly connection: Connection;
-  public readonly config: AmqpChannelConfig | ExtensionAmqpChannelConfig;
+export class AmqpChannel extends Channel<ExtensionAmqpChannelConfig> {
+  public connection?: any;
+  public channel?: any;
+  public readonly config: ExtensionAmqpChannelConfig;
 
-  constructor(config: AmqpChannelConfig | ExtensionAmqpChannelConfig) {
+  constructor(config: ExtensionAmqpChannelConfig) {
     super(config);
-    this.connection = new Connection(config.connectionUri);
+    this.config = config;
+  }
+
+  async init(): Promise<void> {
+    if (this.connection && this.channel) {
+      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> {
-    await this.connection.close();
-    return Promise.resolve();
+    if (this.channel) {
+      await this.channel.close();
+      this.channel = undefined;
+    }
+    if (this.connection) {
+      await this.connection.close();
+      this.connection = undefined;
+    }
   }
 }

src/channel/amqp.channel.ts

@@ -3,7 +3,7 @@ import { RABBITMQ_HEADER_ROUTING_KEY } from '../const';
 import { IMessagingConsumer } from '@nestjstools/messaging';
 import { ConsumerMessageDispatcher } from '@nestjstools/messaging';
 import { ConsumerMessage } from '@nestjstools/messaging';
-import { Injectable, OnApplicationShutdown } from '@nestjs/common';
+import { Injectable, OnModuleDestroy } from '@nestjs/common';
 import { MessageConsumer } from '@nestjstools/messaging';
 import { ConsumerDispatchedMessageError } from '@nestjstools/messaging';
 import { RabbitmqMigrator } from '../migrator/rabbitmq.migrator';
@@ -12,7 +12,7 @@ import { Buffer } from 'buffer';
 @Injectable()
 @MessageConsumer(AmqpChannel)
 export class RabbitmqMessagingConsumer
-  implements IMessagingConsumer<AmqpChannel>
+  implements IMessagingConsumer<AmqpChannel>, OnModuleDestroy
 {
   private channel?: AmqpChannel = undefined;
 
@@ -22,53 +22,68 @@ export class RabbitmqMessagingConsumer
     dispatcher: ConsumerMessageDispatcher,
     channel: AmqpChannel,
   ): Promise<void> {
-    await this.rabbitMqMigrator.run(channel);
+    await channel.init();
     this.channel = channel;
+    await this.rabbitMqMigrator.run(channel);
 
-    channel.connection.createConsumer(
-      {
-        queue: channel.config.queue,
-        queueOptions: { durable: true },
-        requeue: false,
-      },
-      async (msg): Promise<void> => {
-        const rabbitMqMessage = msg as RabbitMQMessage;
+    const amqpChannel = channel.channel;
+    if (!amqpChannel) {
+      throw new Error('AMQP channel not initialized');
+    }
+
+    await amqpChannel.consume(
+      channel.config.queue,
+      async (msg) => {
+        if (!msg) return;
 
-        let message = rabbitMqMessage.body;
+        let message: any = msg.content;
         if (Buffer.isBuffer(message)) {
-          const messageContent = message.toString();
-          message = JSON.parse(messageContent);
+          message = JSON.parse(message.toString());
         }
 
         const routingKey =
-          rabbitMqMessage.headers?.[RABBITMQ_HEADER_ROUTING_KEY] ??
-          rabbitMqMessage.routingKey;
+          msg.properties.headers?.[RABBITMQ_HEADER_ROUTING_KEY] ??
+          msg.fields.routingKey;
+
+        await dispatcher.dispatch(new ConsumerMessage(message, routingKey));
 
-        dispatcher.dispatch(new ConsumerMessage(message, routingKey));
+        amqpChannel.ack(msg);
       },
+      { noAck: false },
     );
-
-    return Promise.resolve();
   }
 
   async onError(
     errored: ConsumerDispatchedMessageError,
     channel: AmqpChannel,
   ): Promise<void> {
-    if (channel.config.deadLetterQueueFeature) {
-      const publisher = channel.connection.createPublisher();
-      const envelope = {
-        headers: {
-          'messaging-routing-key': errored.dispatchedConsumerMessage.routingKey,
+    if (channel.config.deadLetterQueueFeature && channel.channel) {
+      const exchange = 'dead_letter.exchange';
+      const routingKey = `${channel.config.queue}_dead_letter`;
+
+      await channel.channel.assertExchange(exchange, 'direct', {
+        durable: true,
+      });
+
+      channel.channel.publish(
+        exchange,
+        routingKey,
+        Buffer.from(JSON.stringify(errored.dispatchedConsumerMessage.message)),
+        {
+          headers: {
+            'messaging-routing-key':
+              errored.dispatchedConsumerMessage.routingKey,
+          },
         },
-        exchange: 'dead_letter.exchange',
-        routingKey: `${channel.config.queue}_dead_letter`,
-      };
-      await publisher.send(envelope, errored.dispatchedConsumerMessage.message);
-      await publisher.close();
+      );
     }
+  }
 
-    return Promise.resolve();
+  async onModuleDestroy(): Promise<void> {
+    if (this.channel) {
+      await this.channel.connection.close();
+      this.channel = undefined;
+    }
   }
 }
 

src/consumer/rabbitmq-messaging.consumer.ts

@@ -1,11 +1,9 @@
-import { Envelope } from 'rabbitmq-client';
-
 export class AmqpMessageBuilder {
   private constructor(
-    private exchangeName: string = undefined,
-    private routingKey: string = undefined,
-    private headers: { [key: string]: string } = undefined,
-    private message: object = undefined,
+    private exchangeName?: string,
+    private routingKey?: string,
+    private headers?: { [key: string]: any },
+    private message?: object,
   ) {}
 
   static create(): AmqpMessageBuilder {
@@ -22,16 +20,15 @@ export class AmqpMessageBuilder {
     return this;
   }
 
-  withHeaders(headers: { [key: string]: string }): AmqpMessageBuilder {
+  withHeaders(headers: { [key: string]: any }): AmqpMessageBuilder {
     this.headers = headers;
     return this;
   }
 
-  addHeader(key: string, value: string): AmqpMessageBuilder {
+  addHeader(key: string, value: any): AmqpMessageBuilder {
     if (!this.headers) {
       this.headers = {};
     }
-
     this.headers[key] = value;
     return this;
   }
@@ -42,30 +39,34 @@ export class AmqpMessageBuilder {
   }
 
   buildMessage(): AmqpMessage {
-    if (this.exchangeName === undefined) {
+    if (!this.exchangeName) {
       throw new Error('Exchange name must be defined');
     }
 
-    if (this.routingKey === undefined) {
-      throw new Error('RoutingKey name must be defined');
+    if (!this.routingKey) {
+      throw new Error('Routing key must be defined');
     }
 
-    if (this.message === undefined) {
-      throw new Error('Message name must be defined');
+    if (!this.message) {
+      throw new Error('Message must be defined');
     }
 
     return {
       message: this.message,
       envelope: {
         exchange: this.exchangeName,
         routingKey: this.routingKey,
-        headers: this.headers,
+        headers: this.headers ?? {},
       },
     };
   }
 }
 
-interface AmqpMessage {
+export interface AmqpMessage {
   message: object;
-  envelope: Envelope;
+  envelope: {
+    exchange: string;
+    routingKey: string;
+    headers?: { [key: string]: any };
+  };
 }

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

@@ -1,20 +1,15 @@
 import { RoutingMessage } from '@nestjstools/messaging';
 import { IMessageBus } from '@nestjstools/messaging';
 import { Injectable } from '@nestjs/common';
-import { Connection } from 'rabbitmq-client';
-import { AmqpMessageOptions } from '../message/amqp-message-options';
 import { AmqpChannel } from '../channel/amqp.channel';
-import { ExchangeType } from '@nestjstools/messaging';
+import { AmqpMessageOptions } from '../message/amqp-message-options';
 import { AmqpMessageBuilder } from './amqp-message.builder';
 import { RABBITMQ_HEADER_ROUTING_KEY } from '../const';
+import { ExchangeType } from '../channel/rmq-channel.config';
 
 @Injectable()
 export class AmqpMessageBus implements IMessageBus {
-  private readonly connection: Connection;
-
-  constructor(private readonly amqpChanel: AmqpChannel) {
-    this.connection = amqpChanel.connection;
-  }
+  constructor(private readonly amqpChannel: AmqpChannel) {}
 
   async dispatch(message: RoutingMessage): Promise<object | void> {
     if (
@@ -37,9 +32,19 @@ export class AmqpMessageBus implements IMessageBus {
     );
 
     const amqpMessage = messageBuilder.buildMessage();
-    const publisher = await this.connection.createPublisher();
-    await publisher.send(amqpMessage.envelope, amqpMessage.message);
-    await publisher.close();
+
+    if (!this.amqpChannel.channel) {
+      throw new Error('AMQP channel not initialized. Did you call init()?');
+    }
+
+    await this.amqpChannel.channel.publish(
+      amqpMessage.envelope.exchange,
+      amqpMessage.envelope.routingKey,
+      Buffer.from(JSON.stringify(amqpMessage.message)),
+      {
+        headers: amqpMessage.envelope.headers,
+      },
+    );
   }
 
   private createMessageBuilderWhenUndefined(
@@ -49,13 +54,13 @@ export class AmqpMessageBus implements IMessageBus {
 
     messageBuilder
       .withMessage(message.message)
-      .withExchangeName(this.amqpChanel.config.exchangeName);
+      .withExchangeName(this.amqpChannel.config.exchangeName);
 
-    if (this.amqpChanel.config.exchangeType === ExchangeType.DIRECT) {
+    if (this.amqpChannel.config.exchangeType === ExchangeType.DIRECT) {
       messageBuilder.withRoutingKey(this.getRoutingKey(message));
     }
 
-    if (this.amqpChanel.config.exchangeType === ExchangeType.TOPIC) {
+    if (this.amqpChannel.config.exchangeType === ExchangeType.TOPIC) {
       messageBuilder.withRoutingKey(message.messageRoutingKey);
     }
 
@@ -67,10 +72,11 @@ export class AmqpMessageBus implements IMessageBus {
   ): AmqpMessageBuilder {
     const options = message.messageOptions as AmqpMessageOptions;
     const messageBuilder = AmqpMessageBuilder.create();
+
     messageBuilder
       .withMessage(message.message)
       .withExchangeName(
-        options.exchangeName ?? this.amqpChanel.config.exchangeName,
+        options.exchangeName ?? this.amqpChannel.config.exchangeName,
       )
       .withRoutingKey(options.routingKey ?? this.getRoutingKey(message))
       .withHeaders(options.headers);
@@ -79,9 +85,9 @@ export class AmqpMessageBus implements IMessageBus {
   }
 
   private getRoutingKey(message: RoutingMessage): string {
-    return this.amqpChanel.config.bindingKeys !== undefined
-      ? this.amqpChanel.config.bindingKeys.length > 0
-        ? this.amqpChanel.config.bindingKeys[0]
+    return this.amqpChannel.config.bindingKeys !== undefined
+      ? this.amqpChannel.config.bindingKeys.length > 0
+        ? this.amqpChannel.config.bindingKeys[0]
         : message.messageRoutingKey
       : message.messageRoutingKey;
   }