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:

| Field Name | Is Required | Type | Description | Example | | :------------------: | :---------: | :---------------------------------------------: | :--------------------------------------------------------------------------------------------------------------------------------------------------------------: | :--------------------------------: | | lookupdHTTPAddresses | No | string[] | http address list for nsq lookupds | ['http://localhost:4161'] | | strategy | No | 'round_robin' or 'fan_out' | message sending strategy | round_robin | | discardHandler | No | (arg: any) => void | handler function to process when message is discarded | (arg: any) => console.log(arg) | | maxInFlight | No | number | The maximum number of messages to process at once | 1 | | requeueDelay | No | number | The default amount of time (milliseconds) a message requeued should be delayed by before being dispatched by nsqd. | 90000 | | lookupdPollInterval | No | number | The frequency in seconds for querying lookupd instances. | 60 | | maxAttempts | No | number | The number of times a given message will be attempted (given to MESSAGE handler) before it will be handed to the DISCARD handler and then automatically finished | 3 | | serializer | No | Serializer in @nestjs/microservices | The instance of Serializer class which provides a serialize method to serialize the outbound message | serialize(value: any) => value | | deserializer | No | ConsumerDeserializer in @nestjs/microservices | The instance of ConsumerDeserializer class which provides a deserialize method to deserialize the inbound message | deserialize(value: any) => value |