Introduction
This documentation provides a comprehensive guide to integrating Node.js with the our MQTT broker (CrystalMQ) or any MQTT broker of your choice. It covers essential tasks such as establishing connections, subscribing to topics, unsubscribing, and exchanging messages. By following these steps, you can effectively implement MQTT communication within Node.js applications.
Pre-requisites
Before getting started, ensure you have the following:
- Node JS - Make sure you have installed Node JS with version 18 or higher.
- MQTT Library
npm install mqtt
MQTT Connect
This section covers various ways to connect to an MQTT broker using Node JS. Before proceeding, ensure your broker supports the desired connection method (TCP, TLS, WebSocket, etc.) and you have the correct connection parameters:
- Broker Address
- Port
- Username/Password (if required)
- CA Certificate (for TLS connections)
Basic TCP Connection
const mqtt = require('mqtt');
const brokerUrl = 'mqtt://broker-url:1883';
const conn_params = {
clientId: "testclient1",
};
const client = mqtt.connect(brokerUrl, conn_params);
client.on('connect', () => {
console.log('Connected to MQTT broker');
});
Connect with Authentication
To connect to an MQTT Broker that requires Authentication, set Username & Password in connection parameters.
const conn_params = {
clientId: "testclient1",
username: "username",
password: "password",
};
Connect over TLS/SSL
To connect securely using TLS/SSL:
const fs = require('fs');
const conn_params = {
protocol: 'mqtts',
host: 'localhost',
port: 8883,
clientId: "testclient1",
ca: [fs.readFileSync('/path/to/root.crt')],
};
Connect via WebSocket
To connect over WebSocket instead of TCP keep the broker url as follows:
const brokerUrl = 'ws://broker-url:10443/mqtt';
MQTT Publish
To send data to a topic:
const publishInterval = 2000;
const topic = 'testtopic';
client.on('connect', () => {
console.log('Connected to MQTT broker');
const message = JSON.stringify({"temperature": 27.6, "humidity": 34})
setInterval(() => {
client.publish(topic, message, (err) => {
if (err) {
console.error('Error publishing message:', err);
} else {
console.log('Message published:', message);
}
});
}, publishInterval);
});
MQTT Subscribe
To receive messages, subscribe to a topic filter. The broker will forward matching messages to your client.
client.on('connect', () => {
console.log('Connected to MQTT broker');
client.subscribe('some_topic', (err) => {
if (err) {
console.error('Subscription error:', err);
} else {
console.log(`Subscribed to topic: ${topic}`);
}
});
});
client.on('message', (topic, message) => {
try {
const data = JSON.parse(message.toString()); // parse JSON if applicable
console.log(`Received on '${topic}':`, data);
} catch (e) {
console.log(`Received non-JSON message on '${topic}':`, message.toString());
}
});
client.unsubscribe('testtopic', (err) => {
if (err) {
console.error('Unsubscribe error:', err);
} else {
console.log('Unsubscribed from topic: testtopic');
}
});
MQTT Disconnect
Always disconnect cleanly to avoid resource leaks and ensure proper session handling on the broker:
client.end(() => {
console.log('Disconnected from MQTT broker');
});
Sample MQTTv311 Client
const mqtt = require('mqtt');
const fs = require('fs');
// Toggle Authentication (set to "ENABLED" or "DISABLED")
const AUTH = "ENABLED";
// ------------ MQTT Broker Configuration -----------
const brokerUrl = 'mqtt://broker-url:1883'; // Replace with your actual broker URL/IP
// MQTT connection parameters
const conn_params = {
clientId: "testclient1", // Unique client ID
// Uncomment below lines if broker requires authentication
// username: "username",
// password: "password",
// Uncomment below lines for TLS/SSL secured connection
// protocol: 'mqtts',
// host: 'broker-url',
// port: 8883,
// ca: [fs.readFileSync('/path/to/root.crt')], // Root CA certificate
};
let client;
// ------------ MQTT Client Initialization -----------
if (AUTH === "ENABLED") {
client = mqtt.connect(brokerUrl, conn_params); // With URL and options
} else {
client = mqtt.connect(conn_params); // Without broker URL
}
// ------------ MQTT Topic & Publish Interval -----------
const topic = 'testtopic'; // Topic to publish and subscribe
const publishInterval = 2000; // Publish every 2 seconds
// ------------ On Successful Connection -----------
client.on('connect', () => {
console.log('Connected to MQTT broker');
// Subscribe to the topic
client.subscribe(topic, (err) => {
if (err) {
console.error('Subscription error:', err);
} else {
console.log(`Subscribed to topic: ${topic}`);
}
});
// Sample JSON payload to publish
const message = JSON.stringify({ temperature: 27.6, humidity: 34 });
// Publish the message periodically
setInterval(() => {
client.publish(topic, message, { retain: true }, (err) => {
if (err) {
console.error('Error publishing message:', err);
} else {
console.log('Message published:', message);
}
});
}, publishInterval);
});
// ------------ Message Received Callback -----------
client.on('message', (topic, message) => {
try {
// Attempt to parse message as JSON
const data = JSON.parse(message.toString());
console.log(`Received on '${topic}':`, data);
} catch (e) {
// Fallback if message is not JSON
console.log(`Received non-JSON message on '${topic}':`, message.toString());
}
});
// ------------ Additional Event Callbacks -----------
client.on('error', (err) => {
console.error('Error:', err);
});
client.on('close', () => {
console.log('Disconnected from MQTT broker');
});
client.on('reconnect', () => {
console.log('Reconnecting to MQTT broker...');
});
client.on('offline', () => {
console.warn('MQTT client is offline');
});
// ------------ Unsubscribe & Disconnect -----------
// Unsubscribe from the topic
// client.unsubscribe('testtopic', (err) => {
// if (err) {
// console.error('Unsubscribe error:', err);
// } else {
// console.log('Unsubscribed from topic: testtopic');
// }
// });
// Disconnect the client
// client.end(() => {
// console.log('Disconnected from MQTT broker');
// });
Sample MQTTv5 Client
Ensure a proper termination of your client's connection with the broker to avoid issues and resource leaks on both sides, thereby maintaining system stability.
Use the following code to disconnect the client from the broker:
const mqtt = require('mqtt');
const fs = require('fs');
// Toggle Authentication (set to "ENABLED" or "DISABLED")
const AUTH = "ENABLED";
// ------------ MQTT Broker Configuration -----------
const brokerUrl = 'mqtt://broker-url:1883'; // Replace with your actual broker URL or IP
// MQTT v5 connection parameters
const conn_params = {
clientId: "testclient1", // Unique client ID
protocolVersion: 5, // Use MQTT version 5
clean: true, // Start a clean session
// Uncomment if authentication is needed
// username: "username",
// password: "password",
// Uncomment for TLS connection
// protocol: 'mqtts',
// host: 'broker-url',
// port: 8883,
// ca: [fs.readFileSync('/path/to/root.crt')],
};
let client;
// ------------ MQTT Client Initialization -----------
if (AUTH === "ENABLED") {
client = mqtt.connect(brokerUrl, conn_params);
} else {
client = mqtt.connect(conn_params);
}
// ------------ MQTT Topic & Publish Interval -----------
const topic = 'testtopic';
const publishInterval = 2000; // Publish every 2 seconds
// ------------ On Successful Connection -----------
client.on('connect', (packet) => {
console.log('Connected to MQTT broker using MQTT v5');
// Subscribe with v5 options (optional)
client.subscribe(topic, { qos: 0 }, (err, granted) => {
if (err) {
console.error('Subscription error:', err);
} else {
console.log(`Subscribed to topic: ${topic}`, granted);
}
});
// Payload to publish
const message = JSON.stringify({ temperature: 27.6, humidity: 34 });
// Publish periodically
setInterval(() => {
client.publish(topic, message, {
retain: true,
properties: {
messageExpiryInterval: 60, // Message expires after 60 seconds (MQTT v5 feature)
},
}, (err) => {
if (err) {
console.error('Error publishing message:', err);
} else {
console.log('Message published:', message);
}
});
}, publishInterval);
});
// ------------ Message Received Callback -----------
client.on('message', (topic, message, packet) => {
try {
const data = JSON.parse(message.toString());
console.log(`Received on '${topic}':`, data);
} catch (e) {
console.log(`Received non-JSON message on '${topic}':`, message.toString());
}
});
// ------------ Additional Event Callbacks -----------
client.on('error', (err) => {
console.error('Error:', err);
});
client.on('close', () => {
console.log('Disconnected from MQTT broker');
});
client.on('reconnect', () => {
console.log('Reconnecting to MQTT broker...');
});
client.on('offline', () => {
console.warn('MQTT client is offline');
});
// ------------ Unsubscribe and Disconnect -----------
// client.unsubscribe('testtopic', (err) => {
// if (err) {
// console.error('Unsubscribe error:', err);
// } else {
// console.log('Unsubscribed from topic: testtopic');
// }
// });
// client.end(() => {
// console.log('Disconnected from MQTT broker');
// });
