MQTT-Client-Treiber

Der MQTT-Client-Treiber von PlantPulse Edge ist eine Java-Implementierung, die die Eclipse Paho mqttv3 1.2.5-Bibliothek kapselt. Er abonniert Topics eines externen IIoT-Brokers (HiveMQ / Mosquitto / EMQX / AWS IoT Core / Azure IoT Hub usw.), speichert Nachrichten in einem In-Memory-Cache und gibt den Cache-Wert bei jedem Polling-Zyklus eines Tags zurück. Schreibvorgänge werden sofort veröffentlicht.

Quelle: plantpulse.driver.protocol.mqtt_client.*

Klasse	Verantwortung
`MQTTClientDriver`	`ProtocolDriver`-Implementierung – connect/read/write/close, Topic-Cache, Lazy-Subscribe

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

Unterschied zu Sparkplug B

PlantPulse bietet zwei verschiedene MQTT-basierte Treiber separat an.

Treiber	Payload	Verwendung
`MQTT_CLIENT` (diese Seite)	Beliebige Zeichenfolge/JSON/Binärdaten	Allgemeine IIoT-Broker, benutzerdefinierte Topics, ASCII-/JSON-Payload
`SPARKPLUG_B`	Sparkplug-B-Payload (Protobuf)	Cirrus-Link-/Tahu-/Ignition-Standard, NBIRTH/DBIRTH-Alias

Sparkplug B enthält einen eigenen Wire-Codec und eine Alias-Lernlogik. In Umgebungen, die nur allgemeines MQTT benötigen, ist dieser Treiber hier leichter und reicht mit der Standard-Paho-API aus.

Ablauf

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

Beim ersten read-Aufruf wird bei einem Cache-Miss ein Lazy-Subscribe (mit QoS-Optionswert) durchgeführt und eine leere Zeichenfolge zurückgegeben – ab dem nächsten Zyklus kommen die tatsächlichen Werte an. Topics, die in tag_map registriert sind, werden unmittelbar nach dem Verbindungsaufbau mittels preSubscribeFromTagMap() gesammelt abonniert, sodass Werte bereits im ersten Zyklus ankommen können (abhängig von der Veröffentlichungshäufigkeit der Nachrichten).

Optionen

Schlüssel	Standardwert	Beschreibung
`username`	(keiner)	Benutzername für die Broker-Authentifizierung
`password`	(keiner)	Passwort für die Broker-Authentifizierung
`tls`	`false`	`true` → `ssl://` (Standard 8883), `false` → `tcp://` (Standard 1883)
`qos`	`0`	QoS für Subscribe/Publish (0 / 1 / 2). Werte außerhalb des Bereichs werden automatisch begrenzt.
`keep-alive`	`60`	Keep-Alive-Intervall (Sekunden) — `MqttConnectOptions.setKeepAliveInterval`
`clean-session`	`true`	`MqttConnectOptions.setCleanSession`
`client-id`	(automatisch)	Explizite Client-ID. Wenn nicht angegeben, wird automatisch `PP-<opc_id>-<random6>` (≤ 23 Zeichen) erzeugt

MqttConnectOptions.setAutomaticReconnect(true) ist immer aktiv, sodass eine automatische Wiederverbindung nach vorübergehender Broker-Trennung erfolgt (Paho-Standard-Backoff).

Adresszuordnung

Adresse	Verhalten
`factory/line1/temp`	Einzelnes Topic – die letzte Nachricht genau dieses Topics
`device/+/status`	Wildcard (eine Ebene) – die letzte Nachricht aller passenden Topics (neueste Ankunft)
`factory/#`	Wildcard (mehrere Ebenen) – letztes Segment, alle Unter-Topics
`""` / `null`	leere Rückgabe

