Skip to main content

MQTT Client Driver

PlantPulse Edge's MQTT client driver is a Java implementation that wraps 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.), stores 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.*

ClassResponsibility
MQTTClientDriverProtocolDriver 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.

DriverPayloadUse Case
MQTT_CLIENT (this page)Arbitrary String/JSON/binaryGeneral IIoT brokers, custom topics, ASCII / JSON payloads
SPARKPLUG_BSparkplug B Payload (Protobuf)Cirrus Link / Tahu / Ignition standard, NBIRTH/DBIRTH alias

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

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 call to read, if a cache miss occurs, a lazy subscribe (with the QoS option value) is performed and an empty string is returned — the actual value arrives from the next cycle onward. Topics registered in tag_map are batch-subscribed via preSubscribeFromTagMap() immediately after connect, so values may arrive in the very first cycle (depending on publish frequency).

Options​

KeyDefaultDescription
username(none)Broker authentication username
password(none)Broker authentication password
tlsfalsetrue → ssl:// (default 8883), false → tcp:// (default 1883)
qos0Subscribe / publish QoS (0 / 1 / 2). Values outside the range are automatically clamped.
keep-alive60Keep-alive interval (seconds) — MqttConnectOptions.setKeepAliveInterval
clean-sessiontrueMqttConnectOptions.setCleanSession
client-id(auto)Explicit client ID. If not specified, PP-<opc_id>-<random6> (≤23 characters) is automatically generated

MqttConnectOptions.setAutomaticReconnect(true) is always enabled, so automatic reconnection occurs after a temporary broker disconnection (Paho's default backoff applies).

Address Mapping​

AddressBehavior
factory/line1/tempSingle topic — the last message for exactly this topic
device/+/statusWildcard (one level) — the last message among matching topics (most recently arrived)
factory/#Wildcard (multi-level) — last segment, all sub-topics
"" / nullReturns empty

Wildcard caution: Paho's callback does not cache a separate entry per actual arrival topic; the arriving topic itself becomes the key. In other words, even if you subscribe with device/+/status, lastValueByTopic will contain device/sensor01/status and device/sensor02/status separately, and if the user reads using the wildcard topic itself, there is no entry in the cache (since wildcards themselves are not publish destinations). When registering a wildcard pattern, register it as in device/+/status and then separately track the value of the last arrival topic as the read result — for general use, explicit registration per topic is recommended.

write (publish)​

client.publish(topic, value.getBytes(), qos, /*retain=*/ false);
  • The bytes from value are sent as the payload as-is (default charset: UTF-8).
  • Fixed to retain=false and cannot be changed — since the broker does not preserve the last message, clients that start subscribing later will not receive a value until the next write occurs. For consumers that need retained messages, set up a broker-side bridge or a separate publisher.
  • QoS uses the option value as-is. With QoS 1/2, Paho internally retransmits until acknowledgment.

Connection Lifecycle​

MethodBehavior
connect()Applies options → assembles broker URL → MqttClient.connect() → batch subscribes tag_map
read()Lazy subscribe on cache miss → returns value on cache hit
write()Publishes (retain=false)
close()disconnect() + close() (best-effort, exceptions ignored) → clears 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 like AWS IoT require a separate truststore/keystore configuration (Paho MqttConnectOptions.setSocketFactory exposure not implemented).
  • Retained / will message — currently fixed to retain=false, LWT not set.
  • Wildcard cache separation — reading with the wildcard pattern itself returns an empty result (entries are keyed per arrival topic).
  • JSON path extraction — even if the payload is JSON, the driver returns only the raw String. JSONPath extraction must be handled in the format / formula stage (or via a separate parser node).
  • Payload encoding — currently uses the platform default charset. Binary payloads may break under UTF-8 → consider exposing the payload-encoding option (utf8 / iso-8859-1 / base64) in the future.

Test Location​

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

  • Initial state / meta (isConnected/isWriteSupported/isExternal/getDriverSource)
  • Unconnected read/write/null/empty safety
  • 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 separate — these unit tests operate without an external broker.