MQTT Client Protocol (Message Queue Telemetry Transport)

Supported device types and versions
About the MQTT Sparkplug protocol
Communication line configuration
Communication line parameters
Communication station configuration
I/O tag configuration
Literature
Document revisions

Supported device types and versions

The protocol is an implementation of the MQTT 3.1.1 standard (October 2014). MQTT protocol is a client/server protocol of a subscribe/publish type. It is simple, has little overhead, and is easy to implement. It is used for M2M communication (Machine to Machine) and in the IoT context (Internet of Things). The MQTT server is also called the MQTT broker.
D2000 KOM implements the client part of the protocol. The protocol is implemented on a TCP/IP line.
For the transfer of LoRaWAN data encapsulated within the MQTT protocol, see LoRaWan protocol description.

The Sparkplug 3.0.0 standard (Sparkplug B MQTT Payload) is also partially supported. Support includes:

reading simple value types
reading arrays (with support for the Destination column)
writing simple value types
reading of template items (UDT)
reading of dataset items

The communication was tested/deployed with:

www.TheThings.network cloud
Loriot.io cloud
Slovanet cloud (loralink.slovanet.sk)
Pixii PowerShaper (energy storage solution)
liveobjects.orange-business.com cloud
test.mosquitto.org (Sparkplug B)

