How to Connect IoT Devices via MQTT & Node-RED to Kaa Cloud

In the Internet of Things (IoT) domain, establishing connections between devices and the cloud is paramount for facilitating data collection, processing, and analysis. Despite the prevalence of HTTP in device-to-cloud communication, MQTT has emerged as a noteworthy alternative, offering lightweight, efficient, and real-time communication.

This extensive tutorial will explore the intricacies of connecting devices via MQTT using Node-RED, a widely used open-source visual programming tool. Our destination is the communication with Kaa IoT Platform, a robust and scalable solution for IoT applications.

Also read the Tutorial: Tutorial: Connecting Devices to the IoT Platform with Node-RED & MQTT

Challenge

Our earlier tutorial delved into the "node red http" scenario where data was received through HTTP. However, a more typical case that involves devices sending data via MQTT. The key distinction often lies in the data's representation format, which may differ from JSON or be presented as an array, requiring parsing before publication to the platform. Numerous other scenarios may necessitate data preprocessing. It's important to emphasize that for seamless automatic parsing by the platform, the data needs to conform to a simple JSON structure.

Example of the message

For ease of understanding, let's examine a scenario similar to the one discussed in the Connecting Devices via HTTP using Node-RED to Kaa IoT Platform tutorial. However, in this instance, we assume that the device is capable of publishing data via MQTT.

As a reminder, in our example, we have a pressure sensor that periodically sends multiple readings within a single data packet:

{
  "DeviceType": "pressure_sensor",
  "DeviceId": "PS875035",
  "MessageType": "batch_msg",
  "Data": {
    "start": "2023-12-15 08:00:00",
    "interval": 600,
    "values": [
      35,
      37,
      37,
      38,
      40,
      39,
      42,
      43,
      45,
      32
    ]
  }
}

In this message, we have the following parameters:

• DeviceType: the sensor type;
• DeviceId: sensor ID;
• MessageType: message type. In our case, it is a batch message (a message that contains some amount of data);
• start: this field contains the date-time of the first measured value;
• interval: the interval between the measurements;
• values: contains an array of measured values.

The measured data needs to be organized into a time series to facilitate displaying data through timestamped graphs.

To achieve this, two parameters are necessary:

• The data itself.
• Timestamp: a timestamp indicating which point in time the data relates to.

If you're not acquainted with the nuances of timestamp processing on the Kaa IoT platform, we suggest referring to the explanation provided in the preceding tutorial.

Node-RED flow configuration

In the role of the primary data processor, we will use Node-RED hosting.

To facilitate the receiving of data through MQTT, the Node-RED flow must incorporate an MQTT broker. Alternatively, the MQTT broker can be deployed externally, and Node-RED will be just a client of this broker.

For our MQTT broker, we will use node-red-contrib-aedes. Once this broker is installed, a new node becomes available in our configuration:

Let's add this node to our flow:

Leave the node configuration as default:

Now, it's essential to include an MQTT client that will receive data from the broker for further processing:

Client configuration (subscribe to all topics):

Currently, we have a working MQTT broker and client. The MQTT broker is available at:

• URL: <tenant-id>.nodered.kaaiot.com (where <tenant-id> is your tenant ID)
• Port: 1883

To check that the data is coming, let's send a message to the data/<requestId> topic:

mosquitto_pub -h <tenant-id>.nodered.kaaiot.com -p 1883 -t 'data/12345' -m '{"DeviceType":"pressure_sensor","DeviceId":"PS875035","MessageType":"batch_msg","Data":{"start":"2023-12-15 08:00:00","interval":600,"values":[35,37,37,38,40,39,42,43,45,32]}}'

In the debug window, we can see our message:

Now, let's create the function node where we'll parse the HTTP body, replicating the process from the HTTP tutorial.

The goal is to transform the original message into an array of messages, each featuring its own timestamp and value.

var messagesToSend = [];    // messages to send to the Kaa platform
if (msg.payload.DeviceType == "pressure_sensor") {
   var firstMeasurementTimestamp = new Date(msg.payload.Data.start).getTime();
   var measurementInterval = msg.payload.Data.interval;
   for (var i = 0; i < msg.payload.Data.values.length; i++) {
      
       var message = {};
       message.payload = {};
       message.payload.timestamp = firstMeasurementTimestamp + i* (measurementInterval * 1000); // timestamp in milliseconds
       message.payload.value = msg.payload.Data.values[i];
       message.token = msg.payload.DeviceId;
       messagesToSend.push(message);
   }
  
} else {
   node.warn(`Unknown device type ${msg.DeviceType}`);
   return null;
}
return [messagesToSend];

To publish data to the Kaa IoT platform, we can utilize nodes from the Kaa library, specifically the Kaa Timeseries node and Kaa MQTT Connector node for time-series data publication.

Let's begin by configuring the Kaa MQTT Connector node. We should set the Application, Application version, and Client ID for proper integration:

Alright, as a result, we have the next flow:

As the final step of the tutorial, if we navigate to the Device Management section on the Kaa IoT platform and inspect our endpoint, we will be able to view the data:

The source code for the Node-RED flow:

[
    {
        "id": "938f051db4d7c7aa",
        "type": "tab",
        "label": "MQTT_tutorial",
        "disabled": false,
        "info": "",
        "env": []
    },
    {
        "id": "9f51d2c87bc724c0",
        "type": "aedes broker",
        "z": "938f051db4d7c7aa",
        "name": "",
        "mqtt_port": "1883",
        "mqtt_ws_bind": "port",
        "mqtt_ws_port": "",
        "mqtt_ws_path": "",
        "cert": "",
        "key": "",
        "certname": "",
        "keyname": "",
        "persistence_bind": "memory",
        "dburl": "",
        "usetls": false,
        "x": 130,
        "y": 100,
        "wires": [
            [
                "d078d2f97cedf88d"
            ],
            [
                "cf12e1ac60b30448"
            ]
        ]
    },
    {
        "id": "d078d2f97cedf88d",
        "type": "debug",
        "z": "938f051db4d7c7aa",
        "name": "Broker Status Events",
        "active": false,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "true",
        "targetType": "full",
        "statusVal": "",
        "statusType": "auto",
        "x": 360,
        "y": 60,
        "wires": []
    },
    {
        "id": "cf12e1ac60b30448",
        "type": "debug",
        "z": "938f051db4d7c7aa",
        "name": "Broker Publish Events",
        "active": false,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "true",
        "targetType": "full",
        "statusVal": "",
        "statusType": "auto",
        "x": 360,
        "y": 140,
        "wires": []
    },
    {
        "id": "b0248f8755ceb630",
        "type": "mqtt in",
        "z": "938f051db4d7c7aa",
        "name": "Subscriber",
        "topic": "#",
        "qos": "0",
        "datatype": "auto-detect",
        "broker": "cc744aa791563a4f",
        "nl": false,
        "rap": true,
        "rh": 0,
        "inputs": 0,
        "x": 100,
        "y": 260,
        "wires": [
            [
                "f7ed0eecc86cb812",
                "96859fa0fd3f2c18"
            ]
        ]
    },
    {
        "id": "f7ed0eecc86cb812",
        "type": "debug",
        "z": "938f051db4d7c7aa",
        "name": "Published events",
        "active": true,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "true",
        "targetType": "full",
        "statusVal": "",
        "statusType": "auto",
        "x": 310,
        "y": 200,
        "wires": []
    },
    {
        "id": "96859fa0fd3f2c18",
        "type": "function",
        "z": "938f051db4d7c7aa",
        "name": "parse MQTT body",
        "func": "var messagesToSend = [];    // messages to send to the Kaa platform\n\nif (msg.payload.DeviceType == \"pressure_sensor\") {\n\n    var firstMeasurementTimestamp = new Date(msg.payload.Data.start).getTime();\n    var measurementInterval = msg.payload.Data.interval;\n\n    for (var i = 0; i < msg.payload.Data.values.length; i++) {\n        \n        var message = {};\n        message.payload = {};\n        message.payload.timestamp = firstMeasurementTimestamp + i* (measurementInterval * 1000); // timestamp in milliseconds\n        message.payload.value = msg.payload.Data.values[i];\n        message.token = msg.payload.DeviceId;\n\n        messagesToSend.push(message);\n    }\n    \n} else {\n    node.warn(`Unknown device type ${msg.DeviceType}`);\n    return null;\n}\n\nreturn [messagesToSend];",
        "outputs": 1,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "libs": [],
        "x": 310,
        "y": 260,
        "wires": [
            [
                "26a2d5fc0d5db2ac",
                "b44c02c371448fd8"
            ]
        ]
    },
    {
        "id": "26a2d5fc0d5db2ac",
        "type": "debug",
        "z": "938f051db4d7c7aa",
        "name": "messages to KaaIoT",
        "active": true,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "true",
        "targetType": "full",
        "statusVal": "",
        "statusType": "auto",
        "x": 580,
        "y": 200,
        "wires": []
    },
    {
        "id": "b44c02c371448fd8",
        "type": "kaa-timeseries",
        "z": "938f051db4d7c7aa",
        "x": 560,
        "y": 260,
        "wires": [
            [
                "e16e6166dd03320d"
            ]
        ]
    },
    {
        "id": "e16e6166dd03320d",
        "type": "kaa-connector",
        "z": "938f051db4d7c7aa",
        "connection_type": "MQTT",
        "domain": "cloud.kaaiot.com",
        "application": "cf2p1cmgthh95ci5h4kg",
        "version": "v1",
        "client_id": "mqttClient",
        "x": 780,
        "y": 260,
        "wires": [
            []
        ]
    },
    {
        "id": "cc744aa791563a4f",
        "type": "mqtt-broker",
        "name": "",
        "broker": "127.0.0.1",
        "port": "1883",
        "clientid": "",
        "autoConnect": true,
        "usetls": false,
        "protocolVersion": "4",
        "keepalive": "60",
        "cleansession": true,
        "birthTopic": "",
        "birthQos": "0",
        "birthPayload": "",
        "birthMsg": {},
        "closeTopic": "",
        "closeQos": "0",
        "closePayload": "",
        "closeMsg": {},
        "willTopic": "",
        "willQos": "0",
        "willPayload": "",
        "willMsg": {},
        "userProps": "",
        "sessionExpiry": ""
    }
]

Conclusion

Congratulations on successfully establishing the connection between your devices and the Kaa IoT Platform through the utilization of Node-RED! By leveraging the seamless integration capabilities of Node-RED along with the robust functionalities provided by Kaa, you've opened up a realm of possibilities for your IoT projects.

Seize the potential of Kaa's scalable infrastructure, comprehensive data management tools, and secure communication protocols to propel your IoT initiatives to unprecedented levels. Dive into the advantages of Kaa Cloud today and witness how it can streamline your device connectivity, data processing, and application development. Embrace the future of IoT innovation with Kaa as your gateway to a smarter, connected world.

Should you encounter any challenges, feel free to contact us and connect your device with the support of our engineers.