NATS JetStream Custom Transporter for NestJS microservice

Goals of this library

• Cover all NATS JetStream features.
• Enjoy NestJS interceptors, exceptions filters, pipes, guards .
• If NestJS transporter doesn't fit NATS JetStream needs, NATS is preferred.
• Doesn't cover NATS microservice features

Install

Library

1npm i @nestjs/microservices
2npm i nats
3npm i @chance/nestjs-nats-jetstream-microservice

Server

NATS server could run locally

1docker run -d --name nats -p 4222:4222 -p 6222:6222 -p 8222:8222 nats --jetstream -m 8222

or using Synadia NGS for quick setup.

Cli

NATS cli covers all needs from the command line

1brew tap nats-io/nats-tools
2brew install nats-io/nats-tools/nats

Or download official release.

Configuration

Streams

Streams are 'message stores', each stream defines how messages are stored and what the limits (duration, size, interest) of the retention are. Streams consume normal NATS subjects, any message published on those subjects will be captured in the defined storage system. You can do a normal publish to the subject for unacknowledged delivery, though it's better to use the JetStream publish calls instead as the JetStream server will reply with an acknowledgement that it was successfully stored.

Subjects can be queried using NATS syntax

Configurations options could be set using the library

1const bootstrap = async () => {
2  const options: CustomStrategy = {
3    strategy: new NatsJetStreamServer({
4      connectionOptions: {
5        servers: "127.0.0.1:4222",
6        // Name the client for connection hostname to avoid overlap
7        name: `nats-connection.${os.hostname()}`,
8      },
9      // Stream will be created if not exist
10      // To work we need all this stream to be available
11      assertStreams: [
12        {
13          name: "booking",
14          description: "Booking domain with all its events",
15          subjects: ["booking.>"],
16        } as Partial<StreamConfig>
17      ],
18    }),
19  };
20
21  // hybrid microservice and web application
22  const app = await NestFactory.create<NestFastifyApplication>(HotelBookingModule);
23  const microService = app.connectMicroservice(options);
24  await microService.listen();
25  return app;
26};
27bootstrap();
28

Consumer options

A consumer is a stateful view of a stream. It acts as interface for clients to consume a subset of messages stored in a stream and will keep track of which messages were delivered and acknowledged by clients. Unlike with core NATS which provides an at most once delivery guarantee of a message, a consumer can provide an at least once delivery guarantee.

Configuration options could be set using library and the decorator. Or be set using the cli and using the named consumer with the decorator

1@Controller()
2export class BotNatsController {
3  constructor(private scheduleCleaning: ScheduleCleaningCommandHandler) {}
4
5  // Consumer will be created if not exists
6  // Updated if exists
7  @EventPattern("ConsumerName", {
8    description: "Trigger cleaning side effect when room is booked",
9    filter_subject: "booking.*.room-booked-event.>",
10    deliver_to: "cleanupInbox",
11    durable: "cleanupStack",
12    manual_ack: true,
13  } as ConsumeOptions)
14  async cleanup(
15    @Payload() event: RoomBookedEvent,
16    @Ctx() context: NatsJetStreamContext
17  ) {
18    // DO the work
19    context.message.ack();
20  }
21}
22

Publishing events

1@Module({
2  imports: [
3    NatsJetStreamTransport.register({
4      connectionOptions: {
5      servers: "127.0.0.1:4222",
6      name: "hotel-booking-publisher",
7    }})
8  ],
9  controllers: [BotNatsController,],
10  providers: [],
11})
12export class HotelBookingModule {}

1@Injectable()
2export class BookRoomCommandHandler {
3  constructor(private client: NatsJetStreamClient) {}
4  async handle(command: BookRoomCommand) {
5    // CloudEvent like syntax here, but nothing's mandatory
6    const event = new RoomBookedEvent(
7      { ...command.data, date: isoDate.toISOString() },
8      source,
9      correlationId
10    );
11    const uniqueBookingSlug = `booked-${correlationId}`;
12    this.client
13      .publish(
14        'my.super.subject',
15        event,
16        // deduplication trick : booking slug is unique using message ID
17        // dupe-window should be configured on stream, default 2mn
18        { msgID: uniqueBookingSlug }
19      )
20      .then((res: PubAck) => {
21        if (!res.duplicate) {
22          return res;
23        }
24        throw new ConflictException('MsgID already exists error');
25      });
26  }
27}