Note: communication with the cloud liveobjects.orange-business.com via WebSockets (wss://liveobjects.orange-business.com:443/mqtt) was also tested. The program https://github.com/jimparis/unwebsockify.git was used as a WSS wrapper. This program started with the following parameters:
./unwebsockify.py --port 1883 --listen 172.16.0.1 wss://liveobjects.orange-business.com:443/mqtt
The D2000 KOM process connected to address 172.16.0.1 on port 1883. The WSS wrapper connected to the defined URL and wrapped the MQTT communication data in a WebSocket envelope.

Each PUBLISH message contains a topic (Topic), data (Payload), and level of confirmation (QoS). PUBLISH messages can be sent both by the client and the server. The clients at the beginning of the communication will use the SUBSCRIBE message to indicate what topics (parameter of Topic Filter protocol) they are interested in.
The protocol defines the following levels of confirmation of PUBLISH messages - QoS (Quality of Service):

QoS_0 - PUBLISH message is not confirmed, it may be lost
QoS_1 - PUBLISH message is confirmed by the other side's PUBACK, it may be duplicated
QoS_2 - PUBLISH message is confirmed by the other side's PUBREC which is then confirmed back by the PUBREL message and that one by a final PUBCOMP message.

The level of confirmation of the messages sent by the D2000 KOM process is defined by the protocol parameter Publish QoS. The D2000 KOM process considers the writing of the output tag to be successfully finished depending on the QoS:

QoS_0 - after the data is successfully sent via the TCP connection
QoS_1 - after receiving PUBACK
QoS_2 - after receiving PUBCOMP

The MQTT communication starts with the CONNECT message sent by the client (D2000 KOM). The message contains User Name, Password, and other parameters, from which only Clean Session Flag and Client ID can be modified (parameter Will Flag is not used, as well as Will QoS and Will Retain, parameter Keep Alive is set to 0). The server replies with a CONNACK message with a return code that contains information about the success of the connect operation.

Then the client sends a SUBSCRIBE message with a filter of topics (Topic Filter parameter), specifying which topics it is interested in, and with the required maximum level of confirmation (parameter Subscribe QoS).

The server responds with a return code that contains information about the success and maximum QoS that was assigned to the requested topics.

Then follows a phase of communication, during which both the client and the server send PUBLISH messages (the client with any topic, the server with topics relating to the filter of topics of the received SUBSCRIBE message) and confirm them according to the value of the QoS parameter of the received PUBLISH messages.

If the server does not send a message for longer than Ping Interval seconds, the client sends the PING request message, to which the server must respond with the PING response message (within the time specified by the Reply Timeout parameter).
If parameters change on the line, the connection is closed and re-created.

About the MQTT Sparkplug protocol

Sparkplug's MQTT protocol defines 3 groups of applications:

Edge Node - supports the MQTT protocol and connects to the MQTT server. It sends him data obtained from the Device or its own or aggregated data
Device/Sensor - represents a physical or logical device connected to an Edge Node and providing data, process data, or metrics
Host Application - represents a data consumer (SCADA/MES system, Historian, analysis tool) that connects to the MQTT server and receives MQTT data from the Edge Node/Device and, if necessary, commands. D2000 is a Host Application.

MQTT Sparkplug defines the following message types:

NBIRTH – Birth certificate for Sparkplug Edge Nodes (information about the connection of the Edge Node to the MQTT Server)
NDEATH – Death certificate for Sparkplug Edge Nodes (information about the loss of the connection between the MQTT Server and the Edge Node)
DBIRTH – Birth certificate for Devices (information about Device connection to Edge Node)
DDEATH – Death certificate for Devices (information about the loss of Edge Node connection with Device)
NDATA – Edge Node data message (data from Edge Node)
DDATA – Device data message (data from Device)
NCMD – Edge Node command message (Host Application command for Edge Node)
DCMD – Device command message (Host Application command for Device)
STATE – Sparkplug Host Application state message (information from the state of the Host Application for the Edge Node/Device [online/offline]).

MQTT Sparkplug defines a Topic syntax for all message types, which has the form namespace/group_id/message_type/edge_node_id/[device_id], where:

namespace is a spBv1.0 constant (for Sparkplug B)
group_id is an arbitrary name of a logical group (e.g. by device type, operation, organizational structure, etc.)
message_type is the message type
edge_node_id is the identifier of the Edge Node
device_id is the identifier of the Device (only if the message is from/for a Device)

Examples:

spBv1.0/Sparkplug Devices/DDATA/Marvin/Sensor1
pBv1.0/MIMIC/DDATA/20:19:AB:F4:0E:0D/sensor11
spBv1.0/SparkplugDevices/NBIRTH/JsonScada

Note: STATE messages have a Topic of the form spBv1.0/STATE/sparkplug_host_id, where sparkplug_host_id is the unique name of the host application.

Examples:

spBv1.0/STATE/CFGHUB2023
spBv1.0/STATE/D2Kom

All message types except STATE have a binary Sparkplug payload. The STATE messages have a JSON payload (contains only the items "online" (True/False value) and "timestamp" (time stamp in milliseconds).

The Sparkplug payload contains metrics - each required metric can be configured into a single I/O tag. The metric contains a text identifier (specified in the I/O tag configuration as sparkplug_address), value type, value (simple/array/structured), optionally timestamp, and other attributes. D2000 supports working with simple values and array values - the first field item is stored in the I/O tag; it is possible to configure the Destination Column for the I/O tag, in which all array items are stored (or the first N, if the structure does not have enough rows).
Note: if a metric arrives with the is_null flag set, the I/O tag will have the Invalid value.

Communication line configuration

Communication line category: TCP/IP-TCP.
Host: IP address of MQTT server (or redundant addresses separated by a comma or semicolon).
Port: the default port number is 1883 or 8883 for the encrypted SSL/TLS variant.
Line number: unused, set the value to 0.

Note: The default port for the MQTT protocol is 1883 or 8883 for the SSL/TLS version. D2000 KOM does not contain an implementation of the SSL/TLS protocol variant, but it is possible to configure it by using the stunnel utility http://www.stunnel.org working in a client mode (client = yes). Stunnel running on the same computer as the D2000 KOM should listen to the 1883 local port and after connecting of D2000 KOM process to the port should encrypt the communication using SLL/TLS and send it to the target MQTT server (typically on port 8883).

Note: when using an SSL/TLS connection, you must pay attention to the expiration of the certificates used. The expiration date can be found on Linux with the command

openssl x509 -text -in file.crt | grep "After"

Forced disconnection: If all stations on the line are in the simulation mode or the communication is stopped for them, the line will be disconnected (the communication socket will be closed). If the simulation is disabled for at least one station and the communication is not stopped for it (the Parameters tab of the Station type object), the line will be connected again.

Communication line parameters

Dialog link configuration - Protocol parameters tab.
They affect some optional protocol parameters. The following protocol line parameters can be entered:

Table 1

Parameter	Description	Unit / size	Default value
Full Debug	Activates detailed debug information about sending and receiving values.	YES/NO	NO
User Name	User name used in a CONNECT message to connect to the MQTT server.	-
Password	Password used in a CONNECT message to connect to the MQTT server.	-
Topic Filter	The name of one topic or a multiple-topic filter sent within the SUBSCRIBE message. Using the filter the MQTT client specifies topics, within which it wants to receive messages. Note: topics are hierarchically sorted, a slash (/) is used as the separator, a plus (+) is used as a one-level mask, a hash (#) character is used as a mask for multiple levels. Examples of filter: a/b , level1/+ , # , +/+/+/up Note: the change of the Topic Filter parameter will be reflected after restarting the communication - e.g. due to the breakdown of the TCP connection, as long as all stations on the line are switched off (StOff) and switched on again, or after a restart of the KOM process. In the first two cases, the message UNSUBSCRIBE is sent to the original Topic Filter and then SUBSCRIBE to the new Topic Filter (this can be important in so-called persistent sessions when the Client ID parameter is specified and the MQTT server remembers the state of the client even after the TCP connection is broken). Note: for Payload Type=Sparkplug, the filter spBv1.0/# is sufficient to receive all Sparkplug messages.	-	#
Subscribe QoS	The desired maximum level of validation (QoS) sent within the SUBSCRIBE message. The MQTT server can then send PUBLISH messages with such or lower levels of confirmation (but not higher). PUBLISH messages sent by the MQTT server will be confirmed by the D2000 KOM process according to the level of confirmation specified in them. The higher the level of confirmation, the more messages between the client and the server are exchanged (1 at QoS_0, 2 at QoS_1, and 4 at QoS_2).	QoS_0 QoS_1 QoS_2	QoS_1
Client ID	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.	-	D2000kom
Clean Session Flag	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 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).	YES/NO	NO
Publish QoS	Level of confirmation (QoS) used to send PUBLISH messages through the D2000 KOM process. Sending the PUBLISH message is the outcome of writing into the output tag with the OUT_VALUE address. The higher the confirmation level, the more messages between the client and server are exchanged (1 for QoS_0, 2 for QoS_1, and 4 for QoS_2).	QoS_0 QoS_1 QoS_2	QoS_0
Publish Retain	Setting the Retain flag used when sending PUBLISH messages by the D2000 KOM process. Activating the Retain flag causes the last message sent by the D2000 KOM process to be available on the MQTT server to other clients immediately after they are connected, as well as after the D2000 KOM process is disconnected.	YES/NO	NO
Keep Alive	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).	0-65535 sec	0
Ping Interval	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.	0-3600 sec	60
Payload Type	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 - the message is parsed as Sparkplug B payload (binary coded).	Text only JSON Sparkplug	Text only
Time Field Name	If Payload Type=JSON, the name of the field with a timestamp. If the field name is not specified or the field is not found, the current time is assigned to the values. For more information on the field name format, see I/O tags with addresses JA=json_address.	-	-
Time Mask	Mask for parsing a value in the field with a timestamp. Special masks are: 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.	-	yyyy-mm-dd hh:mi:ss.mss
Ignore Missing Time	Ignoring a missing timestamp - if it is not present in the JSON payload, no warning will be issued.	YES/NO	NO
Will Flag	Parameter Will Flag of a CONNECT message. A value of Yes means that the server will send a Last Will message to interested parties if the connection to the D2000 KOM process is lost. Note: If Payload Type=Sparkplug and the Sparkplug Host ID parameter is not empty, this parameter is ignored and Last Will will be sent (see the description of the Sparkplug Host ID parameter).	YES/NO	NO
Will QoS	The acknowledgment level (QoS) used when sending a Last Will message in the event of a loss of connection to the D2000 KOM process. Note: If Payload Type=Sparkplug and the Sparkplug Host ID parameter is not empty, this parameter is ignored and QoS_1 level will be sent (see the description of the Sparkplug Host ID parameter).	QoS_0 QoS_1 QoS_2	QoS_0
Will Retain	The setting of the Retain flag used when sending a Last Will message if the connection to the D2000 KOM process is lost. Note: If Payload Type=Sparkplug and the Sparkplug Host ID parameter is not empty, this parameter is ignored and Retain is set to YES (see the description of the Sparkplug Host ID parameter).	YES/NO	NO
Will Topic	The topic used to send the Last Will message if the connection to the D2000 KOM process is lost. Note: If Payload Type=Sparkplug and the Sparkplug Host ID parameter is not empty, this parameter is ignored (see the description of the Sparkplug Host ID parameter).	-
Will Message	Contents of the Last Will report if the connection to the D2000 KOM process is lost. Note: If Payload Type=Sparkplug and the Sparkplug Host ID parameter is not empty, this parameter is ignored (see the description of the Sparkplug Host ID parameter).	-
Reply Timeout	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.	sec	20
Wait Timeout	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).	sec	0.100
Max. Wait Retry	The number of repetitions of reading from TCP connection. See the description of the Wait Timeout parameter.	-	3
Payload Encoding	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	-	ISO-8859-1
Sparkplug parameters
Report Unknown Metrics	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).	YES/NO	NO
Nonstandard Topic	Activation of parsing of non-standard SparkPlug topics. 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.	YES/NO	NO
Ignore Unknown Quality	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.	YES/NO	NO
Item Separator	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. "=>").		->
Sparkplug Host ID	Identifier of Host Application. If specified, the D2000 KOM process will send a STATE message according to the MQTT Sparkplug standard after connecting to the MQTT server. With this message it 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, 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).	-	-

