Getting started with Node.js and MQTT

Introduction

As we may already know, Node.js is an async and event-driven JavaScript runtime and engine, powering lots of the server-side, networked applications that exist today. In this post, we are going to explore an interplay of Node.js with MQTT, a publish/subscribe (pub/sub) protocol and standard for the Internet of Things (IoT) world.

In this post, we plan to cover just the important, public MQTT APIs and functions, and also explore a simple publisher and subscriber script in Node.js.

What is MQTT?

In 1999, Andy Standford-Clark of IBM and Arlen Nipper designed the initial version of the MQTT protocol. At that time, the goal was to build a protocol that could support low bandwidth, was lightweight, and that consumed minimal resources because devices connected via satellite links were very expensive.

There are two versions of the specification: MQTT 3.1.1 and MQTT 5.0.0. Most commercial MQTT brokers now support MQTT 5, but many of the IoT managed cloud services only support MQTT 3.1.1, which used to be the most popular and widely supported version.

It is highly recommended that new IoT deployments use version 5.0.0 (approved in 2018) because its new features focus more on robust systems and cloud native scalability. You can read more of the highlights and detailed differences between both versions on MQTT’s GitHub page.

Usage today

MQTT is a lightweight client-server protocol that implements the pub/sub message transport model and has been used to connect remote peers/devices in critical IoT applications, Machine-to-Machine (M2M) communication, and many other areas that require a smooth interface for message transport with limited bandwidth but exceptional performance. This is a result of its straightforward design, ease of use, and open specification.

MQTT has been managed by the OASIS technical steering committee since 2014. The committee oversees maintaining, updating, and sustaining the standard, including organizing use-case documents and protecting its intellectual property rights.

It relies on TCP/IP, a reliable and connection-oriented low-level networking stack with a transport layer independent of the data payload structure, enabling an ordered and correct two-way, host-to-host networking model of communication.

The pub/sub pattern also enables messages to be distributed from one application to many different applications at the same time. However, the publisher and the subscriber in this case are usually independent applications (decoupled) and only connected via a broker or a server. For example, the popular pub/sub messaging platform RabbitMQ, makes use of MQTT under the hood.

MQTT use cases

MQTT has found lots of use cases, which we will explore below.

1. Enable message broadcasting: A client can publish messages or a payload to multiple other client apps, who, of course, need to subscribe to these messages via topics on a broker beforehand
2. Provide lightweight and minimal message headers/resources: This allows for minimal bandwidth usage in message transport, so MQTT can effectively scale to serve millions of IoT devices
3. Reliable message transport or delivery at its core: Most IoT devices rely on this, but on a high level, MQTT defines service layers that guarantee message transport is taken care of
4. Build highly secure message transport apps: MQTT shines here, since message payloads can be encrypted and secured with TLS and other modern authentication mechanisms like OAUTH
5. Ensuring client apps have a persistent connection: This is handy for temporary message storage in areas where network connectivity may be unreliable
   1. Related, MQTT helps reduce the time it takes for devices on a bad cellular network to reconnect
6. Find applications in different industries and domains including in the automotive, oil and gas, manufacturing, logistics, transport, and smart home device industries

You can find more details about these on the use case page of MQTT’s docs. One of the most popular open source home automation projects, home-assistant, is based on the MQTT protocol.

MQTT’s pub/sub architecture

The MQTT protocol comprises of the broker, which acts like a central server that channels messages from publisher clients to subscriber clients, and one or more client apps, which can either be subscribers or publishers of messages or data.

Brokers can be said to be like the postman, who ensures messages are delivered to their respective recipients. They should be designed to be highly scalable and secure, since they are the central moot point for MQTT clients.

On a high level, the broker acts as a gateway that routes published messages from publisher apps to the appropriate subscribers. Usually, brokers can be deployed on multiple clusters or instances and made to sit behind a load balancer for better fault tolerance. MQTT clients publish messages to a central broker (usually to a topic) and other clients can subscribe to the same topic on the broker to receive these messages.

Therefore, a client publishes a message to the topic(s), while other clients subscribe to the topic, to indicate they are interested in receiving messages about. The broker has a sort of filtering mechanism, which it uses to control which subscriber it should send messages to. What this means is that the broker checks or filters for a subscriber (or a list of subscribers) on a particular topic or groups of topics and sends messages to them.

