Using MQTT API to get reporting data (C++ example)

Overview

The SICON.OS MQTT API provides real time notifications that you can process in your application. MQTT is a tcp based protocol follows the principles of a subscriber/publisher pattern. Your application can connect to the MQTT Broker of SICON.OS from anywhere in the network. After you subscribed to a topic like events or reports, your application is notified of new data as it is published.

In this guide, we show how you can connect to the SICON.OS MQTT API to retrieve reporting data.

Reporting Data

A report is a collection of device data that gets stored to the SICON.OS database and published to the MQTT broker in short intervals. Depending on the device configuration, this can happen regularly or process triggered. For example, when a work cycle was completed.

The message format is JSON - a universal data format for exchanging key/value pairs.

Lets look at one message. At 10:00:05 the leakage rate was at 57 mbar/s. SICON.OS gathers the data and publishes it on topic device/78/postReport. The topic works like an address of folders.

device -> All device data

device/78 -> All data from a specific device

device/78/postReport -> Only reports from device 78

We can decide later whether to listen to multiple devices or only to a specific device.

You can choose to also listen to all devices with MQTTs topic placeholders. # stands for the entire subtree, whereas + is used for only one segment.

For example: device/+/postReport subscribes to all devices.

If you want to see all available topics, please refer to the SICON.OS MQTT API reference guide.

Download the example

This guide uses Visual Studio to configure and run the project. You can download it for free from Microsoft.

We pre-built and combined all required sources in our git repository for you to include. It also contains a ready to run example.

If you do not have git installed, you can download a zip file from the git repository.

BASH

$ git clone https://github.com/GPS-GmbH/sicon-mqtt-example-cpp.git

Open the mqtt-api-demo.sln file. This opens the project in Visual Studio.

In the header section, all dependencies are included.

CPP

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"
#include "jsmn.h"

#if !defined(_WIN32)
    #include <t.h>
#else
    #include <windows.h>
#endif

#if defined(_WRS_KERNEL)
    #include <OsWrapper.h>
#endif

Among the few standard libraries already included on your system, there are two external dependencies we need to set up.

Paho MQTT Client - A library to set up a client and subscribe to topics.
JSMN - A JSON parser library used to extract specific fields from the message mentioned above.

Linking the libraries

In order to access these external dependencies, they need to be linked to your project.

For better portability, these files should reside inside the project folder. But it is not required to have them at the same place, as long as these dependencies are available in your system.

First obtain the header, dll and lib files.

Lastly, drag and drop the following files to your header files section in Visual Studio.

jsmn.h
MQTTAsync.h
MQTTClient.h
MQTTClientPersistance.h
MQTTExportDeclarations.h
MQTTProperties.h
MQTTReasonCodes.h
MQTTSubscribeOpts.h
paho-mqtt3a.dll
paho-mqtt3a.lib
paho-mqtt3c.dll
paho-mqtt3c.lib

Define Constants

Copy the contents below to your main source file.

CPP

#define ADDRESS     "tcp://10.8.0.74:1883"
#define CLIENTID    "ExampleClientSub"
#define TOPIC       "device/+/postReport"
 // The device reporting config defines which Y value has the leakage rate
#define REPORTINGIDENTIFIER_EJEKTOR1       "Y_1"
#define REPORTINGIDENTIFIER_EJEKTOR2       "Y_2"
#define REPORTINGIDENTIFIER_EJEKTOR3       "Y_3"
#define REPORTINGIDENTIFIER_EJEKTOR4       "Y_4"
#define REPORTINGIDENTIFIER_COUNTER1       "X_1"
#define REPORTINGIDENTIFIER_COUNTER2       "X_2"
#define REPORTINGIDENTIFIER_COUNTER3       "X_3"
#define REPORTINGIDENTIFIER_COUNTER4       "X_4"
#define QOS         1

// global state

int disc_finished = 0;
int subscribed = 0;
int finished = 0;

Here you can change IP Address, Device ID and Reporting Identifiers.

As address you can choose a hostname or IP address to your SICON.OS installation.

CPP

// Example for Hostname
#define ADDRESS     "tcp://siconos-2033.local:1883"
// Or alternatively IP Address - use the SICON.OS IT or IIT adress
#define ADDRESS     "tcp://192.168.77.10:1883"

Finding the correct Device ID

Navigate to the IP address or hostname in your browser. Enter your credentials and click “Login”.

Choose the device which you want to read the report, e.g. the leakagerate, from the device list.

The address bar now reveals the ID as the last parameter.

Set the topic to include this ID. For example: device/16/postReport.

CPP

#define TOPIC       "device/16/postReport"

Reporting identifiers

The reporting identifiers are defined for the leakage rate of an SCTSi vacuum terminal.

If that is your application usecase, you don’t have to change anything. Otherwise you can check the device description files reporting config (SDD) to see which indices are mapped to which X or Y value.

Using the SICON.OS REST API, you can programatically retrieve a list of devices and their reporting config, to filter out relevant fields.

Setting up an MQTT Client connection

For handling the complete lifecycle of an MQTT connection, we need to set up callback handlers.

onConnect - connection established, set up a subscription
onConnectFailure - connection can't be established
onConnectionLost - the active connection was terminated by network change or lost connection - here we try to reconnect
onDisconnect - the active connection was terminated programatically
onDisconnectFailure - the active connection was not able to be terminated gracefully
onMessageArrive - After successfull connect and subscribe, handle how messages are