Communication station configuration

Communication protocol "MQTT Client Protocol".
Station address: the station address corresponds to the Topic field in the PUBLISH message received from the MQTT server. The address can be a specific Topic, a regular expression, a # character representing all Topics, or a topic .* representing all Topics that are not suitable for other stations. The processing priority is as follows:
- If there is a station with address # on the line, all messages are directed to its I/O tags and no further search is performed.
- Otherwise, all other stations on the line are searched (except the .* address). If the Topic matches the address of a station, the message is directed to that station and no further search is performed.
- Otherwise, all other stations on the line are searched (except the .* address), and their address is evaluated as a regular expression. If the Topic matches the station address, the message is directed to that station and no further search is performed.
  Stations are searched in descending order (by station address), so more specific terms go first (e.g., status/battery before status/batt.*)
- Finally, if there is a station with a .* address, the message is addressed to it.
Polling parameters on the Time parameters tab - recommended value is Delay=0.

Note: for a SparkPlug MQTT server, Topic has the form 'namespace/group_id/message_type/edge_node_id/[device_id]', where message_type indicates the message type (e.g. DDATA, DBIRTH, DDEATH).
A regular expression (e.g. spBv1.0/Sparkplug Devices/.*/MyDevice/Sensor2) can be used instead of message_type to cover all message types.
If Payload Type=Sparkplug, it is possible to omit the namespace and message_type parts and write the Topic in the abbreviated form 'group_id/edge_node_id/[device_id]' (e.g. Sparkplug Devices/MyDevice/Sensor2).

