MQTT subscriber

Subscribes to an MQTT topic (or topic filter) and emits every received message as an EventData record. Use it as the starting point of a flow that reacts to MQTT messages.

Pins

Pin	Kind	Types	Description
Start	start		Starts the subscription. Called by the system at flow start when unconnected. Further calls are ignored.
Continue	continue	Viaducta.Plugins.Event.Types.EventData	Called once for every received message.

Configuration

Field	Type	Default	Description
Topic	string (max 255)		Required. Topic or topic filter to subscribe to. Wildcards + and # are passed to the broker (sensors/+/temperature).
ClientId	string		MQTT client id. Empty: a random id at every start. A fixed id must be unique on the broker: two clients with the same id disconnect each other.
Host	string		Required. Broker host name or IP for TCP. With WebsocketConnection the full WebSocket URI (for example ws://broker:8080/mqtt).
Port	integer (1-65535)	1883	Broker port for TCP. Ignored with WebsocketConnection.
Username	string (max 256)		User name. Credentials are sent when Username is set.
Password	password (max 256)		Password.
WebsocketConnection	boolean	false	true: connect over WebSocket, Host is the URI.
UseMqtt5	boolean	false	true: MQTT 5.0, false: MQTT 3.1.1.
UseRetainFlag	boolean	false	Not used by the subscriber.
QoS	dropdown: 0 QoS0 (At most once), 1 QoS1 (At least once), 2 QoS2 (Exactly once)	0	QoS of the subscription.

Behaviour

• The connection is opened at flow start. An empty Host or Topic, an invalid QoS or a refused or failed first connect stops the flow with an unrecoverable error. After that, a lost connection is reconnected in the background (after 1 s, doubling up to 30 s) and logged as warnings. The plugin keeps the flow running (KeepAlive) and disconnects when the flow stops.
• On Start the plugin subscribes to Topic with QoS; a refused subscription is an unrecoverable error. After every reconnect it subscribes again (clean session: the broker forgets subscriptions).
• Every message is mapped to EventData:
  • Data - message body (byte[]), Channel - topic the message arrived on,
  • DataContentType - MQTT 5 content type, ReplyTo - MQTT 5 response topic, Headers - MQTT 5 user properties (last value wins for repeated names),
  • Id - a new random id (MQTT has no message id), Time - receive time; Source, Type and Subject stay null.
• Each message gets new metadata with CorrelationId = Id.
• Messages are processed one at a time, in order: the next message is taken after the flow finished the previous one, and QoS 1 / 2 acknowledgements are sent only then. If processing fails, the error is logged and the message is skipped (it is acknowledged, not redelivered).

Notes

• Messages published while the flow is stopped or disconnected are lost unless the broker keeps them (retained messages, QoS with persistent sessions are not used: the session is clean).
• No TLS options for TCP connections; use a wss:// URI with WebsocketConnection for encryption.
• Keep secrets in expression settings, e.g. "Password": "=Secrets.MqttPassword".
• Data is binary; convert it to text downstream before parsing JSON.

Example

{ "Topic": "sensors/+/temperature", "Host": "mqtt.local", "Port": 1883, "QoS": 1 }

MQTT subscriber (Continue) → Script (decode Data) → Database writer