MQTT PubSub Examples

These examples are standalone PubSub API recipes. They use a local MQTT broker so they can run without an external broker such as Mosquitto.

The easiest setup is the Mako Server mako.zip Developer Edition, which includes the broker module used by these examples.

Run them from the LSP-Examples/OPC-UA/pubsub directory:

mako mqtt_publish_node_subscribe.lua
mako mqtt_publish_serverless.lua
mako mqtt_subscribe.lua

The examples stop immediately with a clear error message if the mqttbroker module cannot be loaded.

If you are learning OPC UA PubSub for the first time, start with Learning OPC UA with Mako Server. The examples below are more compact API-focused recipes.

Monitoring OPC UA server changes

It is possible to connect an MQTT PubSub publisher to an OPC UA server and publish node value changes. This example creates a local OPC UA server, creates a local MQTT broker, publishes a server node value through MQTT PubSub, and verifies the decoded subscriber message.

local ua = require("opcua.api")

-- create server
local uaServer = ua.newServer()
uaServer:initialize()

local ObjectsFolder = "i=85"
local int32DataValue = {
  Type = ua.VariantType.UInt32,
  Value = 10
}
local request = {
  NodesToAdd = {ua.newVariableParams(ObjectsFolder, "writeHook", int32DataValue)}
}

-- Add a node
local resp = uaServer:addNodes(request)
local results = resp.Results
assert(results[1].StatusCode, ua.StatusCode.Good)
local nodeId =  results[1].AddedNodeId

-- Create MQTT client
local config = {
  bufSize = 128 -- max size of MQTT message
}

local uaMmqtt = ua.newMqttClient(config, uaServer)

-- Array with fields parameters.
local fields = {
  -- #1
  {
    nodeId = nodeId,   -- ID of a node, which changes will be monitored
    name = "MqttNode", -- Name of a field, will as a field in JSON
  }
}

-- create dataset with fields
local classId = "5fa38ebb-44d2-a3ec-d251-1030c777f10a"
uaMmqtt:createDataset(fields, classId)

-- Connect to MQTT broker
local tranportProfileUri = ua.TranportProfileUri.MqttJson
local endpointUrl = "opc.mqtt://test.mosquitto.org:1883"
uaMmqtt:connect(endpointUrl, tranportProfileUri)

-- Start periodic publishing
local dataTopic = "rtl/json/data/urn:arykovanov-note:opcua:server/group/dataset"
uaMmqtt:startPublishing(dataTopic, "test_cyclic_publisher", 2000)

-- Run server.
uaServer:run()

-- Function which will periodically write data to address space
-- Those changes will be hooked by MQTT client for publishing data.
local writeRequest = {
  NodesToWrite = {
    {
      NodeId = nodeId,
      AttributeId = ua.AttributeId.Value,
      Value = {
        Type = ua.VariantType.UInt32,
        Value=123
      }
    }
  }
}


uaServer:write(writeRequest)

uaMmqtt:stopPublishing()
uaServer:shutdown()

Full source

Publishing data to MQTT broker

It is possible to publish data to an MQTT broker without an OPC UA server. To do this, create an MQTT PubSub client, configure fields for messages, connect it to the local MQTT broker, and publish manually supplied values.

local ua = require("opcua.api")

local uaMqtt = ua.newMqttClient()

local fields = {
  { name = "Value1" },
  { name = "Value2" }
}

local tranportProfileUri = ua.TranportProfileUri.MqttBinary
local endpointUrl = "opc.mqtt://test.mosquitto.org:1883"
uaMqtt:connect(endpointUrl, tranportProfileUri)

local datasetId = uaMqtt:createDataset(fields)

local dataTopic = "rtl/uadp/data/urn:arykovanov-note:opcua:server/group/dataset"
local publisherId = "test_manual_publisher"

for i=64, 74 do
  uaMqtt:setValue(datasetId, "Value1", {Type=ua.VariantType.UInt32, Value=i})
  uaMqtt:setValue(datasetId, "Value2", {Type=ua.VariantType.UInt32, Value=i*2})
  uaMqtt:publish(dataTopic, publisherId)

  ba.sleep(1000)
end

Full source

Subscribing to MQTT messages

OPC UA PubSub messages can be received and decoded by OPC UA MQTT client. This example starts a local broker, publishes one JSON message and one binary UADP message, and subscribes to both. When an existing MQTT client object is passed to uaMqtt:connect(...), the transport profile must be explicit, so the example uses one JSON subscriber and one binary subscriber.

local ua = require("opcua.api")

-- Create MQTT client instance
local mqttClient = ua.newMqttClient()

-- Connect to MQTT broker
local function callbackCallback(status)
  ua.printTable("status", status)
end
mqttClient:connect("opc.mqtt://test.mosquitto.org:1883", callbackCallback)

-- The only message callback for both JSON and binary data
local function messageCallback(payload, err)
  if err then
    print("Error:" .. tostring(err))
  end
  ua.printTable("payload", payload)
end

-- Subscribe on a topic with binary data
mqttClient:subscribe("rtl/uadp/data/urn:arykovanov-note:opcua:server/group/dataset", messageCallback)
-- Subscribe on a topic with JSON data
mqttClient:subscribe("rtl/json/data/urn:arykovanov-note:opcua:server/group/dataset", messageCallback)

-- Wait for some time
local cnt=1
while cnt <= 3 do
   ba.sleep(1000)
   cnt = cnt+1
end

Full source