MQTT Client Driver — Technical Reference

The MQTT client driver in PlantPulse Edge is a Java implementation wrapping the Eclipse Paho mqttv3 1.2.5 library. It subscribes to topics on an external IIoT broker (HiveMQ / Mosquitto / EMQX / AWS IoT Core / Azure IoT Hub, etc.), keeps the messages in an in-memory cache, and returns the cached value on each tag polling cycle. Writes are published immediately.

Source: plantpulse.driver.protocol.mqtt_client.*

Class	Responsibility
`MQTTClientDriver`	`ProtocolDriver` implementation — connect/read/write/close, topic cache, lazy subscribe

Library: lib/org.eclipse.paho.client.mqttv3-1.2.5.jar

Differences from Sparkplug B

PlantPulse provides two separate MQTT-based drivers.

Driver	payload	When to use
`MQTT_CLIENT` (this page)	Arbitrary String/JSON/binary	General IIoT brokers, user-defined topics, ASCII / JSON payloads
`SPARKPLUG_B`	Sparkplug B Payload (Protobuf)	Cirrus Link / Tahu / Ignition standard, NBIRTH/DBIRTH alias

Sparkplug B includes its own wire codec plus alias learning logic; in environments that only need plain MQTT, this driver is lighter and the standard Paho API alone is sufficient.

Operation flow

broker ──PUBLISH──> Paho MqttClient ──MqttCallback.messageArrived(topic, payload)──> Driver
                                                                                       │
                                                                                       └─ lastValueByTopic.put(topic, new String(payload))

PlantPulse 폴러 ──read(addr)──> lastValueByTopic[addr]   (in-memory hit)
                  write(addr, value) ──Paho.publish(topic, value, qos, retain=false)──> broker

On the first read, a cache miss triggers a lazy subscribe (using the QoS option value) and returns an empty string — the actual value arrives from the next cycle onward. Topics registered in tag_map are subscribed in bulk via preSubscribeFromTagMap() immediately after connect, so a value may already arrive on the first cycle (depending on publish frequency).

Options

Key	Default	Description
`username`	(none)	broker authentication username
`password`	(none)	broker authentication password
`tls`	`false`	`true` → `ssl://` (default 8883), `false` → `tcp://` (default 1883)
`qos`	`0`	subscribe / publish QoS (0 / 1 / 2). Out-of-range values are clamped automatically.
`keep-alive`	`60`	keep-alive interval (seconds) — `MqttConnectOptions.setKeepAliveInterval`
`clean-session`	`true`	`MqttConnectOptions.setCleanSession`
`client-id`	(automatic)	explicit client id. If not specified, `PP-<opc_id>-<random6>` (≤23 chars) is generated automatically

MqttConnectOptions.setAutomaticReconnect(true) is always enabled, so the connection is automatically re-established after a temporary broker outage (using the Paho default backoff).

Address mapping

address	Behavior
`factory/line1/temp`	single topic — the last message on exactly this topic
`device/+/status`	wildcard (single level) — the last message among matching topics (most recently received)
`factory/#`	wildcard (multi) — last segment, all sub-topics
`""` / `null`	returns empty

Wildcard caveat: The Paho callback does not cache a separate entry per matching pattern — the topic on which the message actually arrived becomes the key. That is, even if you subscribed with device/+/status, lastValueByTopic holds device/sensor01/status and device/sensor02/status as separate entries, and reading with the wildcard topic itself finds no entry in the cache (a wildcard is never a publish destination). When registering a wildcard pattern such as device/+/status, you must separately track the value of the last arriving topic for the read result — for normal use, explicit per-topic registration is recommended.

write (publish)

client.publish(topic, value.getBytes(), qos, /*retain=*/ false);

The bytes of value are sent as the payload as-is (UTF-8 default charset).
retain=false is fixed and cannot be changed — since the broker does not retain the last message, a client that subscribes later will not receive a value until the next write occurs. For consumers that require retained messages, use a broker-side bridge or a separate publisher.
QoS uses the option value as-is. With QoS 1/2, Paho internally retransmits until acknowledged.

Connection lifecycle

Method	Behavior
`connect()`	apply options → assemble broker URL → `MqttClient.connect()` → bulk subscribe `tag_map`
`read()`	lazy subscribe on cache miss → return value on cache hit
`write()`	publish (retain=false)
`close()`	`disconnect()` + `close()` (best-effort, exceptions ignored) → clear cache
`connectionLost(cause)`	`setConnected(false)` in the callback — Paho attempts automatic reconnection

Limitations and future work

mTLS / client certificates — currently username/password only. X.509 mTLS environments such as AWS IoT require separate truststore / keystore configuration (Paho MqttConnectOptions.setSocketFactory exposure not implemented).
retained / will message — currently fixed at retain=false, LWT not configured.
wildcard cache separation — reading with the wildcard pattern itself returns empty (entries are per arriving topic).
JSON path extraction — even if the payload is JSON, the driver returns only the raw String. JSONPath extraction must be handled at the format / formula stage (or by a separate parser node).
payload encoding — currently uses the platform default charset. Binary payloads may be corrupted under UTF-8 → exposing a payload-encoding option (utf8 / iso-8859-1 / base64) is under review.

Test location

test/java/plantpulse/driver/protocol/mqtt_client/MQTTClientDriverTest.java — 31 tests.

initial state / metadata (isConnected/isWriteSupported/isExternal/getDriverSource)
safety of read/write/null/empty while disconnected
getDebugBaseUrl — tcp/ssl, explicit port, default port (1883/8883)
option parsing — username/password/tls/keep-alive/clean-session/qos/client-id, invalid values → default
static helpers — clampQos, defaultClientId (length/random/null), nullIfEmpty, parseInt, parseBool
constant verification (DEFAULT_PORT, DEFAULT_PORT_TLS, DEFAULT_KEEP_ALIVE_SEC, DEFAULT_QOS)

Integration tests that depend on a real broker are maintained separately — these unit tests run without an external broker.

Differences from Sparkplug B​

Operation flow​

Options​

Address mapping​

write (publish)​

Connection lifecycle​

Limitations and future work​

Test location​