MQTT Sparkplug Protocol

The MQTT Datasource can be used to receive and send messages in line with the Sparkplug protocol, using the Lua Callback options.

This section addresses the following topics:

Message Encoding and Decoding

The embedded syslib.sparkplugb Lua library contains functions to decode and encode sparkplug message payloads (encoded using the Google Protobuf data format) for subscription and publishing respectively to an MQTT Sparkplug broker. For information regarding the compression/decompression of the payload, see the Lua Scripting page on Compressing Data.

In order to receive messages in the Sparkplug binary format, they can be decoded to Lua table format using the decode_payload function:

local sp = require("syslib.sparkplugb")

local payload, ok = sp:decode_payload(payload_binary)   --payload_binary variable contains payload data in the binary format

-- if the call succeeds (ok == true) then payload is the Lua table containing sparkplug data. Otherwise, when ok==false payload (first result) will contain error message.

You can also call using a path to a file that contains the binary payload:

local sp = require("syslib.sparkplugb")

local payload, ok = sp:decode_payload(path_to_file, true)   --in this case path_to_file variable contains the path to file on the server containing payload binary data.

-- if the call succeeds (ok == true) then payload is the Lua table containing sparkplug data. Otherwise, when ok==false payload (first result) will contain error message.

Here is an example that can be added to the On Message Receive property of an MQTT Subscriber I/O Item:

local SPB = require("syslib.sparkplugb")
local JSON = require "rapidjson"
local VAR = require "esi-variables"

-- Example function to process Sparkplug messages
return function(guid, mid, topic, payload, qos, retain)

  local payload, ok = SPB:decode_payload(payload, false)

  local result = nil
  if ok then
    if payload.metrics then

      -- parse timestamp from sparkplug message
      local posixTs = tonumber(payload.timestamp)
      assert(posixTs, "Invalid timestamp")
      local timestamp = syslib.gettime(posixTs)

      -- foreach metric, get values and relative path
      for _, metricObj in pairs(payload.metrics) do
        local relPath = metricObj.name
        local value = nil
        for k,v in pairs(metricObj) do
          if k:find("value") then
            value = v
          end
        end
        VAR:SET(topic .. "/" .. relPath,value, 0, timestamp)
      end
    end
    result = JSON.encode({ok=true, result=payload})
  else
    result = JSON.encode({ok=false,err=payload})
  end

  -- return full payload as return for the IO-Item
  local targetObjectPath = nil -- the subscriber IO-Item itself will be used
  return targetObjectPath, result, 0, syslib.currenttime()

end

Payload Compression

The Sparkplug B payload can be compressed (and decompressed) using the deflate (ZLIB RFC 1950) or the gzip algorithm. For more information see Compressing Data.

Sparkplug Message Validation

The following functions can be used to validate that MQTT messages received from a broker are compliant with Sparkplug:

Functions

validate_payload(payload, message_type)

Checks if an MQTT paylod is valid for a specific message type.

Description
Parameters
Name Type Description

payload

string

Sparkplug B payload; for message_type STATE the payload needs to be JSON format. For all other message types the payload needs to be in binary format

message_type

string

Options are:

  • NBIRTH – Birth certificate for Sparkplug Edge Nodes

  • NDEATH – Death certificate for Sparkplug Edge Nodes

  • DBIRTH – Birth certificate for Devices

  • DDEATH – Death certificate for Devices

  • NDATA – Node data message

  • DDATA – Device data message

  • NCMD – Node command message

  • DCMD – Device command message

  • STATE – Sparkplug Host Application state message
Returns

Name

Type

Description

ok

boolean

true if the payload is valid for the specified message_type; otherwise false

message

string

error message

parse_nd_topic(mqtt_topic)

Description

Checks if MQTT topics are valid for the message types NBIRTH, NDEATH , DBIRTH, DDEATH, NDATA, DDATA, NCMD, and DCMD. If the topic is valid, the exact message_type can be identified in the returned result table. For the validation against STATE messages, see the parse_state_topic(mqtt_topic) function below.

Parameters
Name Type Description

mqtt_topic

string

an MQTT sparkplug B topic of the form "namespace/group_id/message_type/edge_node_id/[device_id] for Edge Node messages
Returns
Name Type Description

ok

boolean

true if the parsing of the mqtt_topic succeeded; otherwise false

result

string/table

If ok is false, this contains the error message;
If ok is true, this contains a Lua table of following structure:

for edge node topics
{
  "is_sparkplug_namespace":true,
  "has_all_sparkplug_elements":true,
  "message_type":"NBIRTH",
  "group_id":"Group1",
  "is_known_message_type":true,
  "namespace":"spBv1.0",
  "edge_node_id":"EdgeNode1"
}
for device topics
{
  "is_sparkplug_namespace":true,
  "has_all_sparkplug_elements":true,
  "message_type":"NBIRTH",
  "group_id":"Group1",
  "is_known_message_type":true,
  "namespace":"spBv1.0",
  "edge_node_id":"EdgeNode1",
  "device_id":"device1"
}

parse_state_topic(mqtt_topic)

Description

Checks if an MQTT topic is a valid STATE type message. For the validation against all other message types, see the parse_nt_topic(mqtt_topic) function above.

Parameters
Name Type Description

mqtt_topic

string

an MQTT Sparkplug B STATE topic of the form "spBv1.0/STATE/host_application_id"
Returns
Name Type Description

ok

boolean

true if the parsing of the mqtt_topic succeeded; otherwise false

result

string

If ok is false, this contains the error message;
If ok is true, this contains one the following Lua table:

{
  "namespace":"spBv1.0",
  "is_sparkplug_namespace":true,
  "has_all_sparkplug_elements":true,
  "is_state_topic":true,
  "sparkplug_host_id":"host_application_id"
}

Example for Sparkplug Message Validation

local sp = require("syslib.sparkplugb")
local j = require("rapidjson")

return function(guid, mid, topic, payload, qos, retain)
  local result, ok =  sp:parse_nd_topic(topic)
  if ok then
      if result.is_sparkplug_namespace and result.edge_node_id ~= nil and result.is_known_message_type then
          -- this is Edge Node message and payload should have binary protobuf format
          local _, vok = sp:validate_payload(payload, result.message_type)
          if vok then
              local result1, ok1 = sp:decode_payload(payload)
              if ok1 then
                  -- do something with result1
              end
          end
      end
      elseif result.is_sparkplug_namespace then
          local result2, ok2 =  sp:parse_state_topic(topic)
          if ok2 and result2.is_sparkplug_namespace and result2.sparkplug_host_id ~= nil and result2.is_state_topic and result2.has_all_sparkplug_elements then
              --  this is STATE message and payload should have JSON format
              local _, vok = sp:validate_payload(payload, result.message_type)
                  if vok then
                    -- and first decode JSON to Lua table
                      local result1= j.decode(payload)
                      if result1 then
                          -- do something with result1
                      end
              end
          end
      end
  end
end