MQTT-Microservices mit NestJS: Transport, Patterns und Request/Response

NestJS bringt in @nestjs/microservices einen MQTT-Transport mit. Ein Nest-Service kann damit Gerätetelemetrie verarbeiten und Request/Response-Aufrufe über einen Broker beantworten – mit denselben Decorators wie bei jedem anderen Transport. Du konfigurierst Transport.MQTT, schreibst Handler mit @EventPattern (Fire-and-Forget) oder @MessagePattern (Request/Response) und rufst andere Services über einen ClientProxy auf.

In diesem Tutorial baust du einen kleinen Telemetrie-Service, siehst genau, was Nest auf die Leitung schreibt, und testest jeden Teil mit einem echten Broker.

Installation und Bootstrap

Der MQTT-Transport nutzt intern MQTT.js, das du selbst installierst:

npm i @nestjs/microservices mqtt

Eine Nest-App kann ein reiner Microservice sein oder eine hybride HTTP-App, die zusätzlich auf MQTT lauscht. Hybrid ist das häufigste Setup, weil du neben den Message-Handlern meist Health Checks und eine REST-API willst:

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();

Das options-Objekt wird an MQTT.js durchgereicht, also funktioniert hier alles, was MQTT.js akzeptiert (protocolVersion: 5, keepalive, reconnectPeriod, ca). Gib jeder Instanz eine eindeutige Client-ID – zwei Replicas mit derselben ID werfen sich gegenseitig in einer Schleife vom Broker. Der Client-ID-Generator hilft beim Festlegen eines Schemas.

Für einen reinen Microservice ohne HTTP nutzt du NestFactory.createMicroservice<MicroserviceOptions>(AppModule, { transport: Transport.MQTT, options }) und rufst app.listen() auf.

Event-Handler: Telemetrie von Geräten

@EventPattern abonniert ein Topic und führt deinen Handler für jede Nachricht aus. MQTT-Wildcards sind im Pattern erlaubt, und MqttContext liefert das tatsächliche Topic und das Rohpaket:

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})`);
  }
}

Geräte müssen nichts über Nest wissen. Steckt ein Payload nicht im Nest-Umschlag { "pattern": ..., "data": ... }, verwendet Nest das Topic als Pattern und das geparste JSON (oder den Rohstring, falls es kein JSON ist) als Payload. Ein Sensor, der {"t": 21.4} auf testmqtt/nestjs/sensors/dev-42/temperature publiziert, landet direkt in onTemperature.

Prüfe mit dem Topic Matcher, ob dein Pattern wirklich auf die Topics deiner Geräte passt. Ein Pattern, das nicht passt, scheitert lautlos: Die Nachricht erreicht den Broker, aber niemand verarbeitet sie.

Request/Response mit @MessagePattern

@MessagePattern macht aus einem Topic eine Remote-Prozedur. Was der Handler zurückgibt (Wert, Promise oder Observable), geht an den Aufrufer zurück:

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);
  }
}

Auf der Leitung nutzt Nest eine einfache Konvention statt der MQTT-5-Eigenschaft Response Topic:

Der Aufrufer abonniert <pattern>/reply und ordnet Antworten über die id zu. Zwei Folgen sind in Produktion wichtig: Der Service-Account braucht Publish- und Subscribe-Rechte auf dem Reply-Topic, und jeder Aufrufer desselben Patterns erhält jede Antwort (er ignoriert nur fremde IDs). Für RPC mit hohem Volumen solltest du spezifische Patterns wählen. Das allgemeine Muster beschreibt Request/Response über MQTT.

Einen Service per ClientProxy aufrufen

Registriere einen Client im Modul des Services, der den Aufruf macht:

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 {}

Dann injizierst du ihn und rufst send() für Request/Response oder emit() für Events auf:

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);
  }
}

Drei Details stolpern viele:

• send() liefert ein kaltes Observable. Es wird nichts publiziert, bis jemand subscribt – firstValueFrom erledigt das für dich.
• Es gibt kein Standard-Timeout. Antwortet kein Handler, wartet der Aufruf ewig. Hänge immer ein timeout() an.
• MqttRecordBuilder setzt Optionen pro Nachricht: QoS, Retain und mit protocolVersion: 5 auch User Properties über setProperties(). Ohne ihn gelten die Client-Defaults.

Skalierung: mehrere Replicas

Laufen drei Replicas desselben Services, abonniert jede dieselben Topics – jedes Event wird also dreimal verarbeitet. Für Telemetrie, die du in eine Datenbank schreibst, ist das meist falsch. Shared Subscriptions aus MQTT 5 lösen das: Der Broker liefert jede Nachricht nur an ein Mitglied der Gruppe. Aktuelle Nest-Versionen akzeptieren ein $share/<group>/-Präfix im Pattern:

@EventPattern('$share/telemetry/testmqtt/nestjs/sensors/+/temperature')
store(@Payload() reading: Reading) { /* eine Replica pro Nachricht */ }

Prüfe, ob dein Broker Shared Subscriptions unterstützt und ob deine @nestjs/microservices-Version das Präfix beim Routing entfernt (eine Testnachricht senden und bestätigen, dass nur eine Replica sie loggt). Hintergrund: Shared Subscriptions.

Den Service End-to-End testen

Beide Handler-Typen lassen sich testen, ohne einen Client zu schreiben.

1. Starte die App mit gesetzten MQTT_USER und MQTT_PASS und prüfe im Log, dass sie sauber hochfährt.
2. Öffne den Online-MQTT-Client, verbinde dich per WSS mit broker.connectmq.com (Port 8084, Pfad /mqtt) und abonniere testmqtt/nestjs/#.
3. Publiziere {"t": 21.4} auf testmqtt/nestjs/sensors/dev-42/temperature. Das Nest-Log sollte dev-42: 21.4 °C ausgeben.
4. Publiziere einen Request im Nest-Umschlag auf testmqtt/nestjs/sum:

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

1. Innerhalb von Millisekunden sollte auf testmqtt/nestjs/sum/reply eine Nachricht mit "id":"manual-test-1" und "response":6 erscheinen.

Kommt keine Antwort, prüfe in dieser Reihenfolge: Ist der Pattern-String auf beiden Seiten identisch (ein abschließender Slash ist ein anderes Topic)? Hat der Account Rechte auf das /reply-Topic? Hat der Handler eine Exception geworfen (dann enthält die Antwort err)? Verbindungsprobleme zeigen sich meist als Connection refused: Not authorized oder Reconnect-Schleifen im Log; der Leitfaden zu MQTT.js-Fehlern behandelt sie.

NestJS-Topics live beobachten

Für automatisierte Tests hältst du Handler schlank und testest die Logik direkt per Unit-Test; ergänze dann eine kleine Integrations-Suite, die den Microservice gegen einen echten Broker startet. Der Leitfaden zu Integrationstests zeigt, wie du Topics pro Testlauf isolierst.

Häufige Fehler

NestJS lässt MQTT wie einen beliebigen Transport wirken, aber die Broker-Semantik gilt weiterhin: Topics, ACLs, Client-IDs und QoS. Wenn du das sauber aufsetzt, jedes send() mit einem Timeout versiehst und jedes Pattern mit einem echten Broker testest, kannst du dem Ganzen auch in Produktion vertrauen.

Häufig gestellte Fragen