Rabbitmq Implementation In Nestjs

RabbitMQ is a powerful and flexible messaging broker that enables communication between services in a decoupled manner. In a microservices architecture, RabbitMQ plays a vital role in ensuring that different services can interact and exchange data asynchronously. NestJS, a progressive Node.js framework, provides excellent support for building scalable microservices, including RabbitMQ integration.

In this article, we’ll walk through how to integrate RabbitMQ in a NestJS application, including both sending and receiving messages. We’ll also explore a real-world example where RabbitMQ is used to process tasks asynchronously.

Why Use RabbitMQ in NestJS?

RabbitMQ is widely used in distributed systems to enable communication between microservices, handle asynchronous tasks, and scale services independently. Some common use cases of RabbitMQ in microservices include:

• Event-Driven Architecture: Services emit events that other services consume.
• Task Queues: Perform time-consuming tasks asynchronously, like email sending or file processing.
• Decoupling Services: Communicate between services without direct dependencies, improving scalability and fault tolerance.

In NestJS, integrating RabbitMQ for message queues helps ensure reliable message delivery and makes it easier to scale applications.

Prerequisites

Before we dive into the code, make sure you have the following setup:

1. NestJS Framework: You should have a NestJS project up and running. If you haven’t created one yet, you can do so by running:

   nest new nest-rabbitmq-demo
2. RabbitMQ Setup: You’ll need access to a RabbitMQ server. You can use a local instance or a hosted solution (such as CloudAMQP or RabbitMQ Cloud).

Step 1: Install Dependencies

For RabbitMQ integration, we'll need the amqplib library, which allows us to interact with RabbitMQ through Node.js.

To install it, run the following:

npm install amqplib

Step 2: Set Up the RabbitMQ Service

Let’s now create a service that will handle both sending and receiving messages to/from RabbitMQ.

Create a new file queue.service.ts under the src folder:

import { Injectable, OnModuleInit, OnModuleDestroy } from "@nestjs/common";
import * as amqp from "amqplib";
import { Logger } from "@nestjs/common";

@Injectable()
export class QueueService implements OnModuleInit, OnModuleDestroy {
  private channel: amqp.Channel;
  private connection: amqp.Connection;
  private readonly logger = new Logger(QueueService.name);

  constructor() {}

  // On module initialization, we will connect to RabbitMQ
  async onModuleInit() {
    await this.connectToRabbitMQ();
  }

  // On module shutdown, we will gracefully close the connection and channel
  async onModuleDestroy() {
    await this.closeConnection();
  }

  // Establish connection to RabbitMQ and create a channel
  async connectToRabbitMQ() {
    try {
      this.connection = await amqp.connect("amqps://your-rabbitmq-url"); // Provide your RabbitMQ connection URL
      this.channel = await this.connection.createChannel();
      await this.channel.assertQueue("second_queue", { durable: true }); // durable queue for persistence
      this.logger.log("Connected to RabbitMQ and created the channel");

      // Start listening for messages
      this.listenForMessages();
    } catch (error) {
      this.logger.error("Error connecting to RabbitMQ:", error);
      process.exit(1); // Exit the process if connection fails
    }
  }

  // Send a message via the manually created channel
  async sendMessage(payload: string) {
    try {
      const messageBuffer = Buffer.from(payload);
      this.channel.sendToQueue("second_queue", messageBuffer, {
        persistent: true,
      });
      this.logger.log(`Message sent to queue 'second_queue': ${payload}`);
    } catch (error) {
      this.logger.error("Error sending message to RabbitMQ:", error);
    }
  }

  // Listen for incoming messages from RabbitMQ
  async listenForMessages() {
    try {
      this.channel.consume(
        "second_queue",
        (msg) => {
          if (msg) {
            const content = msg.content.toString();
            this.logger.log(`Received message: ${content}`);

            // Acknowledge the message after processing
            this.channel.ack(msg);
          } else {
            this.logger.warn("No message received");
          }
        },
        { noAck: false }
      ); // Disable auto-ack, need manual acknowledgment
    } catch (error) {
      this.logger.error("Error while listening for messages:", error);
    }
  }

