Programatically declare RabbitMQ consumers in NestJS / Node.js?

Question

I am using a NestJS application to consume a RabbitMQ queue. Each message can be processed no matter the order, so I'm wondering what would be the best practise to declare new consumers for the same queue.

Expected behaviour: The queue is processed by this service, which is using several consumers

Queue: [1,2,3,4,5,6, ...N];

In nestJS you can use the @RabbitSubscribe decorator to assign a function to process the data. What I want to do could be achieved by simply duplicating (and renaming) the function with the decorator, so this function will also be called to process data from the queue

  @RabbitSubscribe({
    ...
    queue: 'my-queue',
  })
  async firstSubscriber(data){
  // 1, 3, 5...
  }


 @RabbitSubscribe({
    ...
    queue: 'my-queue',
  })
  async secondSubscriber(data){
  // 2, 4, 6...
  }

I am aware that I could duplicate the project and scale horizontally, but I'd prefer doing this on the same process.

How could I declare subscribers to get this same behaviour programatically, so I could process the data with more concurrent processing?

Answer 1

You will benefit if you use @golevelup/nestjs-rabbitmq package as its supports different messages queque consumption and more if your app is hybrid. First install

npm i @golevelup/nestjs-rabbitmq

then your nestjs app structure should look like this

src --
     |
     app.module.ts
     main.ts
     app.module.ts
     someHttpModule1 --
                      |
                      someHttpModule1.controller.ts
                      someHttpModule1.module.ts
                      someHttpModule1.service.ts
                      ...
     someHttpModule2 --
                      |
                      someHttpModule2.controller.ts
                      someHttpModule2.module.ts
                      someHttpModule2.service.ts
                      ...
     ...
     // Events module is designed for consuming messages from rabbitmq
     events --
             |
             events.module.ts
             someEventConsumerModule1 --
                                       |
                                       someEventConsumerModule1.module.ts
                                       someEventConsumerModule1.service.ts
             someEventConsumerModule2 --
                                       |
                                       someEventConsumerModule2.module.ts
                                       someEventConsumerModule2.service.ts
             ...

In the src/app.module.ts file

// module imports
import { SomeHttpModule1 } from './someHttpModule1/someHttpModule1.module'
import { SomeHttpModule2 } from './someHttpModule2/someHttpModule.module'
import { EventsModule } from './events/events.module'
// and other necessery modules

@Module(
  imports: [
    SomeHttpModule1, 
    SomeHttpModule2,
    EventsModule, 
    // and other dependent modules
  ],
  controller: [],
  providers: []
})

export class AppModule{}

And in your events.module.ts file

// imports
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq'
import { SomeEventConsumerModule1 } from './someEventConsumerModule1/someEventConsumerModule1.module'
import { SomeEventConsumerModule2 } from './someEventConsumerModule2/someEventConsumerModule2.module'
// and so on

@Module({
  imports: [
    RabbitMQModule.forRoot(RabbitMQModule, {
      exchanges: [{
        name: 'amq.direct',
        type: 'direct' // check out docs for more information on exchange types
      }],
      uri: 'amqp://guest:guest@localhost:5672', // default login and password is guest, and listens locally to 5672 port in amqp protocol
      connectionInitOptions: { wait: false }
    }),
    SomeEventConsumerModule1,
    SomeEventConsumerModule2,
    // ... and other dependent consumer modules 
  ]
})
export class EventsModule {}

And below is example for one consumer module (someEventConsumerModule1.module.ts)

// imports
import { SomeEventConsumerModule1Service } from './someEventConsumerModule1.service'
// ...

@Module({
  imports: [
    SomeEventConsumerModule1,
    // other modules if injected
  ],
  providers: [
    SomeEventConsumerModule1Service
  ]
})
export class SomeEventConsumerModule1 {}

And in your service file put your business logic how to handle messages

// imports
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq'
import { Injectable } from '@nestjs/common'
import { ConsumeMessage, Channel } from 'amqplib' // for type safety you will need to install package first
// ... so on

@Injectable()
export class SomeEventConsumerModule1Service {
  constructor(
    // other module services if needs to be injected
  ) {}
  @RabbitSubscribe({
    exchange: 'amq.direct',
    routingKey: 'direct-route-key', // up to you
    queue: 'queueNameToBeConsumed',
    errorHandler: (channel: Channel, msg: ConsumeMessage, error: Error) => {
      console.log(error)
      channel.reject(msg, false) // use error handler, or otherwise app will crush in not intended way
    }
  })
  public async onQueueConsumption(msg: {}, amqpMsg: ConsumeMessage) {
    const eventData = JSON.parse(amqpMsg.content.toString())
    // do something with eventData
    console.log(`EventData: ${eventData}, successfully consumed!`)
  }
  // ... and in the same way
}