CPP

void onConnect(void* context, MQTTAsync_successData* response) {
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	// code returned by various mqtt functions to check for errors
	int returnCode;

	printf("Successful connection\n");

	printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
		"Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
	opts.context = client;
	if ((returnCode = MQTTAsync_subscribe(client, TOPIC, QOS, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start subscribe, return code %d\n", returnCode);
		finished = 1;
	}
}

void onConnectFailure(void* context, MQTTAsync_failureData* response) {
	printf("Connect failed, returnCode %d\n", response->code);
	finished = 1;
}

void onConnectionLost(void* context, char* cause) {
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	int returnCode;

	printf("\nConnection lost\n");
	if (cause)
		printf("     cause: %s\n", cause);

	printf("Reconnecting\n");
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	if ((returnCode = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", returnCode);
		finished = 1;
	}
}

void onDisconnect(void* context, MQTTAsync_successData* response) {
	printf("Successful disconnection\n");
	disc_finished = 1;
}

void onDisconnectFailure(void* context, MQTTAsync_failureData* response) {
	printf("Disconnect failed, returnCode %d\n", response->code);
	disc_finished = 1;
}

// helper function to compare json keys
static int jsoneq(const char* json, jsmntok_t* tok, const char* s) {
	if (tok->type == JSMN_STRING && (int)strlen(s) == tok->end - tok->start &&
		strncmp(json + tok->start, s, tok->end - tok->start) == 0) {
		return 0;
	}
	return -1;
}

int onMessageArrive(void* context, char* topicName, int topicLen, MQTTAsync_message* message) {
	jsmn_parser p;
	jsmntok_t t[128]; /* We expect no more than 128 JSON tokens */
	jsmn_init(&p);
	printf("Message arrived\n");
	printf("     topic: %s\n", topicName);
	char* payload = (char*)message->payload;
	printf("   message: %.*s\n", message->payloadlen, payload);
	int r = jsmn_parse(&p, payload, strlen(payload), t, 128);
	for (int i = 1; i < r; i++) {
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_EJEKTOR1) == 0) {
			printf("leakagerate E1: %.*s mbar/s\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_EJEKTOR2) == 0) {
			printf("leakagerate E2: %.*s mbar/s\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_EJEKTOR3) == 0) {
			printf("leakagerate E3: %.*s mbar/s\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_EJEKTOR4) == 0) {
			printf("leakagerate E4: %.*s mbar/s\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_COUNTER1) == 0) {
			printf("Global Cycle Counter E1: %.*s pcs\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_COUNTER2) == 0) {
			printf("Global Cycle Counter E2: %.*s pcs\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_COUNTER3) == 0) {
			printf("Global Cycle Counter E3: %.*s pcs\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
		if (jsoneq(payload, &t[i], REPORTINGIDENTIFIER_COUNTER4) == 0) {
			printf("Global Cycle Counter E4: %.*s pcs\n", t[i + 1].end - t[i + 1].start, payload + t[i + 1].start);
		}
	}

	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);
	return 1;
}

Hook up event handlers and create the client

In the main function, the MQTT client is being created, configured and attempts to connect.

In every step, we validate for successful configuration.

CPP

int main(int argc, char* argv[])
{
	MQTTAsync client;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onConnect;
	conn_opts.onFailure = onConnectFailure;
	MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
	disc_opts.onSuccess = onDisconnect;
	disc_opts.onFailure = onDisconnectFailure;
	int returnCode;
	int terminalInputChar;

	// Create a new client
	if ((returnCode = MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL))
		!= MQTTASYNC_SUCCESS)
	{
		// validation of options (address, clientid) resulted in an error
		printf("Failed to create client, return code %d\n", returnCode);
		returnCode = EXIT_FAILURE;
		return returnCode;
	}
	
	conn_opts.context = client;

	if ((returnCode = MQTTAsync_setCallbacks(client, client, onConnectionLost, onMessageArrive, NULL)) != MQTTASYNC_SUCCESS)
	{
		// validation of callbacks resulted in an error
		printf("Failed to set callbacks, return code %d\n", returnCode);
		returnCode = EXIT_FAILURE;
		MQTTAsync_destroy(&client);
	}

	// actual connection to the broker
	if ((returnCode = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", returnCode);
		returnCode = EXIT_FAILURE;
		MQTTAsync_destroy(&client);
	}

	while (!subscribed && !finished)
#if defined(_WIN32)
		Sleep(100);
#else
		usleep(10000L);
#endif

	if (finished) return returnCode;
	
	// wait for user input loop to abort the program
	do {
		terminalInputChar = getchar();
	} while (terminalInputChar != 'Q' && terminalInputChar != 'q');

	// user aborted - now gracefully disconnect and free the client
	if ((returnCode = MQTTAsync_disconnect(client, &disc_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start disconnect, return code %d\n", returnCode);
		returnCode = EXIT_FAILURE;
		MQTTAsync_destroy(&client);
	}
	while (!disc_finished)
	{
#if defined(_WIN32)
		Sleep(100);
#else
		usleep(10000L);
#endif
	}
}

Save and Compile the project

Save the file. Afterwards start the debugger in the toolbar. Alternatively you can press CTRL+F5.

This creates an executable named mqtt-api-demo.exe in the root of the project and executes it immediately after.

Building for Release

If you want to share the executable, make sure to add the dll files along the executable.