NATS

NATS là một hệ thống messaging mã nguồn mở đơn giản, an toàn và có hiệu năng cao dành cho các ứng dụng cloud native, messaging IoT và kiến trúc microservices. NATS server được viết bằng ngôn ngữ Go, nhưng các thư viện client để giao tiếp với server thì có sẵn cho hàng chục ngôn ngữ lập trình phổ biến. NATS hỗ trợ cả hai kiểu delivery At Most Once và At Least Once. Nó có thể chạy ở bất cứ đâu, từ server lớn và cloud instance, cho tới edge gateway và cả các thiết bị Internet of Things.

Cài đặt

Để bắt đầu xây dựng microservice dựa trên NATS, trước tiên hãy cài package cần thiết:

$ npm i --save @nats-io/transport-node

Tổng quan

Để dùng transporter NATS, truyền object options sau vào method createMicroservice():

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.NATS,
  options: {
    servers: ['nats://localhost:4222'],
  },
});

const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.NATS,
  options: {
    servers: ['nats://localhost:4222'],
  },
});

Options

Object options phụ thuộc vào transporter bạn chọn. Transporter NATS hỗ trợ các thuộc tính được mô tả tại đây cùng với các thuộc tính sau:

Client

Giống các transporter microservice khác, bạn có nhiều cách để tạo một instance ClientProxy cho NATS.

Một cách để tạo instance là dùng ClientsModule. Để tạo client instance bằng ClientsModule, import nó và dùng method register() để truyền một object options có các thuộc tính giống như đã trình bày ở trên trong method createMicroservice(), cùng với thuộc tính name dùng làm injection token. Đọc thêm về ClientsModule tại đây.

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'MATH_SERVICE',
        transport: Transport.NATS,
        options: {
          servers: ['nats://localhost:4222'],
        }
      },
    ]),
  ]
  ...
})

Bạn cũng có thể dùng các cách khác để tạo client (ClientProxyFactory hoặc @Client()). Bạn có thể đọc thêm về chúng tại đây.

Request-response

Với kiểu message request-response (đọc thêm), transporter NATS không dùng cơ chế Request-Reply có sẵn của NATS. Thay vào đó, một “request” được publish lên một subject nhất định bằng method publish() kèm theo một tên reply subject duy nhất, và các responder lắng nghe trên subject đó rồi gửi response tới reply subject. Reply subject được định hướng ngược lại tới bên gửi request một cách động, bất kể vị trí của cả hai bên.

Event-based

Với kiểu message event-based (đọc thêm), transporter NATS dùng cơ chế Publish-Subscribe có sẵn của NATS. Một publisher gửi message lên một subject và mọi subscriber đang active lắng nghe trên subject đó sẽ nhận được message. Subscriber cũng có thể đăng ký quan tâm tới các subject dùng wildcard, hoạt động khá giống regular expression. Pattern một-tới-nhiều này đôi khi được gọi là fan-out.

Queue group

NATS cung cấp một tính năng load balancing có sẵn gọi là distributed queue. Để tạo một queue subscription, dùng thuộc tính queue như sau:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.NATS,
  options: {
    servers: ['nats://localhost:4222'],
    queue: 'cats_queue',
  },
});

Context

Trong các trường hợp phức tạp hơn, bạn có thể cần truy cập thêm thông tin về request đến. Khi dùng transporter NATS, bạn có thể truy cập object NatsContext.

@MessagePattern('notifications')
getNotifications(@Payload() data: number[], @Ctx() context: NatsContext) {
  console.log(`Subject: ${context.getSubject()}`);
}

@Bind(Payload(), Ctx())
@MessagePattern('notifications')
getNotifications(data, context) {
  console.log(`Subject: ${context.getSubject()}`);
}

Wildcard

Một subscription có thể trỏ tới một subject cụ thể, hoặc có thể dùng wildcard.

@MessagePattern('time.us.*')
getDate(@Payload() data: number[], @Ctx() context: NatsContext) {
  console.log(`Subject: ${context.getSubject()}`); // ví dụ: "time.us.east"
  return new Date().toLocaleTimeString(...);
}

@Bind(Payload(), Ctx())
@MessagePattern('time.us.*')
getDate(data, context) {
  console.log(`Subject: ${context.getSubject()}`); // ví dụ: "time.us.east"
  return new Date().toLocaleTimeString(...);
}

Record builder

