MQTT Connector Configuration

• Connector configuration: mqtt.json
  • Section “broker”
    • Subsection “security”
  • Section “mapping”
    • Subsection “converter”
  • Section “connectRequests”
  • Section “disconnectRequest”
  • Section “attributeUpdates”
    • Server side RPC commands
• Next steps

This guide will help you to get familiar with MQTT Connector configuration for NDU Gateway. Use general configuration to enable this Connector. The purpose of this Connector is to connect to external MQTT broker and subscribe to data feed from devices. Connector is also able to push data to MQTT brokers based on the updates/commands from NDU.

This Connector is useful when you have local MQTT broker in your facility or corporate network and you would like to push data from this broker to NDU.

We will describe connector configuration file below.

Connector configuration: mqtt.json

Connector configuration is a JSON file that contains information about how to connect to external MQTT broker, what topics to use when subscribing to data feed and how to process the data. Let’s review the format of the configuration file using example below.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
{
  "broker": {
    "name":"Default Local Broker",
    "host":"192.168.1.100",
    "port":1883,
    "security": {
      "type": "basic",
      "username": "user",
      "password": "password"
    }
  },
  "mapping": [
    {
      "topicFilter": "/sensor/data",
      "converter": {
        "type": "json",
        "deviceNameJsonExpression": "${serialNumber}",
        "deviceTypeJsonExpression": "${sensorType}",
        "timeout": 60000,
        "attributes": [
          {
            "type": "string",
            "key": "model",
            "value": "${sensorModel}"
          }
        ],
        "timeseries": [
          {
            "type": "double",
            "key": "temperature",
            "value": "${temp}"
          },
          {
            "type": "double",
            "key": "humidity",
            "value": "${hum}"
          }
        ]
      }
    },
    {
      "topicFilter": "/sensor/+/data",
      "converter": {
        "type": "json",
        "deviceNameTopicExpression": "(?<=sensor\/)(.*?)(?=\/data)",
        "deviceTypeTopicExpression": "Thermometer",
        "timeout": 60000,
        "attributes": [
          {
            "type": "string",
            "key": "model",
            "value": "${sensorModel}"
          }
        ],
        "timeseries": [
          {
            "type": "double",
            "key": "temperature",
            "value": "${temp}"
          },
          {
            "type": "double",
            "key": "humidity",
            "value": "${hum}"
          }
        ]
      }
    },
    {
      "topicFilter": "/custom/sensors/+",
      "converter": {
        "type": "custom",
        "extension": "CustomMqttUplinkConverter",
        "extension-config": {
            "temperatureBytes" : 2,
            "humidityBytes" :  2,
            "batteryLevelBytes" : 1
        }
      }
    }
  ],
  "connectRequests": [
    {
      "topicFilter": "sensor/connect",
      "deviceNameJsonExpression": "${SerialNumber}"
    },
    {
      "topicFilter": "sensor/+/connect",
      "deviceNameTopicExpression": "(?<=sensor\/)(.*?)(?=\/connect)"
    }
  ],
  "disconnectRequests": [
    {
      "topicFilter": "sensor/disconnect",
      "deviceNameJsonExpression": "${SerialNumber}"
    },
    {
      "topicFilter": "sensor/+/disconnect",
      "deviceNameTopicExpression": "(?<=sensor\/)(.*?)(?=\/disconnect)"
    }
  ],
  "attributeUpdates": [
    {
      "deviceNameFilter": "SmartMeter.*",
      "attributeFilter": "uploadFrequency",
      "topicExpression": "sensor/${deviceName}/${attributeKey}",
      "valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
    }
  ],
  "serverSideRpc": [
    {
      "deviceNameFilter": ".*",
      "methodFilter": "echo",
      "requestTopicExpression": "sensor/${deviceName}/request/${methodName}/${requestId}",
      "responseTopicExpression": "sensor/${deviceName}/response/${methodName}/${requestId}",
      "responseTimeout": 10000,
      "valueExpression": "${params}"
    },
    {
      "deviceNameFilter": ".*",
      "methodFilter": "no-reply",
      "requestTopicExpression": "sensor/${deviceName}/request/${methodName}/${requestId}",
      "valueExpression": "${params}"
    }
  ]
}

Section “broker”

Subsection “security”

Subsection “security” provides configuration for client authorization at Mqtt Broker.

1
2
3
4
5
    "security": {
      "type": "basic",
      "username": "username",
      "password": "password"
    }

1
2
3
    "security": {
      "type": "anonymous"
    }

1
2
3
4
5
  "security":{
    "caCert": "/etc/thingsboard-gateway/ca.pem",
    "privateKey": "/etc/thingsboard-gateway/privateKey.pem",
    "cert": "/etc/thingsboard-gateway/certificate.pem"
  }

Section “mapping”

This configuration section contains array of topics that the gateway will subscribe to after connecting to the broker and settings about processing incoming messages (converter).

The topicFilter supports special symbols: ‘#’ and ‘+’ to allow to subscribe to multiple topics.

Let’s assume we would like to subscribe and process following data from Thermometer devices:

Now let’s review how we can configure JSON converter to parse this data

Subsection “converter”

This subsection contains configuration for processing incoming messages.

Types of mqtt converters:

