Microservices MQTT avec NestJS : transport, patterns et request/response

NestJS fournit un transport MQTT dans @nestjs/microservices : un service Nest peut consommer la télémétrie des appareils et répondre à des appels request/response via un broker, avec les mêmes décorateurs que pour n'importe quel autre transport. Vous configurez Transport.MQTT, écrivez des handlers avec @EventPattern (fire-and-forget) ou @MessagePattern (request/response) et appelez d'autres services via un ClientProxy.

Ce tutoriel construit un petit service de télémétrie, montre exactement ce que Nest envoie sur le réseau et teste chaque élément avec un vrai broker.

Installation et démarrage

Le transport MQTT s'appuie sur MQTT.js, que vous installez vous-même :

npm i @nestjs/microservices mqtt

Une app Nest peut être un microservice pur ou une app HTTP hybride qui écoute aussi sur MQTT. L'hybride est la configuration la plus courante, car on veut en général des health checks et une API REST à côté des handlers de messages :

import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.MQTT,
    options: {
      url: 'mqtts://broker.connectmq.com:8883',
      username: process.env.MQTT_USER,
      password: process.env.MQTT_PASS,
      clientId: `telemetry-svc-${process.pid}`,
      subscribeOptions: { qos: 1 },
    },
  });

  await app.startAllMicroservices();
  await app.listen(3000);
}
bootstrap();

L'objet options est transmis à MQTT.js : tout ce que MQTT.js accepte (protocolVersion: 5, keepalive, reconnectPeriod, ca) fonctionne ici. Donnez à chaque instance un client ID unique : deux réplicas avec le même ID s'éjectent mutuellement du broker en boucle. Le générateur de client ID aide à choisir un schéma.

Pour un microservice pur sans HTTP, utilisez NestFactory.createMicroservice<MicroserviceOptions>(AppModule, { transport: Transport.MQTT, options }) puis app.listen().

Handlers d'événements : la télémétrie des appareils

@EventPattern s'abonne à un topic et exécute votre handler pour chaque message. Les wildcards MQTT sont autorisés dans le pattern, et MqttContext donne accès au topic réel et au paquet brut :

import { Controller, Logger } from '@nestjs/common';
import { Ctx, EventPattern, MqttContext, Payload } from '@nestjs/microservices';

interface Reading { t: number; h?: number }

@Controller()
export class TelemetryController {
  private readonly log = new Logger(TelemetryController.name);

  @EventPattern('testmqtt/nestjs/sensors/+/temperature')
  onTemperature(@Payload() reading: Reading, @Ctx() ctx: MqttContext) {
    const deviceId = ctx.getTopic().split('/')[3];
    const { qos, retain } = ctx.getPacket();
    this.log.log(`${deviceId}: ${reading.t} °C (qos ${qos}, retain ${retain})`);
  }
}

Les appareils n'ont rien à savoir de Nest. Quand un payload n'est pas dans l'enveloppe { "pattern": ..., "data": ... } de Nest, Nest utilise le topic comme pattern et le JSON parsé (ou la chaîne brute si ce n'est pas du JSON) comme payload. Un capteur qui publie {"t": 21.4} sur testmqtt/nestjs/sensors/dev-42/temperature arrive directement dans onTemperature.

Utilisez le topic matcher pour vérifier que votre pattern correspond bien aux topics publiés par vos appareils. Un pattern qui ne correspond pas échoue en silence : le message arrive au broker et personne ne le traite.

Request/response avec @MessagePattern

@MessagePattern transforme un topic en procédure distante. Ce que retourne le handler (valeur, Promise ou Observable) est renvoyé à l'appelant :

import { Controller } from '@nestjs/common';
import { MessagePattern, Payload } from '@nestjs/microservices';

@Controller()
export class MathController {
  @MessagePattern('testmqtt/nestjs/sum')
  sum(@Payload() numbers: number[]): number {
    return numbers.reduce((a, b) => a + b, 0);
  }
}

Sur le réseau, Nest utilise une convention simple plutôt que la propriété Response Topic de MQTT 5 :

L'appelant s'abonne à <pattern>/reply et associe les réponses par id. Deux conséquences comptent en production : le compte du service a besoin des droits publish et subscribe sur le topic de réponse, et chaque appelant du même pattern reçoit toutes les réponses (il ignore simplement les autres ids). Pour du RPC à fort volume, gardez des patterns spécifiques. Pour la conception générale en MQTT, voir request/response sur MQTT.

Appeler un service avec ClientProxy

