MQTT PubSub API

The MQTT PubSub client is used to connect to an MQTT broker, subscribe to topics, and publish data. It can be used in combination with an OPC UA server to listen to changes in nodes and publish data to an MQTT broker. It is also possible to use it to publish data to an MQTT broker without an OPC UA server. In this case, you’ll need to set data manually.

MQTT broker selection

OPC UA PubSub over MQTT does not require a specific MQTT broker. The endpointUrl form of uaMqtt:connect() can connect to any MQTT broker that is reachable by the underlying MQTT client, for example a public broker, Mosquitto, HiveMQ, EMQX, a cloud broker, or an on-premises broker.

Real Time Logic also provides an optional Lua MQTT broker for BAS-derived runtimes: LMQTT Broker. This broker is useful when a Mako Server, Xedge, Xedge32, or embedded BAS application should host MQTT locally instead of depending on an external broker.

When using Mako Server for development, the recommended setup is the mako.zip Developer Edition, which includes the Lua MQTT broker module.

The broker also provides an in-process client API. Its broker:createClient() method returns an MQTT-client-like object that can be passed directly to uaMqtt:connect(mqttClient, transportProfileUri). This is useful for local OPC UA PubSub publishers because messages can be routed through the broker without opening a loopback TCP connection.

MQTT Client constructor

ua.newMqttClient()
ua.newMqttClient(config)
ua.newMqttClient(config, uaServer)
ua.newMqttClient(config, model)

Create an instance of MQTT client. MQTT client is used to connect to MQTT broker, subscribe to topics, publish data.

Parameters:

  • config – MQTT client configuration. If not provided, then default configuration will be used.

  • uaServer – OPC UA server instance. If provided, MQTT client can connect to nodes and listen value changes.

  • model – OPC UA model. If provided, it will be used for encoding and decoding data structures.

Returns:

MQTT client instance.

If neither uaServer nor model is provided, the MQTT client loads only the base OPC UA model. In that case, it can encode and decode only base OPC UA data types.

MQTT Connect

uaMqtt:connect(endpointUrl[, transportProfileUri][, connectCallback][, mqttc])
uaMqtt:connect(mqttClient, transportProfileUri[, connectCallback])

Connect to MQTT broker.

Parameters:

  • endpointUrl – MQTT endpoint URL, for example mqtt://localhost:1883.

  • mqttClient – Existing MQTT client instance. If provided, this instance is used instead of creating a new client from endpointUrl. The object must provide the MQTT client methods used by this API, such as publish(), subscribe(), and close(). This includes the in-process client returned by the LMQTT Broker broker:createClient() method.

  • transportProfileUri – OPC UA Transport Profile URI. Defines message encoding. Use ua.TranportProfileUri.MqttBinary or ua.TranportProfileUri.MqttJson.

  • connectCallback – Callback function

  • mqttc – Optional MQTT module/table used when endpointUrl is a string.

If transportProfileUri is omitted, received messages are decoded by inspecting the payload.

The endpointUrl form is the normal choice when connecting over the network to an MQTT broker. The mqttClient form is for advanced cases where the application already created an MQTT client object. With LMQTT Broker, this allows an OPC UA PubSub publisher to use a broker-local client:

When using the mqttClient form, transportProfileUri is required. Use a separate OPC UA MQTT client for each transport profile when subscribing to both JSON and binary/UADP messages through existing MQTT client objects.

local function connectCallback(status)
  print("Connected to MQTT broker")
end

uaMqtt:connect("mqtt://localhost:1883", ua.TranportProfileUri.MqttBinary, connectCallback)

local mqttc = require("mqttc")

local onstatus = function(...)
end

local onpublish = function(topic,payload,properties,cpt)
end

local mqttClient = mqttc.create("mosquitto.org", onstatus, onpublish, {port=1883})
uaMqtt:connect(mqttClient, ua.TranportProfileUri.MqttBinary)

local mqttbroker = require("mqttbroker")
local broker = assert(mqttbroker.create(1883))

local uaMqtt = ua.newMqttClient(config)
local datasetId = uaMqtt:createDataset({
  { name = "Temperature" }
})

uaMqtt:connect(broker:createClient(), ua.TranportProfileUri.MqttJson)
uaMqtt:setValue(datasetId, "Temperature", {
  Type = ua.VariantType.Double,
  Value = 21.5
})
uaMqtt:publish("factory/line1/data", "line1")

MQTT subscribe

uaMqtt:subscribe(topic, messageCallback)

Subscribe to MQTT topic.

Parameters:

  • topic – MQTT topic

  • messageCallback – Callback function

messageCallback(message, err)

Callback function for MQTT message.

Parameters:

  • message – Decoded OPC UA message.

  • err – Decode error, or nil when decoding succeeded.

MQTT publishing

uaMqtt:createDataset(fields[, classId])

Create dataset which will describe message content to publish.

Parameters:

fields – Array of field tables. The order of fields is preserved in published MQTT payload messages.

Field

Type

Description

name

string

Field name used in JSON payloads and binary metadata.

nodeId

NodeId, optional

OPC UA server node whose value changes are published.
Parameters:

classId – Dataset class ID. If omitted, a GUID is generated. The method returns the class ID used for the dataset.

uaMqtt:setValue(classId, id, dataValue)

Set field value in the dataset.

Parameters:

  • classId – Dataset class ID returned by uaMqtt:createDataset().

  • id – Field identifier. This can be a field name, node ID, or field index.

  • dataValue – Field value.

uaMqtt:publish(dataTopic, publisherId)

Publish dataset. Will forcibly publish dataset with all the changed fields.

Parameters:

  • dataTopic – Data topic

  • publisherId – Publisher ID

uaMqtt:startPublishing(dataTopic, publisherId, periodMs)

Start publishing dataset periodically with changed fields.

Parameters:

  • dataTopic – Data topic

  • publisherId – Publisher ID. string or numeric value.

  • periodMs – Period in milliseconds

uaMqtt:stopPublishing()

Stop publishing dataset.

uaMqtt:close()

Stop publishing and close the underlying MQTT client, if one is connected.