In summary, a broker reads, acknowledges, and processes messages (which entails determining the subscribers to topic(s) and sending all appropriate messages to them) sent from a publisher client or app.

Note: MQTT relies on a way of filtering messages, whereby the broker sends messages to subscribers who are interested in getting the message content. Messages usually contain a topic, which is used by the broker to figure out how to route the specific messages to the appropriate, subscribed clients.

Subject-based filtering entails how the clients (publisher and subscriber) interact with the broker via topics, which is part of every message payload.

So far, we have made use of some technical terms that might sound new to some readers. In the next section, we will explore some of these terms and what they mean.

Some MQTT technical concepts explained

• Bridge – A connection between two MQTT brokers
• MQTT client – A device or an application written via an MQTT client library that connects to an MQTT broker over a secured network; an MQTT client could be either a publisher or a subscriber app/client
• Message – Simply the message to publish, which can be a Buffer, a String, or a JSON object
• Topics – A string identifier the broker uses to filter and send appropriate messages to connected clients. The topic name is usually structured in such a hierarchical manner with delimiters, called topic levels
  • Note that messages must contain a topic that the broker can use to appropriately route payload to interested clients
  • Example topic name: mytopic/homeautomation/closedoor
• Publisher – A publisher client distributes data or messages to a topic on the server/broker for other subscriber clients who might be interested in getting these messages
• Internet of Things (IoT) – a world of connected devices consisting of embedded systems, automation devices, wireless networks, and control mechanisms
• Decoupling – Decoupling in this context means that publishers/subscribers only need to know the hostname/IP and port of the broker — unlike the traditional client-server architecture, where the client and the server communicate directly via endpoints/APIs, usually in URI formats

Working with MQTT at the application level

Now, let us look at how MQTT works on the application level. We are going to use the Node.js client library for MQTT, mqtt.js.

MQTT.js is an open-source JavaScript library for the MQTT protocol, applicable in both Node.js and the browser. Usually, the library can be used for publishing messages and subscribing to topics on an MQTT broker.

Points to note about the MQTT.js library

• Supports both ES modules and Common.js styles of file imports
• It has a Promise-based API interface, since MQTT itself works asynchronously
• MQTT.js defaults to the old MQTT v3.1.1, to support old brokers, but the current latest version is 5.0
• MQTT clients come with an inbuilt error handler, which comes in handy when a programmer fails to handle errors in their code

Note that there are also client libraries available for a vast variety of programming languages and for the main operating systems (Linux, Windows, and macOS).

Connecting/Disconnecting to a MQTT broker/server

Client connections are always handled by a broker/server, since MQTT subscribers and publishers are separate, decoupled applications. As we mentioned earlier, both publishers and subscribers are MQTT clients, and therefore need to be connected to the same broker/server.

Clients never connect to each other directly; connection is usually between one client and the broker over TCP/IP. Other MQTT implementations or variants also connect over UDP (MQTT<-SN).
Brokers are responsible for:

• Receiving all messages
• Filtering the appropriate ones by determining which subscribing client is subscribed to each message
• Persisting connections/sessions between clients
• Authenticating and authorizing clients
• Sending the messages to the correct clients

Brokers should be easily scaled up and integrated into different backend systems. They can be quite fault-tolerant, as they are the most crucial point of publisher/subscriber communication.

To connect to a broker for the first time, the client sends a CONNECT message. Once initiated, the broker returns a CONNACK message and a status code. Another important thing to note is that the broker always keeps the connection alive or open, unless the client sends a disconnect event or their internet connection breaks.

There are some popular, freely-hosted MQTT brokers out there today. Eclipse’s Mosquitto is one of them, and it runs on all major operating systems.

There are also other commercial cloud-based or hosted brokers, like HIVEMQ. They usually come in handy if we do not intend to install and manage our own brokers. You can find a good MQTT broker for quick testing on the MQTT website.

Installing our client library

To install MQTT.js, run the following command:

npm install mqtt --save  

Note that MQTT.js bundles a command to interact with a broker. To have the MQTT protocol interface available on your system path, we can install it globally:

npm install mqtt -g 

At the end of the installation, our package.json file should look like this:

