Microservicios MQTT con NestJS: transport, patterns y request/response

NestJS incluye un transport MQTT en @nestjs/microservices, de modo que un servicio Nest puede consumir telemetría de dispositivos y responder llamadas request/response a través de un broker con los mismos decoradores que usas con cualquier otro transport. Configuras Transport.MQTT, escribes handlers con @EventPattern (fire-and-forget) o @MessagePattern (request/response) y llamas a otros servicios mediante un ClientProxy.

En este tutorial construimos un pequeño servicio de telemetría, vemos exactamente qué pone Nest en la red y probamos cada pieza con un broker real.

Instalación y arranque

El transport MQTT usa MQTT.js por debajo, y lo instalas tú:

npm i @nestjs/microservices mqtt

Una app Nest puede ser un microservicio puro o una app HTTP híbrida que además escucha en MQTT. La híbrida es la opción más habitual, porque normalmente quieres health checks y una API REST junto a los handlers de mensajes:

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

El objeto options se pasa a MQTT.js, así que todo lo que acepta MQTT.js (protocolVersion: 5, keepalive, reconnectPeriod, ca) funciona aquí. Da a cada instancia un client ID único: dos réplicas con el mismo ID se expulsan mutuamente del broker en bucle. El generador de client ID ayuda a elegir un esquema.

Para un microservicio puro sin HTTP, usa NestFactory.createMicroservice<MicroserviceOptions>(AppModule, { transport: Transport.MQTT, options }) y llama a app.listen().

Handlers de eventos: telemetría de dispositivos

@EventPattern se suscribe a un topic y ejecuta tu handler con cada mensaje. Se permiten wildcards MQTT en el pattern, y MqttContext te da el topic real y el paquete original:

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

Los dispositivos no necesitan saber nada de Nest. Cuando un payload no viene envuelto en el formato { "pattern": ..., "data": ... } de Nest, Nest usa el topic como pattern y el JSON parseado (o el string sin procesar, si no es JSON) como payload. Un sensor que publica {"t": 21.4} en testmqtt/nestjs/sensors/dev-42/temperature llega directamente a onTemperature.

Usa el topic matcher para comprobar que tu pattern coincide de verdad con los topics que publican tus dispositivos. Un pattern que no coincide falla en silencio: el mensaje llega al broker y nadie lo procesa.

Request/response con @MessagePattern

@MessagePattern convierte un topic en un procedimiento remoto. Lo que devuelva el handler (un valor, una Promise o un Observable) se envía de vuelta al llamante:

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

En la red, Nest usa una convención sencilla en lugar de la propiedad Response Topic de MQTT 5:

El llamante se suscribe a <pattern>/reply y empareja las respuestas por id. En producción esto tiene dos consecuencias: la cuenta del servicio necesita permiso de publish y subscribe en el topic de respuesta, y cada llamante del mismo pattern recibe todas las respuestas (simplemente ignora los demás ids). Para RPC de alto volumen, usa patterns específicos. Para el diseño general en MQTT, consulta request/response sobre MQTT.

Llamar a un servicio con ClientProxy

Registra un cliente en el módulo del servicio que hace la llamada:

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

Después inyéctalo y llama a send() para request/response o a emit() para eventos:

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

Tres detalles confunden a mucha gente:

• send() devuelve un Observable frío. No se publica nada hasta que algo se suscribe; firstValueFrom lo hace por ti.
• No hay timeout por defecto. Si ningún handler responde, la llamada espera para siempre. Añade siempre un timeout().
• MqttRecordBuilder define opciones por mensaje: QoS, retain y, con protocolVersion: 5, user properties mediante setProperties(). Sin él, los mensajes usan los valores por defecto del cliente.

Escalado: varias réplicas

Si ejecutas tres réplicas del mismo servicio, cada una se suscribe a los mismos topics, así que cada evento se procesa tres veces. Para telemetría que escribes en una base de datos, eso suele ser un error. Las shared subscriptions de MQTT 5 lo resuelven: el broker entrega cada mensaje a un solo miembro del grupo. Las versiones recientes de Nest aceptan un prefijo $share/<group>/ en los patterns:

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

Comprueba que tu broker soporta shared subscriptions y que tu versión de @nestjs/microservices elimina el prefijo al enrutar mensajes (envía un mensaje de prueba y confirma que solo una réplica lo registra). Contexto: shared subscriptions.

Probar el servicio de punta a punta

Puedes probar ambos tipos de handler sin escribir ningún cliente.

1. Arranca la app con MQTT_USER y MQTT_PASS definidos y comprueba en los logs que el arranque fue correcto.
2. Abre el cliente MQTT online, conéctate a broker.connectmq.com por WSS (puerto 8084, ruta /mqtt) y suscríbete a testmqtt/nestjs/#.
3. Publica {"t": 21.4} en testmqtt/nestjs/sensors/dev-42/temperature. El log de Nest debería mostrar dev-42: 21.4 °C.
4. Publica una petición con el envoltorio de Nest en testmqtt/nestjs/sum:

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

1. En milisegundos deberías ver un mensaje en testmqtt/nestjs/sum/reply con "id":"manual-test-1" y "response":6.

Si la respuesta nunca aparece, revisa en este orden: ¿es idéntico el pattern en ambos lados (una barra final es otro topic)?, ¿tiene la cuenta permiso sobre el topic /reply?, ¿lanzó una excepción el handler (la respuesta lleva entonces err)? Los problemas de conexión suelen aparecer como Connection refused: Not authorized o bucles de reconexión en el log; la guía de errores de MQTT.js los explica.

Ver los topics de NestJS en vivo

Para tests automatizados, mantén los handlers delgados y prueba la lógica con unit tests directos; después añade una pequeña suite de integración que arranque el microservicio contra un broker real. La guía de integration testing muestra cómo aislar topics por ejecución.

Errores comunes

NestJS hace que MQTT parezca un transport más, pero la semántica del broker sigue aplicándose: topics, ACL, client IDs y QoS. Hazlo bien, añade timeouts a cada send() y prueba cada pattern con un broker real antes de confiar en él en producción.

Preguntas frecuentes