@levelup-nestjs/rabbitmq

Description

This module features an opinionated set of decorators for common RabbitMQ patterns including Publish/Subscribe and RPC using Rabbit's Direct Reply-To Queue for optimal performance.

It allows you to expose normal NestJS service methods as messaging handlers that can be configured to support a variety of messaging patterns.

Motivation

NestJS offers an out of the box microservices experience with support for a variety of transports. However, because NestJS microservices strives to work with a variety of transport mechanisms in a generic way, it misses out on some of the powerful functionality offered by individual transport layers.

Some of the most notable missing functionality includes common messaging patterns like publish/subscribe and competing consumers.

Usage

Install

npm install ---save @levelup-nestjs/rabbitmq

yarn add @levelup-nestjs/rabbitmq

Module Initialization

Import and add RabbitMQModule it to the imports array of module for which you would like to discover handlers. It may make sense for your application to do this in a shared module or to re-export it so it can be used across modules more easily. Refer to the NestJS docs on modules for more information.

If you are using exchanges, provide information about them to the module and they will be automatically asserted for you as part of initialization. If you don't, it's possible message passing will fail if an exchange is addressed that hasn't been created yet.

import { RabbitMQModule } from '@levelup-nestjs/rabbitmq';
import { Module } from '@nestjs/common';
import { MessagingController } from './messaging/messaging.controller';
import { MessagingService } from './messaging/messaging.service';

@Module({
  imports: [
    RabbitMQModule.forRoot({
      exchanges: [
        {
          name: 'exchange1',
          type: 'topic'
        }
      ],
      uri: 'amqp://rabbitmq:rabbitmq@localhost:5672'
    }),
    RabbitExampleModule
  ],
  providers: [MessagingService],
  controllers: [MessagingController]
})
export class RabbitExampleModule {}

Exposing RPC Handlers

Simply apply the RabbitRPC decorator to a new or existing NestJS service class. When a message matching the exchange and routing key is received over RabbitMQ, the result of the Service method will be automatically sent back to the requester using the Direct Reply-To Queue.

import { RabbitRPC } from '@levelup-nestjs/rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitRPC({
    exchange: 'exchange1',
    routingKey: 'rpc-route',
    queue: 'rpc-queue'
  })
  public async rpcHandler(msg: {}) {
    return {
      response: 42
    };
  }
}

Exposing Pub/Sub Handlers

Simply apply the RabbitSubscribe decorator to a new or existing NestJS service class. When a message matching the exchange and routing key is received over RabbitMQ, the service method will automatically be invoked with the message allowing it to be handled as necessary.

import { RabbitSubscribe } from '@levelup-nestjs/rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route',
    queue: 'subscribe-queue'
  })
  public async pubSubHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }
}

Message Handling

NestJS Plus provides sane defaults for message handling with automatic acking of messages that have been successfully processed by either RPC or PubSub handlers. However, there are situtations where an application may want to Negatively Acknowledge (or Nack) a message. To support this, the library exposes the Nack object which when returned from a handler allows a developer to control the message handling behavior. Simply return a Nack instance to negatively acknowledge the message.

By default, messages that are Nacked will not be requeued. However, if you would like to requeue the message so that another handler has an opportunity to process it use the optional requeue constructor argument set to true.

import { RabbitRPC } from '@levelup-nestjs/rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitRPC({
    exchange: 'exchange1',
    routingKey: 'rpc-route',
    queue: 'rpc-queue'
  })
  public async rpcHandler(msg: {}) {
    return {
      if (someCondition) {
        return 42;
      } else if (requeueCondition) {
        return new Nack(true);
      } else {
        // Will not be requeued
        return new Nack();
      }
    };
  }
}

Competing Consumers

The competing consumer pattern is useful when building decoupled applications especially when it comes to things like RPC or Work Queues. In these scenarios, it often desirable to ensure that only one handler processes a given message especially if your app is horizontally scaled.

In the previous examples, both RPC and Pub/Sub would be using the Competing Consumer pattern by default through the use of a named queue parameter. If running multiple instances of the application, each instance would bind to the same named queue and receive the messages in a round robin fashion.

If you don't want this behavior, simply don't provide a queue name. A unique one will be generated automatically and all instances of the handler will receive their own copy of the message.

Important RPC behavior has not been tested without the use of a named queue as this would cause multiple messages to potentially be sent back in response to a single request. If you're using RPC it is highly recommended that you specify a named queue. The API may be updated in the future to specifically require this.

import { RabbitSubscribe } from '@levelup-nestjs/rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route1',
    queue: 'subscribe-queue'
  })
  public async competingPubSubHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }

  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route2'
  })
  public async messagePerInstanceHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }
}

TODO

Possible validation pipeline using class-validator and class-transformer to ensure messages are well formatted
Integrate hooks for things like logging, metrics, or custom error handling