//package.json
{
"name": "mqtt-demo", 
  "version": "1.0.0", 
  "description": "A node.js and MQTT demo", 
  "main": "index.js", 
  "scripts": { 
    "start-publisher": "nodemon publisher.js", 
    "start-consumer": "nodemon subscriber.js" 
  }, 
  "keywords": [ 
    "Node.js", 
    "MQTT", 
    "Pub/Sub", 
    "IoT", 
    "message", 
    "transport" 
  ], 
  "author": "Alexander Nnakwue", 
  "license": "MIT", 
  "dependencies": { 
    "dotenv": "^10.0.0", 
    "mqtt": "^4.3.2" 
  }, 
  "devDependencies": { 
    "nodemon": "^2.0.15" 
  } 
} 

To test the program, you can run the publisher on one terminal window and run the subscriber on another terminal window.

Creating an MQTT publishing client

Now, let us create an MQTT client that publishes messages. To do so, we can import the MQTT.js library and use the connect method.

const mqtt = require('mqtt') 
require('dotenv').config() 


const clientId = 'mqttjs_' + Math.random().toString(8).substr(2, 4) 
const client  = mqtt.connect(process.env.BROKER_URL, {clientId: clientId}); 

Notice that we have added the BROKER_URL to our environment file. As we can see, the connect method accepts a given URL (broker server URL) and an optional server options object. The accepted protocols can either be MQTT, ws, wss, tcp, tls and so on. The connect method returns a connected client.

To attempt a reconnection when an MQTT client connection is broken, we can set the reconnectPeriod option (an interval between two reconnection times) to be greater than zero. The default value is 1 second, which means that after disconnecting it will try to reopen the connection again almost immediately.

const client = mqtt.connect(process.env.BROKER_URL, {clientId: clientId, clean: false, reconnectPeriod: 1}); 

If we set the value of the reconnectPeriod client option to 0, the reconnection will be disabled and terminated when the connection is dropped.

When the resubscribe option is set to its default value (true), clients can automatically reconnect and resubscribe to a previously-subscribed topic when connections are broken. Most especially for self-hosted brokers, we might want to authenticate with a username and a password. More details of the server options object can be found on the MQTT GitHub.

Publishing data and messages

Once connected to a broker, an MQTT client can send messages almost immediately. The publishing event takes a message payload and a topic name, which the broker can use to identify the subscribed parties. Also, there is an option callback that checks for an error or when the message packet has been transmitted.

The published message type or packet sent has the following attributes:

• topicName
• dupFlag
• qos
• payload
• packetId or messageId
• retainFlag

Command as shown below.

{ 
  cmd: 'publish', 
  topic: 'test/connection', 
  payload: '{"1":"Hello world","2":"Welcome to the test connection"}', 
  qos: 1, 
  retain: true, 
  messageId: 12041, 
  dup: false 
} MQTT publish packet 

When a client sends a message to a broker, the broker processes the message based on some criteria set by a developer. This should include the QoS level, which determines what kind of guarantee a message has for reaching the intended recipients and ensures message delivery guarantees.

The processing stage usually entails reading, acknowledging, and identifying the clients who have subscribed to the topics. The last step is to send messages to the subscribed clients.

Here is the complete code for the publisher.js file, including some of the public methods/APIs:

//publisher.js 
const mqtt = require('mqtt') 
require('dotenv').config() 

//the client id is used by the MQTT broker to keep track of clients and and their // state
const clientId = 'mqttjs_' + Math.random().toString(8).substr(2, 4) 
const client  = mqtt.connect(process.env.BROKER_URL, {clientId: clientId, clean: false}); 

// console.log(process.env.BROKER_URL, 'client', clientId) 

const topicName = 'test/connection' 

client.on("connect",function(connack){   
   console.log("client connected", connack); 
// on client connection publish messages to the topic on the server/broker  
  const payload = {1: "Hello world", 2: "Welcome to the test connection"} 
  client.publish(topicName, JSON.stringify(payload), {qos: 1, retain: true}, (PacketCallback, err) => { 

      if(err) { 
          console.log(err, 'MQTT publish packet') 
      } 
  }) 

  //assuming messages comes in every 3 seconds to our server and we need to publish or process these messages 
  setInterval(() => console.log("Message published"), 3000); 
}) 

client.on("error", function(err) { 
    console.log("Error: " + err) 
    if(err.code == "ENOTFOUND") { 
        console.log("Network error, make sure you have an active internet connection") 
    } 
}) 

client.on("close", function() { 
    console.log("Connection closed by client") 
}) 

client.on("reconnect", function() { 
    console.log("Client trying a reconnection") 
}) 