Enregistrez un client dans le module du service qui fait l'appel :

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { MathController } from './math.controller';
import { TelemetryController } from './telemetry.controller';
import { ReportService } from './report.service';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'MQTT_CLIENT',
        transport: Transport.MQTT,
        options: {
          url: 'mqtts://broker.connectmq.com:8883',
          username: process.env.MQTT_USER,
          password: process.env.MQTT_PASS,
        },
      },
    ]),
  ],
  controllers: [MathController, TelemetryController],
  providers: [ReportService],
})
export class AppModule {}

Injectez-le ensuite et appelez send() pour le request/response ou emit() pour les événements :

import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy, MqttRecordBuilder } from '@nestjs/microservices';
import { firstValueFrom, timeout } from 'rxjs';

@Injectable()
export class ReportService {
  constructor(@Inject('MQTT_CLIENT') private readonly mqtt: ClientProxy) {}

  async total(values: number[]): Promise<number> {
    return firstValueFrom(
      this.mqtt.send<number, number[]>('testmqtt/nestjs/sum', values).pipe(timeout(5000)),
    );
  }

  announce(status: string) {
    const record = new MqttRecordBuilder({ status, at: Date.now() })
      .setQoS(1)
      .setRetain(true)
      .build();
    return this.mqtt.emit('testmqtt/nestjs/status', record);
  }
}

Trois détails piègent souvent :

• send() renvoie un Observable froid. Rien n'est publié tant que personne ne s'abonne ; firstValueFrom s'en charge pour vous.
• Il n'y a aucun timeout par défaut. Si aucun handler ne répond, l'appel attend indéfiniment. Ajoutez toujours un timeout().
• MqttRecordBuilder définit des options par message : QoS, retain et, avec protocolVersion: 5, des user properties via setProperties(). Sans lui, les messages utilisent les valeurs par défaut du client.

Montée en charge : plusieurs réplicas

Avec trois réplicas du même service, chacun s'abonne aux mêmes topics : chaque événement est donc traité trois fois. Pour de la télémétrie écrite en base, c'est généralement une erreur. Les shared subscriptions de MQTT 5 règlent le problème : le broker ne livre chaque message qu'à un seul membre du groupe. Les versions récentes de Nest acceptent un préfixe $share/<group>/ dans les patterns :

@EventPattern('$share/telemetry/testmqtt/nestjs/sensors/+/temperature')
store(@Payload() reading: Reading) { /* un réplica par message */ }

Vérifiez que votre broker prend en charge les shared subscriptions et que votre version de @nestjs/microservices retire le préfixe lors du routage (envoyez un message de test et confirmez qu'un seul réplica le journalise). Contexte : shared subscriptions.

Tester le service de bout en bout

Vous pouvez tester les deux types de handlers sans écrire de client.

1. Lancez l'app avec MQTT_USER et MQTT_PASS définis et vérifiez dans les logs qu'elle démarre correctement.
2. Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com en WSS (port 8084, chemin /mqtt) et abonnez-vous à testmqtt/nestjs/#.
3. Publiez {"t": 21.4} sur testmqtt/nestjs/sensors/dev-42/temperature. Le log Nest doit afficher dev-42: 21.4 °C.
4. Publiez une requête dans l'enveloppe Nest sur testmqtt/nestjs/sum :

{"pattern":"testmqtt/nestjs/sum","data":[1,2,3],"id":"manual-test-1"}

1. En quelques millisecondes, un message doit apparaître sur testmqtt/nestjs/sum/reply avec "id":"manual-test-1" et "response":6.

Si la réponse n'arrive jamais, vérifiez dans cet ordre : le pattern est-il identique des deux côtés (un slash final donne un autre topic) ? Le compte a-t-il les droits sur le topic /reply ? Le handler a-t-il levé une exception (la réponse contient alors err) ? Les problèmes de connexion se manifestent en général par Connection refused: Not authorized ou des boucles de reconnexion dans le log ; le guide des erreurs MQTT.js les détaille.

Suivre les topics NestJS en direct

Pour les tests automatisés, gardez des handlers minces et testez la logique directement en tests unitaires, puis ajoutez une petite suite d'intégration qui démarre le microservice contre un vrai broker. Le guide des tests d'intégration montre comment isoler les topics à chaque exécution.

Erreurs courantes

NestJS fait de MQTT un transport comme un autre, mais la sémantique du broker s'applique toujours : topics, ACL, client IDs et QoS. Réglez ces points, ajoutez un timeout à chaque send() et testez chaque pattern sur un vrai broker avant de lui faire confiance en production.

Questions fréquentes