Mqtt

MQTT Integration

This guide explains how to configure and use MQTT in a messaging-based NestJS application with @nestjstools/messaging.

The MQTT extension can be used with MQTT 3.1, MQTT 3.1.1, and MQTT 5 brokers. It is suitable for IoT devices, edge services, telemetry, device-status events, and cross-language communication.

---

📦 Installation

Install the core messaging package and the MQTT extension:

npm install @nestjstools/messaging @nestjstools/messaging-mqtt-extension

or:

yarn add @nestjstools/messaging @nestjstools/messaging-mqtt-extension

You also need access to an MQTT broker. The examples below use a broker available at mqtt://localhost:1883.

---

🛠️ Channel Configuration

Import MessagingMqttExtensionModule, define a message bus, and connect it to an MQTT channel:

import { Module } from '@nestjs/common';
import { MessagingModule } from '@nestjstools/messaging';
import {
  MessagingMqttExtensionModule,
  MqttChannelConfig,
} from '@nestjstools/messaging-mqtt-extension';

@Module({
  imports: [
    MessagingMqttExtensionModule,
    MessagingModule.forRoot({
      buses: [
        {
          name: 'mqtt-message.bus',
          channels: ['mqtt-events'],
        },
      ],
      channels: [
        new MqttChannelConfig({
          name: 'mqtt-events',
          brokerUrl: 'mqtt://localhost:1883',
          clientId: 'orders-service',
          enableConsumer: true,
          defaultQos: 1,
          subscriptions: [
            { topicFilter: 'orders/#', qos: 1 },
          ],
        }),
      ],
      debug: true,
    }),
  ],
})
export class AppModule {}

This configuration creates the mqtt-message.bus message bus. It publishes messages through the mqtt-events channel and consumes every topic beginning with orders/.

The clientId must be unique for every concurrently running MQTT client.

TLS and Authentication Example

Use mqtts:// or wss:// when connecting through TLS:

new MqttChannelConfig({
  name: 'mqtt-events',
  brokerUrl: 'mqtts://broker.example.com:8883',
  clientId: 'orders-service',
  username: 'orders',
  password: process.env.MQTT_PASSWORD,
  protocolVersion: 5,
  clean: false,
  sessionExpiryInterval: 3600,
  defaultQos: 1,
  subscriptions: [
    { topicFilter: 'orders/#', qos: 1 },
  ],
  // ca, cert, key, and rejectUnauthorized are also supported.
});

---

⚙️ Channel Configuration Options

Property	Description	Default
name	Internal messaging channel name.	Required
brokerUrl	Broker URL using mqtt, mqtts, ws, or wss.	Required
clientId	MQTT client identifier. It must be unique per running client.	MQTT.js default
username	Optional broker username.	—
password	Optional broker password.	—
protocolVersion	MQTT protocol version: 3, 4, or 5.	4
clean	Starts a clean MQTT session when enabled.	true
keepalive	Keepalive interval in seconds.	60
reconnectPeriod	Delay between reconnect attempts in milliseconds.	1000
sessionExpiryInterval	MQTT 5 session expiry interval.	—
subscriptions	MQTT topic filters consumed by the channel.	[]
defaultQos	Default QoS used for publishing and subscriptions.	0
enableConsumer	Enables subscriptions and message-handler dispatch.	true
ca, cert, key	TLS certificate configuration for mqtts and wss.	—
rejectUnauthorized	Controls TLS certificate verification.	—

---

📤 Dispatching a Message

Inject the configured message bus and dispatch a RoutingMessage:

import { Injectable } from '@nestjs/common';
import {
  IMessageBus,
  MessageBus,
  RoutingMessage,
} from '@nestjstools/messaging';

@Injectable()
export class OrdersService {
  constructor(
    @MessageBus('mqtt-message.bus')
    private readonly messageBus: IMessageBus,
  ) {}

  async createOrder(): Promise<void> {
    await this.messageBus.dispatch(
      new RoutingMessage(
        { orderId: '123' },
        'orders/created',
      ),
    );
  }
}

When no MQTT-specific options are supplied, orders/created is used both as the NestJSTools routing key and the MQTT publish topic.

---

📥 Handling a Message

Create a handler for the routing key and register it as a provider in a NestJS module:

import {
  IMessageHandler,
  MessageHandler,
} from '@nestjstools/messaging';

interface OrderCreated {
  orderId: string;
}

@MessageHandler('orders/created')
export class OrderCreatedHandler
  implements IMessageHandler<OrderCreated>
{
  async handle(message: OrderCreated): Promise<void> {
    console.log(`Created order: ${message.orderId}`);
  }
}

An application subscribed to orders/# receives the MQTT message and dispatches it to this handler.

---

🔀 MQTT Topic and Routing Key

The MQTT topic determines which MQTT clients receive a message. The NestJSTools routing key determines which @MessageHandler processes it.

By default, the routing key is also used as the MQTT topic. You can configure them independently with MqttMessageOptions:

import { RoutingMessage } from '@nestjstools/messaging';
import { MqttMessageOptions } from '@nestjstools/messaging-mqtt-extension';

await this.messageBus.dispatch(
  new RoutingMessage(
    { orderId: '123' },
    'order.created',
    new MqttMessageOptions({
      topic: 'orders/events',
      qos: 1,
    }),
  ),
);

In this example:

• MQTT publishes the message to orders/events.

• NestJSTools dispatches the received message to @MessageHandler('order.created').

This allows several logical message types to share one MQTT topic while still being handled independently.

---

🌐 Wildcard and Cross-Language Messages

The extension can consume messages published by any MQTT client, including applications that do not use NestJS.

Use MQTT wildcards to subscribe to multiple topics:

subscriptions: [
  { topicFilter: 'devices/+/status' },
]

For a regular JSON or text MQTT message published to devices/device-17/status, the concrete topic becomes the routing key:

@MessageHandler('devices/device-17/status')
export class Device17StatusHandler {}

To route every matching topic to one handler, define a fixed routingKey on the subscription:

subscriptions: [
  {
    topicFilter: 'devices/+/status',
    routingKey: 'device.status',
  },
]

@MessageHandler('device.status')
export class DeviceStatusHandler {}

JSON payloads are dispatched as objects. Non-JSON payloads are dispatched as strings.

---

📨 Per-Message Options

Use MqttMessageOptions to override the publish topic or default QoS, retain a message, or set MQTT 5 properties:

new MqttMessageOptions({
  topic: 'orders/events',
  qos: 1,
  retain: false,
  properties: {
    responseTopic: 'orders/replies',
    messageExpiryInterval: 60,
  },
});

Property	Description
topic	Overrides the MQTT publish topic.
qos	Overrides the channel's default QoS.
retain	Retains the message at the broker.
dup	Sets the MQTT duplicate-delivery flag.
properties	Sets MQTT 5 publish properties such as response topic, correlation data, and message expiry.

---

🚚 Delivery Behavior

QoS	Delivery behavior
0	At most once. The message may be lost and is not retried.
1	At least once. Handlers must tolerate duplicate delivery.
2	Exactly once between the MQTT client and broker. Application-level duplicates can still occur after failures.

MQTT acknowledges the packet before the NestJS handler finishes. If a handler fails, the broker cannot automatically retry or dead-letter that packet. Keep handlers idempotent and implement application-level retry or dead-letter handling when required.