Achtung bei Wildcards: Der Paho-Callback cached nicht separat für jedes tatsächlich ankommende Topic; das ankommende Topic selbst wird zum Schlüssel. Wenn Sie also mit device/+/status abonnieren, werden unter lastValueByTopic device/sensor01/status und device/sensor02/status getrennt abgelegt. Liest der Benutzer direkt mit dem Wildcard-Topic, gibt es keinen Cache-Eintrag (ein Wildcard selbst ist kein Ziel einer Veröffentlichung). Bei der Registrierung von Wildcard-Mustern sollten Sie daher wie in device/+/status vorgehen und den Wert des zuletzt angekommenen Topics separat im read-Ergebnis verfolgen – im normalen Betrieb wird eine explizite Registrierung pro Topic empfohlen.

write (publish)

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

Die Bytes von value werden unverändert als Payload gesendet (Standard-Charset UTF-8).
retain=false ist festgelegt und kann nicht geändert werden – der Broker speichert die letzte Nachricht nicht, sodass Clients, die später abonnieren, keinen Wert erhalten, bis der nächste Schreibvorgang erfolgt. Für Verbraucher, die retained Nachrichten benötigen, richten Sie eine Broker-Bridge oder einen separaten Publisher ein.
QoS wird wie in den Optionen angegeben verwendet. Bei QoS 1/2 wiederholt Paho intern die Übertragung bis zur Bestätigung.

Verbindungs-Lebenszyklus

Methode	Verhalten
`connect()`	Optionen anwenden → Broker-URL zusammenbauen → `MqttClient.connect()` → `tag_map` Sammel-Subscribe
`read()`	Bei Cache-Miss Lazy-Subscribe → bei Cache-Treffer Wert zurückgeben
`write()`	Veröffentlichen (retain=false)
`close()`	`disconnect()` + `close()` (Best-Effort, Ausnahmen ignorieren) → Cache leeren
`connectionLost(cause)`	`setConnected(false)` im Callback – Paho versucht automatische Wiederverbindung

Einschränkungen und zukünftige Arbeiten

mTLS / Client-Zertifikate – derzeit nur Benutzername/Passwort. Für X.509-mTLS-Umgebungen wie AWS IoT sind separate Truststore-/Keystore-Einstellungen erforderlich (Paho-MqttConnectOptions.setSocketFactory-Freigabe nicht implementiert).
Retained-/Will-Nachrichten – derzeit retain=false festgelegt, kein LWT konfiguriert.
Wildcard-Cache-Trennung – Lesen mit einem Wildcard-Muster selbst liefert ein leeres Ergebnis (Einträge pro angekommenem Topic).
JSON-Pfad-Extraktion – Auch wenn die Payload JSON ist, gibt der Treiber nur rohe Zeichenfolgen zurück. Die JSONPath-Extraktion muss in format/Formelstufe erfolgen (oder über einen separaten Parser-Knoten).
Payload-Kodierung – derzeit wird der Standard-Charset der Plattform verwendet. Binäre Payloads können UTF-8-beschädigt werden → zukünftige Prüfung der payload-encoding-Option (utf8 / iso-8859-1 / base64).

Testort

test/java/plantpulse/driver/protocol/mqtt_client/MQTTClientDriverTest.java – 31 Tests.

Anfangszustand / Metadaten (isConnected/isWriteSupported/isExternal/getDriverSource)
Sicherheit bei read/write/null/empty im nicht verbundenen Zustand
getDebugBaseUrl – tcp/ssl, expliziter Port, Standard-Port (1883/8883)
Options-Parsing – Benutzername/Passwort/tls/keep-alive/clean-session/qos/client-id, ungültige Werte → Standard
Statische Helfer – clampQos, defaultClientId (Länge/random/null), nullIfEmpty, parseInt, parseBool
Konstantenprüfung (DEFAULT_PORT, DEFAULT_PORT_TLS, DEFAULT_KEEP_ALIVE_SEC, DEFAULT_QOS)

Integrationstests mit echtem Broker werden separat durchgeführt – diese Unit-Tests laufen ohne externen Broker.

Unterschied zu Sparkplug B​

Ablauf​

Optionen​

Adresszuordnung​

write (publish)​

Verbindungs-Lebenszyklus​

Einschränkungen und zukünftige Arbeiten​

Testort​