---
id: event--mqtt-subscriber
title: "MQTT subscriber"
summary: "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."
audience: user
keywords: "start, continue, topic, clientid, host, port, username, password, websocketconnection, usemqtt5, useretainflag, qos"
---
# 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

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

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