client.on("offline", function() { 
    console.log("Client is currently offline") 
})  

Next, we can go ahead and implement the subscribing client, which consumes the messages on the topics.

Subscribing to messages

To receive messages about the topics we are interested in, the client calls the subscribe event to the broker. The subscribed message usually contains a message packet payload, as shown below.

//stdout
Packet { 
  cmd: 'publish', 
  retain: true, 
  qos: 0, 
  dup: false, 
  length: 73, 
  topic: 'test/connection', 
  payload: <Buffer 7b 22 31 22 3a 22 48 65 6c 6c 6f 20 77 6f 72 6c 64 22 2c 22 32 22 3a 22 57 65 6c 63 6f 6d 65 20 74 6f 20 74 68 65 20 74 65 73 74 20 63 6f 6e 6e 65 63 ... 6 more bytes> 
} {"1":"Hello world","2":"Welcome to the test connection"}``` 

//
[ { topic: 'test/connection', qos: 0 } ] granted 

Instead of having topic names as regular delimited strings, we can also store topics as wildcards so that subscribers can easily subscribe to topic patterns, rather than a single topic at a time. The publisher and subscriber clients need to be aware of the topic names for patterns beforehand to do this.

Note that the subscribing clients need to know the structure of the data they’ll receive beforehand, so they can appropriately handle the data. The publisher sends messages to a particular topic in the broker in a particular format, and the intended subscriber of that message needs to know how that data is structured, so they are able to handle it correctly without breaking the application.

The complete code for the subscriber client is is shown in the file below.

// subscriber.js 
const mqtt = require('mqtt') 
const client = mqtt.connect("mqtt://test.mosquitto.org") 
const topicName = 'test/connection' 


// connect to same client and subscribe to same topic name  
client.on('connect', () => { 
    // can also accept objects in the form {'topic': qos} 
  client.subscribe(topicName, (err, granted) => { 
      if(err) { 
          console.log(err, 'err'); 
      } 
      console.log(granted, 'granted') 
  }) 
}) 


// on receive message event, log the message to the console 
client.on('message', (topic, message, packet) => { 
  console.log(packet, packet.payload.toString()); 
  if(topic === topicName) { 
   console.log(JSON.parse(message)); 
  } 
}) 
client.on("packetsend", (packet) => { 
    console.log(packet, 'packet2'); 
}) 

Additional MQTT features

Some features of MQTT are described below.

Retained messages

A retained message is an MQTT message with the retained flag/option set to true. By default, when a broker/server receives messages for a topic that doesn’t have subscribers, the messages are discarded. However, MQTT has a mechanism of retaining these messages by setting a flag, which tells the publisher to keep the message. Note that the broker only stores one retained message per topic.

Wide authentication and data security support

MQTT supports various authentication methods and data security mechanisms, including TLS and OAuth, which are usually configured on the MQTT broker. Therefore, this means that clients who implement these servers need to comply with the mechanisms as defined.

By default, MQTT supports a reconnection mechanism that sets up persistent connections in areas of low connectivity. This is important for storing messages, but, unlike traditional queuing systems, brokers do not exclusively store messages. MQTT stores messages by making sure client sessions are persistent and that the QoS levels are greater than 0.

Quality of Service (QoS)

To handle the challenges typical in pub/sub systems, MQTT has implemented three Quality of Service (QoS) levels. These three levels include 0, 1, and 2. and they determine what kind of guarantee a message has for reaching the intended recipient (clients or brokers alike).

To support reliable message delivery, the protocol supports three distinct types of quality of services messages:

• 0 – at most once
• 1 – at least once
• 2 – exactly once

For storing messages, clients must have subscribed to a topic with a Quality of Service greater than 0.

Data agnostic

MQTT is data-agnostic, and the use case of the client determines how the payload is structured. It is therefore possible to send any kind of message, including images, encoded text, encrypted data and so on.

Note: The default unencrypted MQTT port is 1883. The encrypted TCP/IP port 8883 is also supported for using MQTT over SSL.

Conclusion

MQTT offers a two-way pub/sub messaging model suitable for areas where network bandwidth is limited. Services are independent of our major applications since they are decoupled, and brokers or servers can be individually scaled.

Here is a list of open source projects leveraging on the MQTT protocol on GitHub. Details of our demo application can be found on GitHub, and you can also find details of the MQTT standard on GitHub.