MQTT Client Protocol (Message Queue Telemetry Transport)

MQTT 3.1.1

MQTT 5.0

Example:

• for an Edge Node station with the address myGroup/myEdgeNode, a SUBSCRIBE message is sent to the topic spBv1.0/myGroup/NCMD/myEdgeNode
• for a Device station with the address myGroup/myEdgeNode/myDevice1, a SUBSCRIBE message is sent to the topic spBv1.0/myGroup/DCMD/myEdgeNode/myDevice1

Note: Payload Type = Sparkplug Edge Node, a SUBSCRIBE message is automatically sent to the topic spBv1.0/STATE/<primaryHostApplication> if the Sparkplug Host ID parameter is specified (<primaryHostApplication> is the value of this parameter), so that the D2000 KOM process is informed about the connection/disconnection of the Primary Host Application. If a STATE message about the Primary Host Application disconnection is received (STATE message with online=false) and multiple IP addresses are specified in the line configuration, the D2000 KOM process disconnects from the current MQTT server and tries other MQTT servers until it finds one to which the Primary Host Application is connected (STATE message with online=true).

Unique client identifier (Client Identifier) sent within the CONNECT message.
Note: it is possible to enter a blank string - in which case the server can assign a unique name to the client (if it supports such functionality) or return an error. However, if the Client ID is not specified, the Clean Session Flag parameter settings will be ignored (as the server will assign a unique name each time).

The tested MQTT server (thethings.network) returned an error if the Client ID was blank and Clean Session Flag=NO.

Note: Some MQTT brokers (PIXII.COM, Eclipse Mosquitto) identified clients only by Client ID. In practice, this meant that two different D2000 systems that connected to the same broker were considered as one client, and the broker closed an existing connection that it considered old when a new connection was established, or it did not allow a new connection to be created and returned the error Connection Refused, identifier rejected (2).
After setting the Client ID to a unique value, the communications started to work without connection breakdowns.

Parameter Clean Session Flag of the CONNECT message. The NO value means that the server uses the current session state (connection) - e. g. after the collapse and recovery of the TCP connection. This means that all unconfirmed PUBLISH messages with QoS_1 and QoS_2 are resent (optionally also QoS_0, depending on the implementation).
The YES value means that the session is re-created, and unconfirmed PUBLISH messages are not repeated.
Note: If Payload Type = Sparkplug Host and the Sparkplug Host ID parameter is not empty, this parameter is ignored, and Clean Session Flag is set to YES (see the description of the Sparkplug Host ID parameter).
Note: If Payload Type = Sparkplug Edge Node, this parameter is ignored, and Clean Session Flag is set to YES (see the description of the Sparkplug Host ID parameter).

Parameter Keep Alive sent as part of a CONNECT message.
If it is non-zero, it indicates the interval in seconds in which the client must send a message (if it does not send it, the MQTT broker can disconnect the client). If it has no data to send, it will at least send a PING request. A zero value means that the MQTT broker does not require periodic messaging.

The recommended Keep Alive value is several minutes.

The D2000 KOM process sends PING requests according to the settings of the Keep Alive and Ping Interval parameters (whichever interval expires first).

If the MQTT server did not send any message during the specified time interval, the D2000 KOM process sends a PING request and waits for a PING response (until time Reply Timeout).

A value of 0 turns off sending the PING request messages. The parameter allows detection of TCP connection failure.

The setting of message parsing:

• Text only - the message is not parsed, it is assigned to the I/O tag with the address IN_TOPIC
• JSON - the message is parsed as JSON data. If there is an I/O tag with the address IN_TOPIC, the whole message will be assigned to it.
  If there are I/O tags with addresses JA=json_address, they will be populated with the appropriate data from the JSON message. If no such addresses exist in the message, the I/O tags will be invalidated.
• Sparkplug Host - the message is parsed as Sparkplug B payload (binary coded), The D2000 KOM process behaves as a Host Application (data consumer).
• Sparkplug Edge Node- the message is parsed as Sparkplug B payload (binary coded), The D2000 KOM process behaves as an Edge Node (data producer).

Text only

JSON

Sparkplug Host

Sparkplug Edge Node

• UNIX - the numeric value represents the number of seconds from epoch 00:00:00 01.01.1970 UTC.
• UNIXMS - the numeric value represents the number of milliseconds from epoch 00:00:00.000 01.01.1970 UTC.

Note: Whether the time is interpreted as local or UTC with a configured offset depends on the time station parameters settings.

