RabbitMQ Nest Client

rabbitmq-nest-client.ts

import amqp, { Message, Channel } from "amqplib/callback_api";

const handleMessage = (message: Message, channel: Channel) => {
  // Parse message
  const content = JSON.parse(message.content.toString()) as IncomingMessage;

  // Get properties
  const { replyTo, correlationId } = message.properties;

  // Reply to message 5 times
  for (let index = 0; index < 5; index++) {
    const response: OutgoingMessage = {
      id: content.id,
      // Dispose the listener if it is the last message
      isDisposed: index === 4,
      response: `Hello ${content.data} ${index}`,
    };

    // Send response to queue
    channel.sendToQueue(replyTo, Buffer.from(JSON.stringify(response)), {
      correlationId,
    });
  }
};

// Connect to RabbitMQ
amqp.connect("amqp://127.0.0.1", (err, connection) => {
  throwErrorIfExists(err);
  console.log("Connected!");

  // Create a channel
  connection.createChannel((err, channel) => {
    throwErrorIfExists(err);

    // Give queue a name
    const queue = "hello_queue";

    // Create a queue
    channel.assertQueue(queue, { durable: false });

    // Listen for messages
    channel.consume(
      queue,
      (msg) => {
        if (msg === null) return;
        return handleMessage(msg, channel);
      },
      { noAck: true }
    );
  });
});

const throwErrorIfExists = (error: Error) => {
  if (error) {
    throw error;
  }
};

interface IncomingMessage {
  pattern: string;
  data: string;
  id: string;
}

interface OutgoingMessage {
  response: string;
  isDisposed: boolean;
  id: string;
}