Simplest way to implement MQTT in NestJs

MQTT is a lightweight IoT messaging protocol based on the publish/subscribe model. It can provide real-time and reliable messaging services for networked devices with very little code and bandwidth. It is widely used in the industries such as the IoT, mobile Internet, smart hardware, Internet of Vehicles and power energy.

In this article we are going to focus on how to set up mqtt module and use its nest service anywhere in the nest js program for publishing payload to any topic.

1. Generate mqtt module using cli

nest g module mqtt

2. Generate service for mqtt using cli

nest g service mqtt

3. Install mqtt and ps-logger npm package

MQTT.js is a client library for the MQTT protocol, written in JavaScript for node.js and the browser.

ps-logger is a production grade logger which we will use in our project for logging any type of specific data in our console

4. Implement onModuleInit interface to mqttService

Now what we have to do is to implement onModuleInit interface to our mqttService, onModuleInit interface provides us a lifecycle hook onModuleInit() which helps us to execute any program after initialization of all modules.

import { Injectable, OnModuleInit } from "@nestjs/common";

@Injectable()
export class MqttService implements OnModuleInit {

  onModuleInit() 
{

}

5. Now establish a connection with mqtt

So here what we want is to connect with mqtt whenever our project gets initialized.To achieve this we have to establish our mqtt connection inside onModuleInit() method.

import { Injectable, OnModuleInit } from "@nestjs/common";
import {ConfigService} from "@nestjs/config";
import { connect } from "mqtt";
import { debug, error, info } from "ps-logger";

@Injectable()
export class MqttService implements OnModuleInit {
  private mqttClient;

  onModuleInit() {
    const host = this.configService.get<string>('host')
    const port = this.configService.get<string>('port')
    const clientId = `mqtt_${Math.random().toString(16).slice(3)}`;

    const connectUrl = `mqtt://${host}:${port}`;
    const topic = "/nodejs/mqtt/sp";

    this.mqttClient = connect(connectUrl, {
      clientId,
      clean: true,
      connectTimeout: 4000,
      username: this.configService.get<string>('username'),
      password: this.configService.get<string>('password'),
      reconnectPeriod: 1000,
    });

    this.mqttClient.on("connect", function () {
      info("Connected to CloudMQTT");
    });

    this.mqttClient.on("error", function () {
      error("Error in connecting to CloudMQTT");
    });
}

}

Here we are getting all our secrets using configService and we are using the on function of the returned mqttClient instance to monitor the connection status.

6. Publish to topic

Now lets publish our message or payload to any topic dynamically, for this you have to create a publish(topic:string,payload:string) method outside of onModuleInit()

  publish(topic: string, payload: string): string {
    info(`Publishing to ${topic}`);
    this.mqttClient.publish(topic, payload);
    return `Publishing to ${topic}`;
  }

Now you can inject this mqttService to any module and use its publish method to publish your payload to any topic.

Now our code looks like this ->

import { Injectable, OnModuleInit } from "@nestjs/common";
import {ConfigService} from "@nestjs/config";
import { connect } from "mqtt";
import { debug, error, info } from "ps-logger";

@Injectable()
export class MqttService implements OnModuleInit {
  private mqttClient;

  onModuleInit() {
    const host = this.configService.get<string>('host')
    const port = this.configService.get<string>('port')
    const clientId = `mqtt_${Math.random().toString(16).slice(3)}`;

    const connectUrl = `mqtt://${host}:${port}`;
    const topic = "/nodejs/mqtt/sp";

    this.mqttClient = connect(connectUrl, {
      clientId,
      clean: true,
      connectTimeout: 4000,
      username: this.configService.get<string>('username'),
      password: this.configService.get<string>('password'),
      reconnectPeriod: 1000,
    });

    this.mqttClient.on("connect", function () {
      info("Connected to CloudMQTT");
    });

    this.mqttClient.on("error", function () {
      error("Error in connecting to CloudMQTT");
    });
}

  publish(topic: string, payload: string): string {
    info(`Publishing to ${topic}`);
    this.mqttClient.publish(topic, payload);
    return `Publishing to ${topic}`;
  }
}

Thanks for reading this article

Blog