If the MQTT server does not respond to the SUBSCRIBE, UNSUBSCRIBE, and PING requests within the required time or the D2000 KOM process fails to read a complete message (and only part of it is read), the D2000 KOM process declares an error, closes the connection, and opens it again. Value 0 turns off the timeout.
The parameter enables the handling of problematic behavior of the MQTT server.

A timeout of a single reading from a TCP connection. D2000 KOM repeats reading of spontaneous data Max. Wait Retry times and if no data is read, the reading is timeouted and finished (and may be followed by a further reading or writing). By lowering Wait Timeout and Max. Wait Retry parameters, it is possible to achieve a faster writing response of the D2000 KOM process at the expense of a higher CPU load when the MQTT server has no data.
Note: if a lot of messages come from the MQTT server and the D2000 KOM also needs to write values, we recommend setting a lower parameter value (e.g. 0.005 sec) so that writing is not blocked by reading (in any case, after 10 received messages, there is an interruption during which the accumulated writes can be performed).

Payload field encoding. The MQTT protocol does not specify the content of the Payload field; the ISO-8859-1 standard encoding is suitable for both text and binary content, and UTF-8 is suitable if UTF-8 encoded texts are transmitted. Currently supported encodings are:

• UTF-8

• ISO-8859-1

• Windows-1250

-

Number of Topic Aliases that can be sent by the MQTT broker (parameter is sent in the CONNECT message). The value 0 (default) means that Topic Aliases are not used. The parameter is used only for MQTT Version = MQTT 5.0.

Topic Aliases are numeric identifiers from the interval <1 ... Topic Alias ​​Maximum>, which are sent by the D2000 KOM and the MQTT Broker instead of text identifiers.

Note: In the CONNACK message (response to CONNECT), the MQTT broker optionally sends its value "Topic Alias ​​Maximum", which has the meaning of the number of Topic Aliases that can be sent by the D2000 KOM process (e.g. test.mosquitto.org sends the value 10). D2000 KOM uses Topic Aliases only if the value of its configuration parameter Topic Alias ​​Maximum is non-zero.
Note: Topic Aliases are sent by the D2000 KOM process only when writing values ​​(NDATA/DDATA/NCMD/DCMD), it does not use them for NBIRT/DBIRTH/NDEATH/DDEATH messages (within MQTT Sparkplug) due to alias saving.

The parameter activates listings of unknown metrics and topics that do not have their own stations, but end up at a station with the address ".*" (if there is any). The listings will be in the line log as error messages even if the debug on the line is turned off (to facilitate the addition of I/O tags).

Activation of parsing of non-standard SparkPlug topics (if Payload Type = Sparkplug Host).
The standard topic has the form 'namespace/group_id/message_type/edge_node_id/[device_id]' e.g. 'spBv1.0/Owner/DDATA/devicebase1/local_inputs'.
A non-standard topic has several levels instead of group_id, e.g., 'spBv1.0/Basel/Packaging/DDATA/Falcon11/plannedProductionQuantity'.
Note: writing to I/O tags on stations with a non-standard SparkPlug topic is not supported.

Within the metric, it is possible to define a property called Quality of type Int32. According to the Sparkplug standard, it must be one of the values ​​0=BAD, 192=GOOD, 500=STALE. Any other D2000 Kom process reports as an error. The Ignore Unknown Quality parameter can be used to suppress this error message.

The parameter specifies which addresses of the Dataset metric are displayed when browsing:

• Columns only: one address is displayed for each dataset column (e.g. Dataset[*]^ColA)
• Items only: the address of each item is displayed (e.g. Dataset[1]^ColA, Dataset[2]^ColA, ...)
• Columns + Items: addresses for both columns and items are displayed

Separator of individual levels in Templates used when entering the address of the I/O tag in Sparkplug mode.
The default value is "->", so the address of the I/O tag can be e.g. SA=Template1->SubTemplate2->Item. If the template name contains the pair "->", it is possible to replace this separator (e.g. "=>").

If Payload Type = Sparkplug Host:

Identifier of Host Application (D2000 is a Host Application). 
If specified, the D2000 KOM process will send a STATE message (with online=true) according to the MQTT Sparkplug standard after connecting to the MQTT server. This message announces that it is alive (equivalent to the NBIRTH and DBIRTH messages sent by Edge Nodes and Devices). At the same time, it sets the Will Topic/Will Message in the CONNECT message according to the Sparkplug standard (STATE message with online=false), with Will QoS=QoS_1, Will Retain=YES, Clean Session Flag=YES.
If the identifier is not specified, the D2000 KOM does not send the STATE message (and the Will parameters are configurable).
According to the MQTT Sparkplug standard, the identifier is mandatory for the Sparkplug Host mode (without specifying it, the D2000 KOM process pretends to be an "anonymous" client and does not send a STATE message)..

