Sparkplug B Treiber

Der Sparkplug B Treiber von PlantPulse Edge ist eine native Java-Implementierung eines MQTT 3.1.1 Clients + Sparkplug B Payload Protobuf Dekoders. Er hat keine Abhängigkeiten von externen paho/tahu JARs und funktioniert daher ohne zusätzliche Konfiguration, wenn er als einzelnes JAR bereitgestellt wird.

Quelle: plantpulse.driver.protocol.sparkplug.*

Klasse	Verantwortung
`SparkplugBDriver`	Implementierung von `ProtocolDriver` — connect/read/write/close, Alias- und Metrik-Cache
`SparkplugMqttClient`	Eigenständiger MQTT 3.1.1 Client (CONNECT / SUBSCRIBE / PUBLISH / PINGREQ)
`SparkplugBPayload`	Sparkplug B Payload Protobuf Wire-Encoder/-Decoder (varint / fixed32 / fixed64 / length-delim)

Funktionsablauf

broker ──PUBLISH──> SparkplugMqttClient ──onMessage(topic, payload)──> SparkplugBDriver
                                                                       │
                                                                       ├─ topic 파싱 (spBv1.0/G/<msgType>/edge[/dev])
                                                                       ├─ Payload 디코드 → metric list
                                                                       ├─ NBIRTH/DBIRTH 시 alias→name 등록
                                                                       └─ metricCache[edge[/dev]/name] = formatValue()

PlantPulse 폴러 ──read(addr)──> metricCache[addr]   (in-memory hit, 외부 RTT 없음)
                  write(addr, value) ──encode──> NCMD/DCMD publish → broker

Optionen

Schlüssel	Standardwert	Beschreibung
`group-id`	`default`	Sparkplug-Gruppen-ID. Wird als Teil des Wildcards beim Subscribe verwendet. Nachrichten mit unpassender Gruppe werden vom Treiber verworfen.
`edge-node-id`	`+`	Edge-Node-Filter. Bei `+` werden alle Knoten innerhalb der Gruppe berücksichtigt.
`keep-alive`	`60`	MQTT Keep-Alive in Sekunden. Bei 0 ist Ping deaktiviert.
`connect-timeout`	`5000`	TCP-Verbindungs-Timeout in ms.

Das Subscribe-Topic wird nach dem Muster spBv1.0/<group-id>/+/<edge-node-id>/# aufgebaut (+ ist message_type, # ist der Wildcard für den Geräteteil).

Sparkplug-Datentyp-Zuordnung

SparkplugBPayload.Metric.formatValue() konvertiert je nach Wire-Datentyp in einen geeigneten String.

Sparkplug-Datentyp	Speicherfeld	Format-Ergebnis
Int8 / Int16 / Int32	`intValue` (varint)	signed int (`Integer.toString`)
UInt8 / UInt16 / UInt32	`intValue`	unsigned (`& 0xFFFFFFFFL`)
Int64 / DateTime	`longValue`	signed long
UInt64	`longValue`	`Long.toUnsignedString`
Float	`floatValue` (fixed32)	`Float.toString`
Double	`doubleValue` (fixed64)	`Double.toString`
Boolean	`booleanValue` (varint)	`"true"` / `"false"`
String / Text / UUID	`stringValue`	unverändert als UTF-8
Bytes / File	`bytesValue`	Hex-String

Metriken vom Typ is_null=true werden als leerer String ausgewertet.

Alias-Lernen bei NBIRTH / DBIRTH

Kernstück der Sparkplug-Effizienz: BIRTH sendet Metrikname und Alias-Paar → danach senden NDATA/DDATA nur noch den Alias.

1) edge1 의 NBIRTH:
   metric { name="Temperature" alias=7 datatype=Float float_value=23.5 }
   → driver: aliasByPrefix.put("edge1", {7 → "Temperature"})
            metricCache.put("edge1/Temperature", "23.5")

2) edge1 의 NDATA (이름 생략):
   metric { alias=7 datatype=Float float_value=24.1 }
   → driver: name 없음 → alias=7 → "Temperature"
            metricCache.put("edge1/Temperature", "24.1")

Wenn NDATA eintrifft, während NBIRTH fehlt, schlägt die Alias-Auflösung fehl und die Nachricht wird ignoriert (Cache wird nicht aktualisiert). Es wird gewartet, bis der Publisher nach einem Broker-Neustart erneut NBIRTH veröffentlicht.

NDEATH / DDEATH

Wenn ein Edge-Node oder ein Gerät beendet wird, veröffentlicht der Broker NDEATH/DDEATH über das LWT (Last Will and Testament). Der Treiber leert nur die Alias-Zuordnung und behält den Cache bei — der letzte bekannte Wert bleibt als Abfrageergebnis erhalten (der PlantPulse-Monitor zeigt das lastValue-UI unverändert an).

NCMD / DCMD-Senden

write(ProtocolAddress) verzweigt automatisch je nach Anzahl der Adresssegmente.

Adresse	Veröffentlichungs-Topic	Hinweis
`edge1/SetPoint`	`spBv1.0/<group>/NCMD/edge1`	Befehl auf Node-Ebene
`edge1/dev1/SetPoint`	`spBv1.0/<group>/DCMD/edge1/dev1`	Befehl auf Geräteebene

Phase 1 veröffentlicht alle Befehle als Sparkplug-Datentyp String — die Sendeseite (Tahu / Ignition) wandelt sie anhand des definierten Metrik-Datentyps um und verarbeitet sie.

Grenzen und zukünftige Arbeiten

TLS / Authentifizierung — derzeit nur Klartext + anonym. Optionen für Benutzername/Passwort / mTLS sind geplant.
QoS — alle Publish/Subscribe mit QoS 0. Die Sparkplug-Empfehlung sieht QoS 1 nur für NBIRTH vor, NDATA/DDATA mit QoS 0.
DataSet / Template-Metriken — Wire-Bytes bleiben erhalten (datasetRaw/templateRaw), aber Decodierung/Formatierung ist nicht implementiert.
Properties / Metadaten — nur Beibehaltung der Rohbytes, keine Decodierung.
bdSeq — Abgleich der bdSeq in NBIRTH / Validierung von NDEATH ist nicht implementiert.

Testort

test/java/plantpulse/driver/protocol/sparkplug/SparkplugBDriverTest.java — 27 Tests.

Anfangszustand / Options-Parsing
Sicheres Lesen/Schreiben im nicht verbundenen Zustand
BIRTH-Alias-Registrierung, NDATA-Alias-Auflösung, Abweisen bei Gruppenabweichung, DEATH-Alias-Bereinigung
Genauigkeit der formatValue je Datentyp (Int32 / Float / Double / Boolean / String / UInt32 / UInt64 / is_null)
Payload-Encode→Decode-Roundtrip
MQTT-Restlänge / UTF-8-Wire-Dienstprogramme
safeId-ClientID-Bereinigung

Funktionsablauf​

Optionen​

Sparkplug-Datentyp-Zuordnung​

Alias-Lernen bei NBIRTH / DBIRTH​

NDEATH / DDEATH​

NCMD / DCMD-Senden​

Grenzen und zukünftige Arbeiten​

Testort​