Microsserviços MQTT com NestJS: transport, patterns e request/response

O NestJS traz um transport MQTT em @nestjs/microservices, então um serviço Nest pode consumir telemetria de dispositivos e responder chamadas request/response por um broker usando os mesmos decorators de qualquer outro transport. Você configura Transport.MQTT, escreve handlers com @EventPattern (fire-and-forget) ou @MessagePattern (request/response) e chama outros serviços por meio de um ClientProxy.

Neste tutorial você vai construir um pequeno serviço de telemetria, ver exatamente o que o Nest coloca na rede e testar cada parte com um broker real.

Instalação e bootstrap

O transport MQTT usa o MQTT.js por baixo, que você mesmo instala:

npm i @nestjs/microservices mqtt

Uma app Nest pode ser um microsserviço puro ou uma app HTTP híbrida que também escuta MQTT. A híbrida é a configuração mais comum, porque normalmente você quer health checks e uma API REST ao lado dos handlers de mensagens:

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

O objeto options é repassado ao MQTT.js, então tudo o que o MQTT.js aceita (protocolVersion: 5, keepalive, reconnectPeriod, ca) funciona aqui. Dê a cada instância um client ID único: duas réplicas com o mesmo ID derrubam uma à outra do broker em loop. O gerador de client ID ajuda a escolher um padrão.

Para um microsserviço puro, sem HTTP, use NestFactory.createMicroservice<MicroserviceOptions>(AppModule, { transport: Transport.MQTT, options }) e chame app.listen().

Handlers de eventos: telemetria dos dispositivos

@EventPattern assina um topic e executa seu handler para cada mensagem. Wildcards MQTT são permitidos no pattern, e o MqttContext fornece o topic real e o pacote 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})`);
  }
}

Os dispositivos não precisam saber nada sobre o Nest. Quando um payload não vem no envelope { "pattern": ..., "data": ... } do Nest, o Nest usa o topic como pattern e o JSON parseado (ou a string crua, se não for JSON) como payload. Um sensor que publica {"t": 21.4} em testmqtt/nestjs/sensors/dev-42/temperature cai direto em onTemperature.

Use o topic matcher para conferir se o seu pattern realmente casa com os topics que os dispositivos publicam. Um pattern que não casa falha em silêncio: a mensagem chega ao broker e ninguém a trata.

Request/response com @MessagePattern

@MessagePattern transforma um topic em um procedimento remoto. O que o handler retornar (um valor, uma Promise ou um Observable) é enviado de volta a quem chamou:

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

Na rede, o Nest usa uma convenção simples em vez da propriedade Response Topic do MQTT 5:

Quem chama assina <pattern>/reply e associa as respostas pelo id. Em produção isso tem duas consequências: a conta do serviço precisa de permissão de publish e subscribe no topic de resposta, e todo chamador do mesmo pattern recebe todas as respostas (apenas ignora os outros ids). Para RPC de alto volume, use patterns específicos. Para o design geral no MQTT, veja request/response sobre MQTT.

Chamando um serviço com ClientProxy

Registre um cliente no módulo do serviço que faz a chamada:

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

Depois injete-o e chame send() para request/response ou 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);
  }
}

Três detalhes pegam muita gente:

• send() retorna um Observable frio. Nada é publicado até algo se inscrever; o firstValueFrom faz isso por você.
• Não existe timeout padrão. Se nenhum handler responder, a chamada espera para sempre. Sempre adicione um timeout().
• MqttRecordBuilder define opções por mensagem: QoS, retain e, com protocolVersion: 5, user properties via setProperties(). Sem ele, as mensagens usam os padrões do cliente.

Escalando: várias réplicas

Ao rodar três réplicas do mesmo serviço, cada uma assina os mesmos topics, então cada evento é tratado três vezes. Para telemetria gravada em banco de dados, isso geralmente está errado. As shared subscriptions do MQTT 5 resolvem: o broker entrega cada mensagem a apenas um membro do grupo. Versões recentes do Nest aceitam um prefixo $share/<group>/ nos patterns:

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

Confirme que o seu broker suporta shared subscriptions e que a sua versão do @nestjs/microservices remove o prefixo ao rotear mensagens (envie uma mensagem de teste e confira que só uma réplica a registra). Contexto: shared subscriptions.

Testando o serviço de ponta a ponta

Você pode exercitar os dois tipos de handler sem escrever nenhum cliente.

1. Inicie a app com MQTT_USER e MQTT_PASS definidos e confira nos logs que ela subiu sem erros.
2. Abra o cliente MQTT online, conecte em broker.connectmq.com via WSS (porta 8084, path /mqtt) e assine testmqtt/nestjs/#.
3. Publique {"t": 21.4} em testmqtt/nestjs/sensors/dev-42/temperature. O log do Nest deve mostrar dev-42: 21.4 °C.
4. Publique uma requisição no envelope do Nest em testmqtt/nestjs/sum:

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

1. Em milissegundos deve aparecer uma mensagem em testmqtt/nestjs/sum/reply com "id":"manual-test-1" e "response":6.

Se a resposta nunca aparecer, verifique nesta ordem: o pattern é idêntico dos dois lados (uma barra no final é outro topic)?, a conta tem permissão no topic /reply?, o handler lançou exceção (aí a resposta traz err)? Problemas de conexão costumam aparecer como Connection refused: Not authorized ou loops de reconexão no log; o guia de erros do MQTT.js cobre esses casos.

Acompanhe os topics do NestJS ao vivo

Para testes automatizados, mantenha os handlers enxutos e teste a lógica com testes unitários diretos; depois adicione uma pequena suíte de integração que sobe o microsserviço contra um broker real. O guia de testes de integração mostra como isolar topics por execução.

Erros comuns

O NestJS faz o MQTT parecer só mais um transport, mas a semântica do broker continua valendo: topics, ACLs, client IDs e QoS. Acerte esses pontos, coloque timeout em todo send() e teste cada pattern com um broker real antes de confiar nele em produção.

Perguntas frequentes