If Payload Type = Sparkplug Edge Node:

Identifier of Primary Host Application (Identifier of the Host Application that is defined as primary - the most important - for this Edge Node).
If the identifier is specified, the D2000 KOM process waits for a STATE message after connecting to the MQTT server. When it is received and contains online=true, it sends an NBIRTH message with definitions and values ​​of all output I/O tags of the station corresponding to the Edge Node and optionally DBIRTH messages for stations corresponding to Device/Sensor devices (if such stations are configured). If the STATE message contains online=false (i.e., the Primary Host Application is not connected to the MQTT server), the D2000 KOM disconnects from the MQTT server and tries another MQTT server according to the communication link configuration, provided that multiple MQTT servers are configured. If a single MQTT server is configured, the D2000 KOM waits to receive a STATE message with online=true from the Primary Host Application and only then starts sending NBIRTH/DBIRTH/NDATA/DDATA messages.
If the identifier is not specified, the D2000 KOM process sends NBIRTH and possibly DBIRTH messages immediately after connecting to the MQTT server (and subsequently sends NDATA/DDATA messages with value changes).
According to the MQTT Sparkplug standard, the specification of the Primary Host Application is optional for the Sparkplug Edge Node mode.
Note: Regardless of the value of the Sparkplug Host ID parameter, the D2000 KOM sets the Will Topic/Will Message in the CONNECT message according to the Sparkplug standard, with Will QoS=QoS_1, Will Retain=NO, Clean Session Flag=YES.

If Payload Type = Sparkplug Edge Node, the parameter enables Store & Forward functionality for all stations on the line. If communication is down, values ​​are stored in memory and sent with the is_historical flag when communication is restored.
Note: this setting can be changed for individual stations using the Store & Forward station parameter.

Adding a textual representation of the value type (e.g. Int32) and a timestamp to the text I/O tag with IN_SP2JS address used to convert the Sparkplug payload to JSON for optional parsing in an ESL script.

TxtI, TxtO, Qi,
Ci, Co,
Ai, Ao,
Di, Do,
TiR, ToR

If Payload Type=JSON, the message is parsed as JSON data. The json_address value specifies the name of the JSON field whose value is to be assigned to the I/O tag. 
For JSON messages that can be structured, the syntax level1.level2.level3 ... is supported, e.g. rx.current, and if they contain fields (indexed from 1) also syntax level1[index1].level2[index2].level3 ... is possible, e.g. rx.gwrx[1].time.
Since the JSON message itself can be an array, the address can also start with an index, e.g. JA=[1].batt_cell_v_avg

For other examples, see the description of the LoRaWAN protocol's Envelope type I/O tags.

TxtI, TxtO, Qi,
Ci, Co,
Ai, Ao,
Di, Do,
TiR, ToR, TiA, ToA

If Payload Type = Sparkplug Host/Edge Node, the message is parsed as Sparkplug data (a binary format built on Google Protocol Buffers). Sparkplug data contains metrics that have text identifiers (sparkplug_address) or possibly numeric aliases (alias).

Reading template items  is possible by specifying sparkplug_address in the format <TemplateName1><Separator><TemplateName2><Separator> ... <Separator><ItemName> where:

• <TemplateNameX> is the name of the template/sub-template
• <Separator> is the separator for individual levels (standard characters "->", which can be changed with the Item Separator parameter if this sequence occurs in template/item names)
• <ItemName> is the item name of the deepest nested template

Examples of template item addresses:
SA=Template1->SubTemplate2->Item
SA=secUDT->sec

Reading dataset items (equivalent to structured variables in D2000) is possible by specifying sparkplug_address in the format <DatasetName>[<Row>]^<ColumnName> where:

• <DatasetName> is the name of the dataset (it can also be part of a structure, e.g. Template1->SubTemplate2->Dataset3)
• <Row> is the column number (1..N) or the "*" character. In the latter case, it is possible to configure the Destination column to which all rows are written (values ​​from the first row of the corresponding column are written to the I/O tag)
• <ColumnName> is the name of the dataset column

Examples of dataset item addresses:
SA=Performance[3]^ActivePower
SA=Machine2->Parameters[1]^ActivePower
SA=DHS/Formation Data->Reservoir Parameter[*]^Gas density

