MQTT — Tài liệu NestJS tiếng Việt

MQTT (Message Queuing Telemetry Transport) là một messaging protocol mã nguồn mở, nhẹ, được tối ưu cho độ trễ thấp. Protocol này cung cấp một cách kết nối thiết bị vừa có khả năng mở rộng vừa tiết kiệm chi phí, dựa trên mô hình publish/subscribe. Một hệ thống giao tiếp xây dựng trên MQTT gồm publishing server, một broker và một hoặc nhiều client. Nó được thiết kế cho các thiết bị bị giới hạn tài nguyên và các network có băng thông thấp, độ trễ cao hoặc không ổn định.

Installation (cài đặt)

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

$ npm i --save mqtt

Overview (tổng quan)

Để sử dụng MQTT transporter, truyền object options sau vào method createMicroservice():

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.MQTT,
  options: {
    url: 'mqtt://localhost:1883',
  },
});

const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.MQTT,
  options: {
    url: 'mqtt://localhost:1883',
  },
});

Gợi ý Enum Transport được import từ package @nestjs/microservices.

Options (các option)

Object options phụ thuộc vào transporter được chọn. Transporter MQTT cung cấp các property được mô tả tại đây.

Client

Giống như các transporter khác của microservice, bạn có nhiều cách để tạo instance ClientProxy cho MQTT.

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

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'MATH_SERVICE',
        transport: Transport.MQTT,
        options: {
          url: 'mqtt://localhost:1883',
        }
      },
    ]),
  ]
  ...
})

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

Context (ngữ cảnh)

Trong các tình huống 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 MQTT transporter, bạn có thể truy cập object MqttContext.

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

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

Gợi ý @Payload(), @Ctx() và MqttContext được import từ package @nestjs/microservices.

Để truy cập packet gốc của mqtt, dùng method getPacket() của object MqttContext, như sau:

@MessagePattern('notifications')
getNotifications(@Payload() data: number[], @Ctx() context: MqttContext) {
  console.log(context.getPacket());
}

@Bind(Payload(), Ctx())
@MessagePattern('notifications')
getNotifications(data, context) {
  console.log(context.getPacket());
}

Wildcards (ký tự đại diện)

Một subscription có thể trỏ tới một topic cụ thể, hoặc có thể chứa wildcard. Có hai loại wildcard: + và #. + là wildcard cho một cấp (single-level), còn # là wildcard cho nhiều cấp (multi-level), bao trùm nhiều cấp topic.

@MessagePattern('sensors/+/temperature/+')
getTemperature(@Ctx() context: MqttContext) {
  console.log(`Topic: ${context.getTopic()}`);
}

@Bind(Ctx())
@MessagePattern('sensors/+/temperature/+')
getTemperature(context) {
  console.log(`Topic: ${context.getTopic()}`);
}

Ghi chú người dịch Cách phân cấp topic bằng dấu / cùng wildcard +/# khá giống route param trong Express (/sensors/:id/temperature) hoặc router group trong Laravel, nhưng khác biệt quan trọng: MQTT không “bind” giá trị wildcard vào biến như route param — bạn chỉ nhận được toàn bộ topic qua context.getTopic() và phải tự parse ra phần mình cần.

Quality of Service (QoS)

Mọi subscription được tạo bằng decorator @MessagePattern hoặc @EventPattern sẽ subscribe với QoS 0. Nếu cần QoS cao hơn, bạn có thể đặt ở mức global bằng block subscribeOptions khi thiết lập kết nối, như sau:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.MQTT,
  options: {
    url: 'mqtt://localhost:1883',
    subscribeOptions: {
      qos: 2
    },
  },
});

const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.MQTT,
  options: {
    url: 'mqtt://localhost:1883',
    subscribeOptions: {
      qos: 2
    },
  },
});

Per-pattern QoS (QoS theo từng pattern)

Bạn có thể override QoS của subscription MQTT theo từng pattern riêng, bằng cách cung cấp qos trong field extras của decorator pattern. Khi không được chỉ định, subscribeOptions.qos ở mức global sẽ được dùng làm giá trị mặc định.

@EventPattern('critical-events', { extras: { qos: 2 } })
handleCriticalEvent(@Payload() data: any) {
  // This subscription uses QoS 2
}

@EventPattern('metrics', { extras: { qos: 0 } })
handleMetrics(@Payload() data: any) {
  // This subscription uses QoS 0
}

@Bind(Payload())
@EventPattern('critical-events', { extras: { qos: 2 } })
handleCriticalEvent(data) {
  // This subscription uses QoS 2
}

@Bind(Payload())
@EventPattern('metrics', { extras: { qos: 0 } })
handleMetrics(data) {
  // This subscription uses QoS 0
}

Gợi ý Cấu hình QoS theo từng pattern không ảnh hưởng tới hành vi hiện có. Khi extras.qos không được chỉ định, subscription sẽ dùng giá trị subscribeOptions.qos ở mức global.

Record builders (bộ dựng record)

Để cấu hình các option của message (điều chỉnh mức QoS, đặt flag Retain hoặc DUP, hoặc thêm property vào payload), bạn có thể dùng class MqttRecordBuilder. Ví dụ, để đặt QoS thành 2, dùng method setQoS như sau:

const userProperties = { 'x-version': '1.0.0' };
const record = new MqttRecordBuilder(':cat:')
  .setProperties({ userProperties })
  .setQoS(1)
  .build();
client.send('replace-emoji', record).subscribe(...);

Gợi ý Class MqttRecordBuilder được export từ package @nestjs/microservices.

Và bạn cũng có thể đọc các option này ở phía server, bằng cách truy cập MqttContext.

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

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

Trong một số trường hợp bạn có thể muốn cấu hình user properties cho nhiều request, bạn có thể truyền các option này vào ClientProxyFactory.

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

@Module({
  providers: [
    {
      provide: 'API_v1',
      useFactory: () =>
        ClientProxyFactory.create({
          transport: Transport.MQTT,
          options: {
            url: 'mqtt://localhost:1833',
            userProperties: { 'x-version': '1.0.0' },
          },
        }),
    },
  ],
})
export class ApiModule {}

Instance status updates (cập nhật trạng thái instance)

Để nhận cập nhật theo thời gian thực về kết nối 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 riêng cho driver đang dùng. Với driver MQTT, stream status phát ra các event connected, disconnected, reconnecting, và closed.

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

Gợi ý Type MqttStatus được import từ package @nestjs/microservices.

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: MqttStatus) => {
  console.log(status);
});

Listening to MQTT events (lắng nghe event của MQTT)

Trong một số trường hợp, bạn có thể muốn lắng nghe các event nội bộ được microservice phát ra. Ví dụ, bạn có thể lắng nghe event error để thực hiện thêm các xử lý khác khi có lỗi xảy ra. Để làm điều này, dùng method on(), như sau:

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<MqttEvents>('error', (err) => {
  console.error(err);
});

Gợi ý Type MqttEvents được import từ package @nestjs/microservices.

Underlying driver access (truy cập driver bên dưới)

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

Để làm điều này, bạn dùng method unwrap(), trả về underlying driver instance. Generic type parameter nên chỉ rõ loại driver instance bạn mong muốn.

const mqttClient = this.client.unwrap<import('mqtt').MqttClient>();

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

const mqttClient = server.unwrap<import('mqtt').MqttClient>();