Để cấu hình message options, bạn có thể dùng class NatsRecordBuilder (lưu ý: cách này cũng dùng được cho luồng event-based). Ví dụ, để thêm header x-version, dùng method setHeaders như sau:

import * as nats from '@nats-io/nats-core';

// đâu đó trong code của bạn
const headers = nats.headers();
headers.set('x-version', '1.0.0');

const record = new NatsRecordBuilder(':cat:').setHeaders(headers).build();
this.client.send('replace-emoji', record).subscribe(...);

Serializer và deserializer tùy chỉnh

Kể từ NestJS v12, Nest serialize NATS packet dưới dạng JSON string, và các deserializer NATS tùy chỉnh nhận vào object message NATS đầy đủ thay vì một Uint8Array thô. Nếu bạn đã viết một deserializer tùy chỉnh, hãy đọc payload qua msg.json() thay vì tự decode byte theo cách thủ công:

import { Deserializer, IncomingRequest } from '@nestjs/microservices';

export class CustomNatsDeserializer implements Deserializer {
  deserialize(msg: any): IncomingRequest {
    // Trước đây: JSON.parse(new TextDecoder().decode(msg));
    return msg.json();
  }
}

Và bạn cũng có thể đọc các header này ở phía server, bằng cách truy cập NatsContext, như sau:

@MessagePattern('replace-emoji')
replaceEmoji(@Payload() data: string, @Ctx() context: NatsContext): string {
  const headers = context.getHeaders();
  return headers['x-version'] === '1.0.0' ? '🐱' : '🐈';
}

@Bind(Payload(), Ctx())
@MessagePattern('replace-emoji')
replaceEmoji(data, context) {
  const headers = context.getHeaders();
  return headers['x-version'] === '1.0.0' ? '🐱' : '🐈';
}

Trong một số trường hợp bạn có thể muốn cấu hình header cho nhiều request, bạn có thể truyền chúng như options cho ClientProxyFactory:

import { Module } from '@nestjs/common';
import { ClientProxyFactory, Transport } from '@nestjs/microservices';

@Module({
  providers: [
    {
      provide: 'API_v1',
      useFactory: () =>
        ClientProxyFactory.create({
          transport: Transport.NATS,
          options: {
            servers: ['nats://localhost:4222'],
            headers: { 'x-version': '1.0.0' },
          },
        }),
    },
  ],
})
export class ApiModule {}

Cập nhật trạng thái instance

Để nhận cập nhật real-time về connection và trạng thái của driver instance bên dưới, bạn có thể subscribe vào stream status. Stream này cung cấp các cập nhật trạng thái đặc thù cho driver đang dùng. Với driver NATS, stream status phát ra các event connected, disconnected và reconnecting.

this.client.status.subscribe((status: NatsStatus) => {
  console.log(status);
});

Tương tự, bạn có thể subscribe vào stream status của server để nhận thông báo về trạng thái của server.

const server = app.connectMicroservice<MicroserviceOptions>(...);
server.status.subscribe((status: NatsStatus) => {
  console.log(status);
});

Lắng nghe event của Nats

Trong một số trường hợp, bạn có thể muốn lắng nghe các event nội bộ được phát ra bởi microservice. Ví dụ, bạn có thể lắng nghe event error để kích hoạt các xử lý bổ sung khi có lỗi xảy ra. Để làm điều này, dùng method on(), như minh họa dưới đây:

this.client.on('error', (err) => {
  console.error(err);
});

Tương tự, bạn có thể lắng nghe các event nội bộ của server:

server.on<NatsEvents>('error', (err) => {
  console.error(err);
});

Truy cập driver bên dưới

Với các use case nâng cao hơn, bạn có thể cần truy cập trực tiếp driver instance bên dưới. Điều này có thể hữu ích cho các trường hợp như tự đóng connection bằng tay hoặc dùng các method đặc thù của driver. Tuy nhiên, hãy nhớ rằng trong hầu hết trường hợp, bạn không nên cần truy cập driver trực tiếp.

Để làm vậy, bạn có thể dùng method unwrap(), method này trả về driver instance bên dưới. Generic type parameter nên chỉ định type của driver instance mà bạn mong đợi.

const natsConnection =
  this.client.unwrap<import('@nats-io/transport-node').NatsConnection>();

Tương tự, bạn có thể truy cập driver instance bên dưới của server:

const natsConnection =
  server.unwrap<import('@nats-io/transport-node').NatsConnection>();