Note: If the station address is in abbreviated form, commands (DCMD, NCMD) are not processed for it. If it is in the form of a regular expression ('namespace/group_id/.*/edge_node_id/[device_id]'), the station also processes commands (including the command sent by the D2000 KOM process if Send Node Control/Rebirth=YES). Therefore, we recommend specifying the station address in abbreviated form. If it is also necessary to process commands (from other Host Applications), then create another station with an address in the form of a regular expression (e.g. spBv1.0/Sparkplug Devices/DCMD/MyDevice).

The Browse button opens a browse dialog for the station address. As long as the communication is functional, a dialogue containing the Topics received so far will be displayed. The list of received topics can be cleared with the Refresh button.
Double-clicking on a particular line will insert the value from the Address column into the configuration of the station from which the browse dialog was opened.
Note: The Station column shows the station to which the Topic has been assigned (based on the abovementioned processing priorities). For Sparkplug addresses, the abbreviated form of the address is displayed.

Station protocol parameters

Communication station - configuration dialog box - "Protocol parameters" tab.
They influence some of the optional parameters of the protocol.

Table 2

Keyword	Full name	Meaning	Unit	Default value
SWT	Station Will Topic	Will topic of the device. If this parameter is set and a message with the same topic is received, the station will go into a communication error (StHardErr) and the values of the I/O tags will be invalidated. In this way, it is possible to emulate the standard behavior that occurs when there is a communication error with the device (even if the communication between the D2000 Kom process and the MQTT broker is functional).
SWP	Station Will Payload	Content of the Will message. If this parameter is set and a message with the same topic as defined by the Station Will Topic parameter is received, the Payload must also be the same. If this parameter is an empty string, matching the topic with the Station Will Topic parameter is sufficient. Note: this parameter was implemented due to MQTT brokers sending messages with the same Topic when connecting/disconnecting the device, the difference being only in the Payload.
FPT	Payload Type	The setting of message parsing (overriding the line parameter Payload Type): Default - the line parameter Payload Type is respected 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 - the message is parsed as Sparkplug B payload (binary coded).	Default Text only JSON Sparkplug	Default
FTF	Time Field Name	If Payload Type=JSON, the name of the field with a timestamp - overriding the line parameter Time Field Name.	-	-
FTM	Time Mask	Mask for parsing a value in the field with a timestamp - overriding the line parameter Time Mask. Note: Whether the time is interpreted as local or UTC with a configured offset depends on the time station parameters settings.	-	-
Sparkplug parameters
SR	Send Node Control/Rebirth	At the start of the D2000 KOM process, a command (NCMD or DCMD) with the metric 'Node Control/Rebirth' is sent to the SparkPlug station. The response should be a message (NBIRTH/DBIRTH) with all current metrics.	YES/NO	YES

I/O tag configuration

Possible value types of I/O tags: Ci, Co, TxtI, TxtO, Qi, Ci, Co, Ai, Ao, Di, Do, TiR, ToR, TiA, ToA.

The MQTT implementation supports three work modes:

Text mode: The original implementation of the MQTT protocol contained only input text I/O tags with the addresses IN_TOPIC, IN_DATA, and optionally a pair of I/O tags with the addresses IN_ID and ACK_ID. The first two I/O tags were used to publish the received Topic and Payload (which then needed to be parsed in the script), the second two points were used to publish the packet identifier and confirm the processing of the packet. Thus, it was possible to ensure that for data sent with QoS > QoS_0, confirmation was sent only after data processing in the script.
Output I/O tags with the addresses OUT_TOPIC and OUT_VALUE are used for writing.
JSON mode: An extension for processing Payload with JSON data (Payload Type=JSON) was implemented with the help of input I/O tags with addresses JA=json_address. The D2000 KOM process directly parses the JSON payload and sets the values of I/O tagswith JSON addresses. I/O tags with addresses IN_TOPIC, IN_DATA, IN_ID, and ACK_ID may not exist at all.
Output I/O tags with the addresses OUT_TOPIC and OUT_VALUE are used for writing.
Sparkplug mode: An extension for Payload processing with Sparkplug data (Payload Type=Sparkplug) has been implemented using input I/O tags with addresses SA=sparkplug_address. The D2000 KOM process directly parses the Sparkplug payload and sets the values of I/O tags with the Sparkplug addresses. I/O tags with addresses IN_TOPIC, IN_DATA, IN_ID, and ACK_ID may not exist at all.
Writing simple values is possible using output I/O tags with addresses ST=type;SA=sparkplug_address, where type is the Sparkplug definition of a data type (eg Int8, UInt16, DateTime, String, etc). The output I/O tags must be on the station with the Sparkplug address since the Topic is derived from it when writing.
Note: STATE type messages that have a JSON payload can be parsed using I/O tags with a JSON address (typically an I/O tag of Di type with an address of JA=online).

Type of I/O tag	Address	Description
I/O tags for reading data sent by the MQTT server through a PUBLISH message. Note: values of I/O tags are set by the D2000 KOM process in the order IN_TOPIC, IN_DATA, and IN_ID. It is not necessary for the configuration to contain all three I/O tags.
TxtI	IN_TOPIC	Topic (Topic) of received PUBLISH message.
TxtI	IN_DATA	Data (Payload) of received PUBLISH message.
Ci	IN_ID	Identifier of a packet (Packet Identifier) of PUBLISH message that depends on the level of validation (QoS). For messages sent with QoS_0, the identifier is zero, for QoS_1 and QoS_2, it is a positive 16-bit number. Note: if the MQTT server sends also messages with the QoS_0 level of validation and the ACK_ID I/O tag is configured, then we recommend activating the option New value when changing time in the Filter tab, so that repeated writing of the value 0 will cause a new value that differs only in a timestamp to be generated.
I/O tag to confirm the received data to the MQTT server.
Co	ACK_ID	If an output I/O tag with ACK_ID address is defined, the D2000 KOM expects confirmation of the processing of each message by writing a copy of the value of the IN_ID tag. Only after, it sets values from the next received PUBLISH message (if it was received in the meantime) into the IN_TOPIC, IN_DATA, and IN_ID I/O tags (in this order). In the case of the QoS_0 level of confirmation, it is, therefore, necessary to repeatedly set the value of the I/O tag ACK_ID to 0. If the I/O tag ACK_ID does not exist, the values are written into the IN_TOPIC, IN_DATA, and IN_ID I/O tags immediately after the PUBLISH message is received and processed. Note: for the messages received with the QoS_0 level of validation, no confirmation is sent to the MQTT server, only the values of the received PUBLISH message will be published.
I/O tags for sending values to the MQTT server through a PUBLISH message. Note: in order for the D2000 KOM process to send the PUBLISH messages to the MQTT server, both I/O tags must be defined within one station.
TxtO	OUT_TOPIC	The topic of the PUBLISH message being sent.
TxtO	OUT_VALUE	Data (Payload) of the PUBLISH message being sent. Note: sending the message is performed out as a result of writing to the OUT_VALUE I/O tag (i.e. if the Topic does not change then it is sufficient to set the OUT_TOPIC point once - e.g. by using default value).
I/O tags for parsing JSON messages
TxtI, TxtO, Qi, Ci, Co, Ai, Ao, Di, Do, TiR, ToR	JA=json_address	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.
I/O tags for parsing Sparkplug messages
TxtI, TxtO, Qi, Ci, Co, Ai, Ao, Di, Do, TiR, ToR, TiA, ToA	Input I/O tags: SA=sparkplug_address Output I/O tags: ST=type;SA=sparkplug_address	If Payload Type=Sparkplug, 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). 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. Simple types are supported (not template items/dataset items): Int8 Int16 Int32 Int64 UInt8 UInt16 UInt32 UInt64 Float Double Boolean String DateTime Text The PUBLISH message created during writing contains a Topic* derived from the station address. The message type depends on the station address - whether it is Edge Node (NCMD) or Device/Sensor (DCMD). The Payload contains a timestamp, a value type (type), a written value (encoded according to the specified value type), and a metric name (sparkplug_address).

Note: it is also possible to monitor the status of other Sparkplug Host Applications connected to the MQTT server. If the Identifier of Host Application is e.g. "ACME", so it is necessary to create a station with the address "spBv1.0/STATE/ACME" (or in abbreviated form "ACME") and on it an I/O tag of type Di with the address "JA=online" (since the Host Application sends a STATE message with a JSON payload).

Browse

In the Sparkplug mode, for the I/O tags, it is possible to discover the list of metrics, as long as the KOM process is running and communication with an MQTT server is established.
Clicking the Browse button opens the MQTT Item Browser window and displays a list of received metrics. The object list is created dynamically as a result of received messages.

The list of objects is dynamic, i.e. when a new value arrives in the KOM process, it is updated. Filtering in individual columns (with the exception of Time column) is also supported, asterisks can be used in the mask (e.g. *Short*).

Double-clicking on a particular line will insert the value from the Address column into the configuration of the I/O tag from which the MQTT Item Browser window was opened.

The Refresh button clears the list of values in both the CNF and the KOM process and optionally sends the Rebirth command (if Send Node Control/Rebirth=YES).

The Value column contains the received value.

Literature

Links
Official website of MQTT protocol http://mqtt.org

Official website of MQTT Sparkplug protocol https://sparkplug.eclipse.org

Specifications and Standards
MQTT 3.1.1 specification http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html
ISO/IEC 20922:2016 http://www.iso.org/standard/69499.html

Descriptions of Data Formats and API
www.loriot.io - Application API Data Format https://www.loriot.io/home/documentation.html#docu/app-data-format
www.thethingsnetwork.org - API Reference https://www.thethingsnetwork.org/docs/applications/mqtt/api.html

An example of communication

The attached ZIP contains the configuration of two lines and two stations with the MQTT protocol and the definition of the S.MqttTest scheme. Data written over one line is received over the other line. The functionality was verified against the Mosquitto broker installed locally on Windows, listening on TCP port 1883.

Mosquito_test.zip