  // Gracefully close the RabbitMQ connection and channel
  async closeConnection() {
    try {
      if (this.channel) {
        await this.channel.close();
        this.logger.log("Channel closed");
      }
      if (this.connection) {
        await this.connection.close();
        this.logger.log("RabbitMQ connection closed");
      }
    } catch (error) {
      this.logger.error("Error closing RabbitMQ connection:", error);
    }
  }
}

Step 3: Understanding the Code

Let's break down the implementation of this service:

• Connecting to RabbitMQ:

  this.connection = await amqp.connect("amqps://your-rabbitmq-url");
  this.channel = await this.connection.createChannel();

  This establishes a connection to RabbitMQ and creates a communication channel.

• Creating a Queue:

  await this.channel.assertQueue("second_queue", { durable: true });

  Here, we define a durable queue named second_queue. The durable: true ensures that messages are not lost if RabbitMQ crashes.

• Sending a Message:

  this.channel.sendToQueue("second_queue", messageBuffer, { persistent: true });

  This sends a message to the second_queue with persistence enabled, meaning that the message will be saved to disk in case of a RabbitMQ restart.

• Listening for Messages:

  this.channel.consume('second_queue', (msg) => { ... }, { noAck: false });

  We start consuming messages from the second_queue. After processing each message, we acknowledge it with this.channel.ack(msg).

• Graceful Shutdown: The onModuleDestroy lifecycle hook ensures that the connection and channel are properly closed when the module is destroyed.

Step 4: Sending and Receiving Messages

Now that we have our service set up, we can create a controller to expose endpoints for sending and receiving messages via RabbitMQ.

import { Controller, Post, Body } from "@nestjs/common";
import { QueueService } from "./queue.service";

@Controller("queue")
export class QueueController {
  constructor(private readonly queueService: QueueService) {}

  // Endpoint to send a message
  @Post("send")
  async sendMessage(@Body() payload: { message: string }) {
    await this.queueService.sendMessage(payload.message);
    return { status: "Message sent successfully" };
  }
}

This controller exposes a POST route (/queue/send) to send a message to the RabbitMQ queue.

Step 5: Registering the Service and Controller

Make sure to register the service and controller in your app.module.ts:

import { Module } from "@nestjs/common";
import { QueueService } from "./queue.service";
import { QueueController } from "./queue.controller";

@Module({
  imports: [],
  controllers: [QueueController],
  providers: [QueueService],
})
export class AppModule {}

Step 6: Testing the Application

To test the application, follow these steps:

1. Start RabbitMQ: If you're using a local instance, start RabbitMQ, or use a hosted service (e.g., CloudAMQP).
2. Start the NestJS App:

   npm run start
3. Send a Message: Use Postman or cURL to send a POST request to http://localhost:3000/queue/send with a JSON body like:

   {
     "message": "Hello, RabbitMQ!"
   }
4. Check Logs: Once the message is sent, the application should log the message received by RabbitMQ.

Real-World Use Case: Asynchronous Task Processing

Let’s take an example from a real-world scenario: Imagine you have a system where users upload large files to your platform. Instead of processing the file upload immediately, you can use RabbitMQ to send the file processing task to a worker, which can process the file asynchronously. This allows the main service to remain responsive, and the worker can handle tasks like resizing images, processing data, or converting files.

Conclusion

Integrating RabbitMQ into a NestJS application allows you to build scalable, decoupled microservices that communicate asynchronously. With this approach, your services can process tasks independently, improving performance and resilience. In this article, we’ve shown how to set up RabbitMQ in a NestJS application, send and receive messages, and handle connections gracefully.

RabbitMQ plays a critical role in modern microservices architectures, enabling scalable communication and task handling between services.