MQTT

The MQTT protocol is a lightweight communication method designed for Internet of Things (IoT) applications, enabling efficient data exchange in environments with limited bandwidth or high network usage.

An MQTT broker acts as the intermediary between these environments, facilitating message exchange between publishers and subscribers using the MQTT protocol. This setup is especially suited for distributed systems and lightweight messaging scenarios common in IoT environments.

The MQTT Agents allow your Data Stream to send messages to and receive messages from an external MQTT broker using the MQTT protocol.

The MQTT agents are useful in several scenarios:

• Real Time Updates You could send and receive real time data across environments that are connected to the same MQTT Broker.

• Pass Data to another Data Stream You could have one Data Stream that performs pre-processing and then passes it off to several other Data Streams for further processing by connecting them to a common MQTT Broker and assigning a unique Topic for each Data Stream.

Fig 1: Data Stream in a Stream Example

• Parallel Processing and Synchronization You can pass data to multiple Data Streams to do parallel processing and synchronize the results back into one MQTT Listener. This ensures that the result for both Data Streams are received first before doing additional processing. See Message Correlation in Advanced Settings and read more about the request-response pattern here.

Fig 2: Parallel Processing and Synchronization

• Live Monitoring You can monitor the live status of a Data Stream that contains MQTT Agents to check if it is ready to Send or Receive data by enabling the sending of Status Messages in Advanced Settings.

The Listener lets you receive records from sensors set up to send messages to an MQTT broker.

The Action Agent as Remote Agents, allows you to send records between different legs of a single Data Stream running on different Stream Host Collections.

For more information about the MQTT protocol, please click here.

Pre-requisites

The following is required to use this Agent:

• MQTT broker address and credentials (If broker requires authentication)

• Access to broker port e.g. 1883

• Minimum XMPro Data Stream Designer v4.0

When implementing Message Correlation, MQTT v5 is needed.

Current Version

• Download MQTT Listener v3.04

• Download MQTT Action Agent v3.04

Please contact XMPro if you're looking for an older version of this Agent.

Listener Release Notes

Version	Date	Description
v3.04	22 Aug 2024	Catch and log any exceptions encountered when attempting to connect to MQTT broker (Stream Object used to continue trying to reconnect even once the Data Stream was unpublished). Support message correlation to allow synchronization of multiple MQTT Action Agent outputs to one MQTT Listener (added 'Use Message Correlation', 'Correlation Types', and 'Max Queue' properties). Support the selection of the topic source to allow manual entry even when using variables for other properties (added 'Topic Source' property).
v3.02	25 Jul 2024	Support status topics to allow for monitoring on the broker of whether Agent is online (data stream is published) or offline (data stream is unpublished or there is an unexpected shutdown) Added 'Send Status Message'. Added 'Status Topic'.
v3.00	9 Jul 2024	Changed from using M2MQTTDotNetCore NuGet package to MQTTnet. Added support for MQTT Protocol v5.
v2.04	5 Sep 2023	Added Payload Format option 'Plain Text'. Added 'Include Topic in Payload' property - tick to append a Topic column to the output payload. Added 'Include Timestamp in Payload' property - tick to append a Timestamp column to the output payload.
v2.02	31 May 2023	Support Topic single-level (+) and multi-level (#) wildcards.
v2.01	20 Feb 2023	Added the Advanced Settings option. Added variable support.
v1.81	6 Sep 2022	Repackaged to include the Agent's Category.
v1.80	27 Sep 2019	Added option to return source message in the payload.
v1.70	8 May 2019	Added a checkbox to specify whether to use Clean Session or not. Handle re-connection after Broker connection drop - Agent tries to reconnect every 5 seconds until connection is established.
v1.63	4 Mar 2019	Updated help URL.
v1.62	18 Feb 2019	Payload definition - cannot focus out of a new row.
v1.61	22 Nov 2018	Added HEX as payload.
v1.51	7 Nov 2018	Added exception handling to receive method.
v1.50	15 Aug 2018	Added Checkbox option to specify JSON Path for nested payload.
v1.42	14 Aug 2018	Modified listener to handle messages with format as Json Object as well as Json Array.
v1.40	29 May 2018	Modified listener to handle messages with nested structure.
v1.30	17 May 2018	Added Validation.
v1.20	8 May 2018	Fixed unhandled exception that crashed the Stream Host.
v1.10	23 Aug 2024	Initial Release.

Action Agent Release Notes

Version	Date	Description
v3.04	22 Aug 2024	Catch and log any exceptions encountered when attempting to connect to MQTT broker (Stream Object used to continue trying to reconnect even once the Data Stream was unpublished). Fixed the automatic reconnect to broker when the MQTT client is disconnected. Support message correlation to allow synchronization of multiple MQTT Action Agent outputs to one MQTT Listener: added 'Include Correlation Data', 'Correlation Id, and 'Correlation Type'. Support the selection of the topic source to allow manual entry even when using variables for other properties (added 'Topic Source' property).
v3.02	25 Jul 2024	Support status topics to allow for monitoring on the broker of whether Agent is online (data stream is published) or offline (data stream is unpublished or there is an unexpected shutdown). Added 'Send Status Message'. Added 'Status Topic'.
v3.00	9 Jul 2024	Changed from using M2MQTTDotNetCore NuGet package to MQTTnet. Added support for MQTT Protocol v5.
v2.02	20 Feb 2023	Added an Advanced Settings option. Added variable support.
v1.63	6 Sep 2022	Repackaged to include the Agent's Category.
v1.62	7 May 2019	Added option to clear retained messages based on new drop-down option to specify an input column that contains a Boolean value.
v1.61	18 Apr 2019	Added a checkbox to choose whether to specify a static topic or dynamic topic and then provide topic name as a static value or select from drop-down respectively.
v1.60	8 Apr 2019	Changed Topic to a drop-down to allow for dynamic topic name. Added checkbox option to Retain message.
v1.41	4 Mar 2019	Updated Help URL.
v1.40	21 Aug 2018	Added ability to reconnect to broker if the connection drops while the data stream is published.
v1.30	17 May 2018	Added Validation.
v1.20	8 May 2018	Fixed unhandled exception that crashed the Stream Host.
v1.10	4 May 2018	Initial Release.