nestjs-nsq-transporter v2.0.0

Nestjs Transporter For NSQ

Overview

This library provides a nestjs transporter by taking NSQ as its message broker. We could simply take the concept of a transporter as an event-driven framework as follows:

• As a message producer, I could send a message to NSQ to various topics by emitting various events based on the topic names.
• As a message consumer, I could receive a message by subscribing to those emitted events.

For more details on the overall architecture, please kinldy refer to this article

Restriction

This library has not implemented the request-response communication style and currently it only supports the event-dispatching style.

Produce a message

Add the nsq client provider in your module's definition:

import { DynamicModule, Module } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { ClientNsq, NsqOptions } from 'nestjs-nsq-transporter';

@Module({
  providers: [
    {
      provide: 'NSQ_CLIENT',
      useFactory: (): ClientProxy => new ClientNsq(options),
    }
    SomeService,
  ],
})
export class SomeModule {}

Then you can inject the ClientNsq instance and use it to emit the message event:

import { Injectable } from '@nestjs/common';

@Injectable()
export class SomeService {
  constructor(@Inject('NSQ_CLIENT') private nsqProducer: ClientNsq) {}

  sendMessage(topic: string, msg: any) {
    return this.nsqProducer.emit(topic, msg);
  }
}

Notes: You can not blindly pass any key/value as the 2nd argument(msg) of the emit function because there are two special reserved keys: meta and options. The meta is used by the default OutboundEventSerializer to attach extra informations; options is for setting the options for the nsq message producing, see the sample below.

    this.nsqProducer.emit(topic,
      { foo: 1, bar: 2, meta: { component: 'XYZ'}, options: {retry: { retries: 3 } }
    );
    // The `emit` above will send a nsq message with payload as follow with 3 times retry strategy.
    { data: { foo: 1, bar: 2 }, meta: { component: 'XYZ'} }

Receive a message

Connect to the nsq microservice in your app and specify the nsq options as follows:

import { MicroserviceOptions } from '@nestjs/microservices';
import { serverNsq } from 'nestjs-nsq-transporter';

app.connectMicroservice<MicroserviceOptions>({
  strategy: new ServerNsq(options),
});

Then use @EventPattern to mark a function as the handler for messages from specific event pattern:

import { Injectable } from '@nestjs/common';
import { EventPattern, Ctx, Payload } from '@nestjs/microservices';

@Controller()
export class AppController {

  @EventPattern({
    topic: 'topic1',
    channel: 'channel1',
    options: { // optional
      maxAttempts: 3
    }
  })
  messageHandlerForTopic1(@Payload() payload: any, @Ctx() context: NsqContext)
    // Handle messages
    // Notes: if you throw an Error from this messageHandler, it's better to use `RpcException` so that we have explicit log in nsq-transporter.
  }
}

Definition of NsqOptions

The options that passed to either ServerNsq or ClientNsq has the type as NsqOptions and the available fields are as folows: