Passer au contenu
NestJS

MQTT

MQTT (Message Queuing Telemetry Transport) est un protocole de messagerie open source léger, optimisé pour une faible latence. Ce protocole offre un moyen évolutif et économique de connecter des appareils en utilisant un modèle de publication/abonnement. Un système de communication basé sur MQTT se compose d’un serveur de publication, d’un courtier et d’un ou plusieurs clients. Il est conçu pour des appareils contraints et des réseaux à faible bande passante, à haute latence ou peu fiables.

Installation

Pour commencer à construire des microservices basés sur MQTT, installez d’abord le package requis :

Fenêtre de terminal
$ npm i --save mqtt

Overview

Pour utiliser le transporteur MQTT, transmettez l’objet d’options suivant à la méthode createMicroservice() :

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

Options

L’objet options est spécifique au transporteur choisi. Le transporteur MQTT expose les propriétés décrites ici.

Client

Comme d’autres transporteurs de microservices, vous avez plusieurs options pour créer une instance de ClientProxy MQTT. Une méthode pour créer une instance consiste à utiliser le ClientsModule. Pour créer une instance de client avec le ClientsModule, importez-le et utilisez la méthode register() pour passer un objet d’options avec les mêmes propriétés que celles présentées ci-dessus dans la méthode createMicroservice(), ainsi qu’une propriété name utilisée comme jeton d’injection. Lisez-en plus sur le ClientsModule ici.

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

D’autres options pour créer un client (soit ClientProxyFactory ou @Client()) peuvent également être utilisées. Vous pouvez lire à leur sujet ici.

Context

Dans des scénarios plus sophistiqués, vous voudrez peut-être accéder à plus d’informations sur la requête entrante. Lorsque vous utilisez le transporteur MQTT, vous pouvez accéder à l’objet MqttContext.

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

Pour accéder à l’ancien paquet MQTT, utilisez la méthode getPacket() de l’objet MqttContext, comme suit :

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

Wildcards

Un abonnement peut être à un sujet explicite ou il peut inclure des caractères génériques. Deux caractères génériques sont disponibles, + et #. + est un caractère générique de niveau unique, tandis que # est un caractère générique à plusieurs niveaux qui couvre plusieurs niveaux de sujet.

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

Quality of Service (QoS)

Tout abonnement créé avec les décorateurs @MessagePattern ou @EventPattern s’abonnera avec un QoS de 0. Si un QoS plus élevé est requis, il peut être défini globalement à l’aide du bloc subscribeOptions lors de l’établissement de la connexion comme suit :

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

Si un QoS spécifique au sujet est requis, envisagez de créer un transporteur personnalisé.

Record builders

Pour configurer les options de message (ajuster le niveau de QoS, définir les indicateurs Retain ou DUP ou ajouter des propriétés supplémentaires à la charge utile), vous pouvez utiliser la classe MqttRecordBuilder. Par exemple, pour définir QoS à 2, utilisez la méthode setQoS comme suit :

main.ts
const userProperties = { 'x-version': '1.0.0' };
const record = new MqttRecordBuilder(':cat:')
  .setProperties({ userProperties })
  .setQoS(1)
  .build();


client.send('replace-emoji', record).subscribe(...);

Et vous pouvez également lire ces options côté serveur en accédant à MqttContext :

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

Dans certains cas, vous voudrez peut-être configurer des propriétés utilisateur pour plusieurs requêtes, vous pouvez passer ces options au ClientProxyFactory :

main.ts
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 {}