MQTT Microservices with NestJS: Transport, Patterns and Request/Response

NestJS ships an MQTT transport in @nestjs/microservices, so a Nest service can consume device telemetry and answer request/response calls over a broker with the same decorators you use for any other transport. You configure Transport.MQTT, write handlers with @EventPattern (fire-and-forget) or @MessagePattern (request/response), and call other services through a ClientProxy.

This tutorial builds a small telemetry service, shows exactly what Nest puts on the wire, and tests every piece with a live broker.

Install and bootstrap

The MQTT transport uses MQTT.js under the hood, which you install yourself:

npm i @nestjs/microservices mqtt

A Nest app can be a pure microservice or a hybrid HTTP app that also listens on MQTT. Hybrid is the most common setup because you usually want health checks and a REST API next to the message handlers:

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

The options object is passed to MQTT.js, so anything MQTT.js accepts (protocolVersion: 5, keepalive, reconnectPeriod, ca) works here. Give every instance a unique client ID — two replicas with the same ID kick each other off the broker in a loop. The client ID generator is handy for picking a scheme.

For a pure microservice without HTTP, use NestFactory.createMicroservice<MicroserviceOptions>(AppModule, { transport: Transport.MQTT, options }) and call app.listen().

Event handlers: telemetry from devices

@EventPattern subscribes to a topic and runs your handler for each message. MQTT wildcards are allowed in the pattern, and MqttContext gives you the real topic and the raw packet:

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

Devices do not need to know anything about Nest. When a payload is not wrapped in Nest's { "pattern": ..., "data": ... } envelope, Nest uses the topic as the pattern and the parsed JSON (or the raw string, if it is not JSON) as the payload. A sensor publishing {"t": 21.4} to testmqtt/nestjs/sensors/dev-42/temperature lands directly in onTemperature.

Use the topic matcher to check that your pattern really matches the topics your devices publish. A pattern that does not match fails silently: the message arrives at the broker and nobody handles it.

Request/response with @MessagePattern

@MessagePattern turns a topic into a remote procedure. Whatever the handler returns (a value, a Promise or an Observable) is sent back to the caller:

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

On the wire, Nest uses a simple convention rather than the MQTT 5 Response Topic property:

The caller subscribes to <pattern>/reply and matches responses by id. Two consequences matter in production: the service account needs publish and subscribe permission on the reply topic, and every caller of the same pattern receives every reply (they just ignore other ids). For high-volume RPC, keep patterns specific. For the general MQTT design, see request/response over MQTT.

Calling a service with ClientProxy

Register a client in the module of the service that makes the call:

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

Then inject it and call send() for request/response or emit() for events:

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

Three details trip people up:

• send() returns a cold Observable. Nothing is published until something subscribes — firstValueFrom does that for you.
• There is no default timeout. If no handler answers, the call waits forever. Always pipe a timeout().
• MqttRecordBuilder sets per-message options: QoS, retain and, with protocolVersion: 5, user properties via setProperties(). Without it, messages use the client defaults.

Scaling: several replicas

When you run three replicas of the same service, each one subscribes to the same topics, so every event is handled three times. For telemetry that you write to a database that is usually wrong. MQTT 5 shared subscriptions fix it: the broker delivers each message to only one member of a group. Recent Nest versions accept a $share/<group>/ prefix on handler patterns:

@EventPattern('$share/telemetry/testmqtt/nestjs/sensors/+/temperature')
store(@Payload() reading: Reading) { /* one replica per message */ }

Check that your broker supports shared subscriptions and that your @nestjs/microservices version strips the prefix when it routes messages (send one test message and confirm only one replica logs it). Background: shared subscriptions.

Testing the service end to end

You can exercise both handler types without writing a client.

1. Start the app with MQTT_USER and MQTT_PASS set, and watch the logs for a successful start.
2. Open the online MQTT client, connect to broker.connectmq.com over WSS (port 8084, path /mqtt) and subscribe to testmqtt/nestjs/#.
3. Publish {"t": 21.4} to testmqtt/nestjs/sensors/dev-42/temperature. The Nest log should print dev-42: 21.4 °C.
4. Publish a request in Nest's envelope to testmqtt/nestjs/sum:

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

1. Within milliseconds you should see a message on testmqtt/nestjs/sum/reply with "id":"manual-test-1" and "response":6.

If the reply never appears, check in this order: is the pattern string identical on both sides (a trailing slash is a different topic), does the account have permission on the /reply topic, and did the handler throw (the reply then carries err). Connection-level problems usually show up as Connection refused: Not authorized or reconnect loops in the log; the MQTT.js errors guide covers them.

Watch the NestJS topics live

For automated tests, keep handlers thin and unit-test the logic directly, then add a small integration suite that boots the microservice against a real broker. The integration testing guide shows how to isolate topics per test run.

Common errors

NestJS makes MQTT feel like any other transport, but the broker semantics still apply: topics, ACLs, client IDs and QoS. Get those right, add timeouts to every send(), and test each pattern with a real broker before you trust it in production.

Frequently asked questions