1. json – Default converter
2. custom – Custom converter (You can write it by yourself, and it will use to convert incoming data from the broker.)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
    {
      "topicFilter": "/sensor/data",
      "converter": {
        "type": "json",
        "deviceNameJsonExpression": "${serialNumber}",
        "deviceTypeJsonExpression": "${sensorType}",
        "timeout": 60000,
        "attributes": [
          {
            "type": "string",
            "key": "model",
            "value": "${sensorModel}"
          }
        ],
        "timeseries": [
          {
            "type": "double",
            "key": "temperature",
            "value": "${temp}"
          },
          {
            "type": "double",
            "key": "humidity",
            "value": "${hum}"
          }
        ]
      }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
    {
      "topicFilter": "/sensor/+/data",
      "converter": {
        "type": "json",
        "deviceNameTopicExpression": "(?<=sensor\/)(.*?)(?=\/data)",
        "deviceTypeTopicExpression": "Thermometer",
        "timeout": 60000,
        "attributes": [
          {
            "type": "string",
            "key": "model",
            "value": "${sensorModel}"
          }
        ],
        "timeseries": [
          {
            "type": "double",
            "key": "temperature",
            "value": "${temp}"
          },
          {
            "type": "double",
            "key": "humidity",
            "value": "${hum}"
          }
        ]
      }
    }

1
2
3
4
5
6
7
8
9
      "converter": {
        "type": "custom",
        "extension": "CustomMqttUplinkConverter",
        "extension-config": {
            "temperatureBytes" : 2,
            "humidityBytes" :  2,
            "batteryLevelBytes" : 1
        }
      }

Note: You can specify multiple mapping objects inside the array.

Mapping process subscribes to the MQTT topics using topicFilter parameter of the mapping object. Each message that is published to this topic by other devices or applications is analyzed to extract device name, type and data (attributes and/or timeseries values). By default, gateway uses Json converter, but it is possible to provide custom converter. See examples in the source code.

Section “connectRequests”

NDU allows sending RPC commands and notifications about device attribute updates to the device. But in order to send them, the platform needs to know if the target device is connected and what gateway or session is used to connect the device at the moment. If your device is constantly sending telemetry data - NDU already knows how to push notifications. If your device just connects to MQTT broker and waits for commands/updates, you need to send a message to the Gateway and inform that device is connected to the broker.

1. Name in a message from broker:

2. Name in topic address:

This section in configuration looks like:

1
2
3
4
5
6
7
8
9
10
  "connectRequests": [
    {
      "topicFilter": "sensors/connect",
      "deviceNameJsonExpression": "${$.SerialNumber}"
    },
    {
      "topicFilter": "sensor/+/connect",
      "deviceNameTopicExpression": "(?<=sensor\/)(.*?)(?=\/connect)"
    }
  ]

In this case following messages are valid:

1
2
mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensors/connect" -m '{"serialNumber":"SN-001"}'
mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensor/SN-001/connect" -m ''

Section “disconnectRequest”

This configuration section is optional.
Configuration, provided in this section will be used to get information from the broker about disconnecting device.
If your device just disconnects from MQTT broker and waits for commands/updates, you need to send a message to the Gateway and inform that device is disconnected from the broker.

1. Name in a message from broker:

2. Name in topic address:

This section in configuration file looks like:

1
2
3
4
5
6
7
8
9
10
  "disconnectRequests": [
    {
      "topicFilter": "sensors/disconnect",
      "deviceNameJsonExpression": "${SerialNumber}"
    },
    {
      "topicFilter": "sensor/+/disconnect",
      "deviceNameTopicExpression": "(?<=sensor\/)(.*?)(?=\/disconnect)"
    }
  ]

In this case following messages are valid:

1
2
mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensors/disconnect" -m '{"serialNumber":"SN-001"}'
mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensor/SN-001/disconnect" -m ''

Section “attributeUpdates”

This configuration section is optional.
NDU allows to provision device attributes and fetch some of them from the device application. You can treat this as a remote configuration for devices. Your devices are able to request shared attributes from NDU. See user guide for more details.

The “attributeRequests” configuration allows configuring the format of the corresponding attribute request and response messages.

This section in configuration file looks like:

1
2
3
4
5
6
7
8
  "attributeUpdates": [
    {
      "deviceNameFilter": "SmartMeter.*",
      "attributeFilter": "uploadFrequency",
      "topicExpression": "sensor/${deviceName}/${attributeKey}",
      "valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
    }
  ]

Server side RPC commands

NDU allows sending RPC commands to the device that is connected to NDU directly or via Gateway.

Configuration, provided in this section uses for sending RPC requests from NDU to device.

This section in configuration file looks like:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
  "serverSideRpc": [
    {
      "deviceNameFilter": ".*",
      "methodFilter": "echo",
      "requestTopicExpression": "sensor/${deviceName}/request/${methodName}/${requestId}",
      "responseTopicExpression": "sensor/${deviceName}/response/${methodName}/${requestId}",
      "responseTimeout": 10000,
      "valueExpression": "${params}"
    },
    {
      "deviceNameFilter": ".*",
      "methodFilter": "no-reply",
      "requestTopicExpression": "sensor/${deviceName}/request/${methodName}/${requestId}",
      "valueExpression": "${params}"
    }
  ]

As you can use deviceNameFilter and methodFilter to apply different mapping rules for different devices/methods. Once Gateway receives RPC request from the server to the device, it will publish the corresponding message based on requestTopicExpression and valueExpression. In case you expect the reply to the request from device, you should also specify responseTopicExpression and responseTimeout. The Gateway will subscribe to the “response” topic and wait for device reply until “responseTimeout” is detected (in milliseconds).

Example of RPC request (rpc-request.json) that need to be sent from the server:

1
2
3
4
5
6
{
  "method": "echo",
  "params": {
    "message": "Hello!"
  }
}

Next steps

Explore guides related to main NDU features:

• Data Visualization - how to visualize collected data.
• Device attributes - how to use device attributes.
• Telemetry data collection - how to collect telemetry data.
• Using RPC capabilities - how to send commands to/from devices.
• Rule Engine - how to use rule engine to analyze data from devices.