MQTT

The MQTT driver (io.scadex.driver.mqtt) connects the gateway to MQTT brokers. Each tag bound to it subscribes to a topic filter and decodes the payload into a value; a write to such a tag publishes to its topic. This page is for engineers who set up broker connections and tags.

It is built on MQTTnet and declares SDK contract 1.2.

What has been proven

The driver has run against Mosquitto. No other broker has been tried.

Adding a connection

Open Connections, then Devices, then MQTT (/connections/mqtt), and add a broker. The form is rendered from the driver's descriptor:

Field	Key	Meaning	Default
Broker URL	url	Required. Where the broker listens, such as mqtt://host:1883. See the schemes below.	none
Username	username	Credentials sent at connect.	empty
Password	password	Secret; stored encrypted with the module's state and never returned by the API.	empty
Client ID	clientId	The MQTT client id. Empty means scadex-<connection id>.	empty
Default QoS	qos	0 (at most once), 1 (at least once) or 2 (exactly once). Used for every subscription and every publish of this connection.	0

Every connection also has a unique name and an Enabled switch.

URL schemes

Scheme	Transport	Default port
mqtt (also tcp)	TCP, plain	1883
mqtts (also ssl, tls)	TCP with TLS	8883
ws	WebSocket, plain	the port in the URL
wss	WebSocket with TLS	the port in the URL

Any other scheme is refused with "Unsupported broker scheme". A username or password inside the URL is refused: put them in their own fields. With a TLS scheme the driver turns TLS on with the library's defaults; there are no further TLS options (no client certificate, no custom CA) in this version.

The session is opened with a clean session. The saved password is reused only while the host stays the same.

Test

Test connection connects with a throwaway client id (<clientId or scadex>-test-<random>), gives up after 5 seconds, and reports success and latency. The running link is not touched.

How the link runs

• The driver connects with a 10-second timeout and retries every 5 seconds after a failure or disconnect. While disconnected every bound tag is marked bad with its last value kept.
• Each bound tag subscribes to its topic filter at the connection's QoS. A filter already held is re-subscribed, which also makes the broker resend what is retained.
• A message larger than 512 KiB marks the tag bad with "The ... byte message is too large." Queued payloads are capped at 64 MiB per connection; beyond that samples are dropped and counted as Dropped samples.
• Restarting a tag from the tag browser re-subscribes it, so the broker sends its retained value again.

Topics and tags

A tag bound to this driver has the binding document:

{ "topic": "plant/line1/temperature", "payloadFormat": "number" }

Field	Meaning
topic	A topic or a topic filter. + matches one level and # the rest; # must be last and alone in its level, + alone in its level. A filter starting with # or + does not match topics starting with $.
payloadFormat	How to decode the payload. Missing or raw keeps the text.

payloadFormat	Decoding
raw (default)	The payload as a UTF-8 string
number	A finite decimal number, otherwise bad quality "... is not a number."
boolean	true, 1, on or false, 0, off, case-insensitive; otherwise bad
json	Parsed JSON: an object or array becomes a JSON value, a scalar becomes that scalar; invalid JSON is bad

There is no JSON path selection: a json payload lands as one value. Pick a field in an expression tag or a derived member instead (see Tags).

Values are published with the gateway's receive time as the timestamp and quality Good when decoding succeeds.

Browsing for topics

Browse topics on a connection page starts a scan: the driver subscribes to the connection's scan filters (the saved scanTopics, or # when none are set) and lists every topic it sees, with the first 256 bytes of the last payload and whether it was retained. Up to 5,000 topics are kept. The first look waits one second for retained messages. The scan subscription is dropped again two minutes after the last browse, so the broker is not flooded with wildcard traffic for longer than someone is looking.

Pick topics and choose Create a tag. The dashboard asks the driver for proposals (discover), which bind each topic with payloadFormat: "raw", and submits them to the gateway's tag API. Change the format on the tag afterwards. The suggested tag name is the last segment of the topic after /, ., ; or =, non-alphanumeric characters replaced by _.

Writing

A write to a bound tag publishes the value to its topic at the connection's QoS, not retained. The payload is text: true/false for booleans, the shortest round-trip form for numbers, base64 for bytes, an empty payload for null, and the string form otherwise. A tag bound to a wildcard filter cannot be written. If the broker refuses the publish the write fails with "The broker refused the write: <reason>".

Connection status

Each connection is published every two seconds under @gateway.connection.mqtt.<name>.*, where <name> is the connection name turned into a path segment. The leaves are name, state (disabled, connecting, connected or faulted), connected, enabled, address, tags, error and since, plus the driver's details: valuesPerSecond, messagesPerSecond, good, uncertain, bad and waiting (bound tags by last quality), topics (topics seen), lastMessage (ISO 8601) and dropped. Numbers and booleans are real tags with history. The connection page shows Tags subscribed, Topics seen, Last message and Dropped samples.

Routes

The baseline driver routes are mapped under /api/v1/modules/io.scadex.driver.mqtt: GET|POST /connections, GET|PUT|DELETE /connections/{connectionId}, POST /connections/{connectionId}/test, GET /connections/{connectionId}/tags, POST /connections/{connectionId}/browse and POST /connections/{connectionId}/discover. The module adds:

Request	Purpose
GET /servers, GET /servers/{id}	The saved brokers, passwords redacted
POST /servers, PUT /servers/{id}, DELETE /servers/{id}	Create (name, url, username, password, clientId, qos, enabled), update and delete
GET /status	Whether the module runs and, per connection, connected, tagCount, topicCount, droppedSamples, lastMessageAt and error
GET /servers/{id}/tags	The tags bound to this broker with their topic and payloadFormat
GET /servers/{id}/browse	The topics seen by the scan: topic, value, retained, source
PUT /servers/{id}/browse	Body {"topics": [...]}: the filters the scan subscribes to; an invalid filter answers 400

GET routes need the viewer role; the others need engineer. While the module is stopped these routes answer 503; see Modules.