For output I/O tags, the value type must be specified (ST=type). Simple types are supported (not template items/dataset items):

• Int8
• Int16
• Int32
• Int64
• UInt8
• UInt16
• UInt32
• UInt64
• Float
• Double
• Boolean
• String
• DateTime
• Text

Note: There is no difference between the String and Text types.
Note: for Payload Type = Edge Node, the D2000 KOM process sends DBIRTH/NBIRTH messages that contain only I/O tags (both input and output ones) with a defined value type.

For Payload Type = Edge Node, it is also possible to specify a numerical alias (SAL=alias) as a natural number (0, 1, 2, ...) for both input and output I/O tags. The alias of the I/O tag within the station must be unique. Aliasing allows you to reduce the size of the transmitted data: in the NBIRTH/DBIRTH message, both the text identifier (sparkplug_address) and the alias are specified for each I/O tag; in the NDATA/DDATA messages, only aliases that are shorter than the text addresses are sent. Alias ​​is only used if Sparkplug Alias Mode = "Default (SAL=alias)".
Note: If aliases are used in the I/O tag address in Payload Type = Sparkplug Host mode, they will be replaced by aliases from the NBIRTH/DBIRTH message. However, if for some reason the Edge Node does not send these messages (and sends NDATA/DDATA with aliases), aliases can be useful for matching a text name with an alias.

The PUBLISH message created during writing contains a Topic derived from the station address. The message type depends on the station address:

• for Payload Type = Sparkplug Host, depending on whether it is an Edge Node (NCMD) or Device/Sensor (DCMD)
• for Payload Type = Edge Node, depending on whether it is an Edge Node (NDATA) or Device/Sensor (DDATA)

The Payload of the message contains a timestamp, a value type (type), a written value (encoded according to the specified value type), and a metric name (sparkplug_address) or an alias.

TxtI

The I/O tag is used to convert the Sparkplug payload into a JSON representation, which can then be processed, e.g., in an ESL script. Depending on the Convert Datatype/Timestamp to Text parameter, a textual representation of the value type and timestamp is also added. 
An example of value:

{"metrics":[{"datatype":3,"int_value":7338992,"name":"Corrected Vol Acc Stn","timestamp":1729664005479}],"seq":32,"timestamp":1729664005479}

After formatting into a readable form:

{
    "metrics": [
        {
            "datatype": 3,
            "int_value": 7338992,
            "name": "Corrected Vol Acc Stn",
            "timestamp": 1729664005479
        }
    ],
    "seq": 32,
    "timestamp": 1729664005479
}

An example of a more complex value containing properties and a dataset, and also displaying a textual representation of the data type (datatype_txt) and timestamp (timestamp_txt) as a result of the set parameter Convert Datatype/Timestamp to Text:

{
    "metrics": [
        {
            "datatype": 12,
            "datatype_txt": "String",
            "name": "Node Properties/Configuration",
            "string_value": "{}",
            "timestamp": 1730305529539,
            "timestamp_txt": "30-10-2024 17:25:29.539"
        },
        {
            "alias": 30064771073,
            "datatype": 5,
            "datatype_txt": "Uint8",
            "int_value": 0,
            "name": "Node Properties/Missing Param",
            "properties": {
                "keys": [
                    "usage"
                ],
                "values": [
                    {
                        "string_value": "technical information",
                        "type": 12,
                        "type_txt": "String"
                    }
                ]
            },
            "timestamp": 1730305529537,
            "timestamp_txt": "30-10-2024 17:25:29.537"
        },
        {
            "alias": 0,
            "dataset_value": {
                "columns": [
                    "topic_name",
                    "offset",
                    "length",
                    "crc"
                ],
                "num_of_columns": 4,
                "rows": [
                    {
                        "row": [
                            "N/A",
                            0,
                            0,
                            0
                        ]
                    }
                ],
                "types": [
                    12,
                    7,
                    7,
                    7
                ],
                "types_txt": [
                    "String",
                    "UInt32",
                    "UInt32",
                    "UInt32"
                ]
            },
            "datatype": 16,
            "datatype_txt": "DataSet",
            "name": "Node Control/FW Update",
            "timestamp": 1730305529537,
            "timestamp_txt": "30-10-2024 17:25:29.537"
        }
    ],
    "seq": 0,
    "timestamp": 1730305529536,
    "timestamp_txt": "30-10-2024 17:25:29.536"
}