Sparkplug B

Handbuch für den Sparkplug B Publisher (Outbound) + Command-Listener (Inbound) von PlantPulse Edge. Es veröffentlicht standardisierte SPB 1.0-Topics wie NBIRTH / DBIRTH / DDATA / NDEATH und verarbeitet NCMD / DCMD, die vom Host (SCADA) gesendet werden, einschließlich Tag-Schreibvorgängen und Rebirth. Kompatibel mit SCADA / HiveMQ / Ignition usw.

Punkt	Wert
Status	Phase 1 (BIRTH/DATA/DEATH) + Phase 3 (NCMD/DCMD Inbound) aktiv. Phase 2 (Alias) optional per Toggle
Controller-Paket	`plantpulse.app.edge.component.queue.transfer.sparkplug`
Standard	`sparkplug.enable=true` (Betrieb aktiv)
Befehlsempfang	`sparkplug.cmd.enable=true`: NCMD/DCMD-Abonnement und -Verarbeitung
Broker	HiveMQ des Edge (`mqtt.server.*` wird wiederverwendet)
Authentifizierung	MQTT username/password (`app.properties`, `mqtt.server.user` / `mqtt.server.password`). Ab 2026.05+ generiert install.sh automatisch ein zufälliges Passwort pro Box. Die `auth.properties` des HiveMQ-Brokers werden vom container-entrypoint bei jedem Boot automatisch mit den mqtt.*-Werten aus `app.properties` synchronisiert – der Bediener ändert nur `app.properties`

Geheimnisse im Container-Modus ab 2026.05+

MQTT-Konten werden über MQTT_USER / MQTT_PASSWORD_FILE oder MQTT_PASSWORD injiziert. Wenn eine externe SCADA (Ignition / HiveMQ Edge usw.) sich über Sparkplug mit dieser Box verbindet, muss der Bediener das Passwort aus der verwalteten Secret-Datei entnehmen und verwenden.

Topic-Schema

spBv1.0/{group_id}/(N|D){BIRTH|DEATH|DATA|CMD}/{edge_node_id}[/{device_id}]

Punkt	Zuordnung
`group_id`	`EdgeContext.site_id` (z. B.: `SITE_00001`). Kann über `sparkplug.group.id` überschrieben werden
`edge_node_id`	`EdgeContext.id` (z. B.: `EDGE_00303`). Kann über `sparkplug.edge.node.id` überschrieben werden
`device_id`	`opc_id` (z. B.: `OPC_UA_Kepware`) — `sparkplug.device.strategy=per-opc`
Metrik `name`	`tag_id`

Nachrichtentypen

Nachricht	Topic	Kern des Payloads	Zeitpunkt der Veröffentlichung
NBIRTH	`spBv1.0/{g}/NBIRTH/{n}`	`bdSeq`-Metrik, `seq=0`	Direkt nach der Verbindung
DBIRTH	`spBv1.0/{g}/DBIRTH/{n}/{d}`	Alle Tag-Metriken der jeweiligen OPC (initial null), `seq` erhöht	Direkt nach der Verbindung, pro OPC
DDATA	`spBv1.0/{g}/DDATA/{n}/{d}`	Einzelne Metrik (`tag_id`, value, datatype, ts), `seq` erhöht	Bei Ankunft eines Points aus der Warteschlange
NDEATH	`spBv1.0/{g}/NDEATH/{n}`	`bdSeq`-Metrik, kein seq (laut Spezifikation)	MQTT-Will (bei unerwartetem Beenden) / explizite Veröffentlichung beim Disconnect
NCMD	`spBv1.0/{g}/NCMD/{n}`	Host → Edge-Knoten-Befehl (z. B.: `Node Control/Rebirth=true`)	Benutzeraktion des Hosts / Timeout / State-Sync
DCMD	`spBv1.0/{g}/DCMD/{n}/{d}`	Host → Edge-Gerät: Metrik-Schreibvorgang (Zielmetrikname oder Alias + neuer Wert)	Wenn der Host einen bestimmten Tag-Wert der OPC als Sollwert ändert

Lebenszyklus / Ablauf

bdSeq — Wrap von 0 bis 255, Disk-Persistenz (Standard: ./data/sparkplug/bdseq). Bei jedem Sessionstart +1.
seq — Wrap von 0 bis 255. Startet bei 0 in der NBIRTH, wird bei DBIRTH und DDATA jeweils zum Zeitpunkt der Veröffentlichung um 1 erhöht.
NDEATH = MQTT-Will — wird beim CONNECT als Will registriert, sodass der Broker bei unerwartetem Beenden automatisch veröffentlicht. Bei normalem Beenden wird disconnect() explizit veröffentlicht.

Datentyp-Zuordnung

APP_TAG.data_type (freie Zeichenkette) → MetricDataType (Groß-/Kleinschreibung ignorieren, trimmen).

Edge	SPB
Float / Single	Float
Double	Double
Byte / Int8	Int8
Short / Int16	Int16
Int / Integer / Int32	Int32
Long / Int64	Int64
UInt8/16/32/64	UInt8/16/32/64
Boolean / Bool	Boolean
String / Text	String
DateTime / Date	DateTime
Sonstiges / null	String (Fallback, log.warn)

Point.value wird als String gespeichert und muss daher vor dem Einsetzen als SPB-Metrikwert in den oben genannten Typ umgewandelt werden. Bei fehlgeschlagener Umwandlung wird der Metrikwert als null veröffentlicht.

Konfiguration (`src/app.properties`)

# --------------------------------------------------------------------------------------
# SPARKPLUG B
# 브로커 접속은 mqtt.server.* 를 그대로 재사용한다.
# --------------------------------------------------------------------------------------
sparkplug.enable=false
sparkplug.group.id=                  # 비우면 site_id
sparkplug.edge.node.id=              # 비우면 EdgeContext.id
sparkplug.device.strategy=per-opc    # per-opc | single (현재 per-opc 만 구현)
sparkplug.qos=0                      # DDATA 빈번 → 0 권장
sparkplug.bdseq.path=                # 비우면 ./data/sparkplug/bdseq
sparkplug.alias.enable=true          # Phase 2 토글 (현 무영향)
sparkplug.cmd.enable=false           # Phase 3 토글 (현 무영향)

Komponenten

Datei	Rolle
`SparkPlugQueueTransfer.java`	Implementierung von `QueueTransfer`. Stellt nur `init/trans/close` bereit
`SparkPlugSession.java`	Lebenszyklus (CONNECT+Will / NBIRTH / DBIRTH / DDATA / disconnect-NDEATH / seq / bdSeq)
`SparkPlugConfig.java`	Eigenschaften → Konfigurationsobjekt (Broker: `mqtt.server.*` wird wiederverwendet)
`SparkPlugTopicBuilder.java`	Erzeugt Topic-Strings
`SparkPlugDataTypeMapper.java`	data_type → MetricDataType + Wert-Koerzierung
`SparkPlugBdSeqStore.java`	bdSeq Disk-Persistenz (Wrap 0 bis 255)
`PointQueueProcessorFactory.java`	Fügt `SparkPlugQueueTransfer` zur transfer_list hinzu

Abhängigkeits-JARs

Zusätzlich zu WebContent/WEB-INF/lib/:

JAR	Rolle
`tahu-core-1.0.14.jar`	Eclipse Tahu Sparkplug B-Payload-Modell + Encoder
`protobuf-java-3.25.5.jar`	Tahu-Protobuf-Abhängigkeit
`commons-compress-1.27.1.jar`	Tahu-commons-compress-Abhängigkeit

Bibliotheks-Synchronisierung

Die JARs aus WEB-INF/lib werden beim Build aus dem Nexus materialisiert und daher nicht eingecheckt. Für neue Abhängigkeiten müssen die Koordinaten in build.gradle hinzugefügt werden.

Verifikation (Dev-Server / 2026-05-06)

Mit paho-mqtt-Subscriber wird spBv1.0/# erfasst:

[SUMMARY] total=511
  NDEATH: 1   ← 재시작 시 will 자동 발행
  NBIRTH: 1   ← seq=0
  DBIRTH: 10  ← OPC 10개 모두, seq=1..10
  DDATA: 499  ← 13초간 ≈ 38 msgs/s

Die Anzahl der DBIRTH-Metriken entspricht exakt der Anzahl der zugehörigen OPC-Tags von APP_TAG (OPC_UA_Kepware=11, OPC_00303=56, OPC_LS_XBM_0001=28 usw.).

Beispiel für ein DDATA-Topic:

spBv1.0/SITE_00001/DDATA/EDGE_00303/OPC_UA_Kepware  bytes=44  seq=125  ts=...

Werkzeug zur Überprüfung der Payload-Dekodierung:

MQTT_USER="${MQTT_USER:-edge}" \
MQTT_PASSWORD="$(tr -d '\r\n' < /run/secrets/mqtt-password)" \
python3 /tmp/spb_sub.py 100.106.92.21 1883 "$MQTT_USER" "$MQTT_PASSWORD" 60

(Dieses Skript ist eine temporäre Datei und nicht Teil der Build-Artefakte.)

Phase 3 — Empfangsverhalten für Befehle (NCMD/DCMD)

Bei sparkplug.cmd.enable=true abonniert die SparkPlugSession automatisch die folgenden Topics.

spBv1.0/{group}/NCMD/{edge_node}
spBv1.0/{group}/DCMD/{edge_node}/+        # 모든 device 와일드카드

Eingehende Nachrichten werden über die statische Methode parseCmdTopic(String) → CmdTopic validiert und dann verzweigt verarbeitet.

Befehl	Verarbeitung
`NCMD` `Node Control/Rebirth=true`	`rebirthAsync()` — NBIRTH erneut veröffentlichen + DBIRTH für alle OPCs erneut veröffentlichen + (bei aktivem Alias) `aliasReverseMap` aktualisieren (`refreshAliasReverseMap()`)
`DCMD` Metrik-Schreibvorgang	Metrik `name` oder Alias → Rückwärtssuche nach tag_id → Aufruf des Treibers `write(tagId, value)` der entsprechenden OPC → Ergebnis wird im nächsten DDATA-Zyklus berücksichtigt

Topic-Validierung (parseCmdTopic)

Eingabe	Ergebnis
`spBv1.0/Plant1/NCMD/EDGE_00303`	OK (NCMD, deviceId=null)
`spBv1.0/Plant1/DCMD/EDGE_00303/OPC_LS`	OK (DCMD, deviceId=`OPC_LS`)
`null` / leere Zeichenkette / weniger als 4 Segmente	abgelehnt (null)
`spBv1.0/Plant1/NDATA/EDGE_00303`	abgelehnt (Outbound-Typ)
Anderer Namespace (`spBv2.0/...`)	abgelehnt
`spBv1.0/Plant1/DCMD/EDGE_00303` (4 Segmente, deviceId fehlt)	abgelehnt
6 oder mehr Segmente	zusätzliche Segmente werden ignoriert (nur 4–5 werden verwendet)

→ Die Unit-Tests SparkPlugCommandTopicTest garantieren die oben genannten 10 Fälle.

Rebirth-Schutznetz

Der Fehler, bei dem rebirthAsync() nach der erneuten Veröffentlichung von NBIRTH/DBIRTH die Aktualisierung von aliasReverseMap vergaß, sodass der nächste DCMD mit einem veralteten Alias suchte, ist behoben – der Aufruf refreshAliasReverseMap() wurde hinzugefügt (subscribeCommands() verwendet denselben Helfer, DRY).

Roadmap

Phase	Inhalt	Status
1 — BIRTH/DATA/DEATH	NBIRTH / DBIRTH / DDATA / NDEATH (Will)	✅ Abgeschlossen
2 — Alias-Optimierung	`tag_id → alias(Long)`-Tabelle zum Zeitpunkt von NBIRTH/DBIRTH erstellen, DDATA veröffentlicht nur den Alias (Payload-Reduktion um 30–40%). Toggle: `sparkplug.alias.enable`	Optional (Toggle)
3 — NCMD/DCMD Inbound	Verarbeitung von Metrik-Schreibbefehlen des Hosts + Node Control/Rebirth	✅ Abgeschlossen (`sparkplug.cmd.enable=true`)
4 — STATE / Betriebssichtbarkeit	`STATE/{primary_host_id}` abonnieren → DDATA zurückhalten, wenn primärer OFFLINE. `/api/v1/edge`-Antwort um `sparkplug`-Abschnitt erweitern (connected / bdSeq / seq / devices)	TBD

Bekannte Einschränkungen

Bei DDATA value=null wird das isNull-Flag nicht gesetzt: Derzeit wird nur Metric.setValue(null) aufgerufen. Laut Spezifikation ist eine explizite Angabe von isNull=true sicherer → Kandidat für spätere Verbesserung.
Eine einzelne Metrik pro DDATA: Da die Warteschlange einzelne Points überträgt, wird keine Optimierung für gebündelte Veröffentlichung durchgeführt. Bei höherer Frequenz könnte eine batched DDATA entwickelt werden.
Der Standardwert für den bdSeq-Pfad ist relativ zum Arbeitsverzeichnis: Auf dem Produktionsserver wird ein absoluter Pfad (/opt/kopens/.../data/sparkplug/bdseq) empfohlen.
Keine Antwort bei fehlgeschlagenem DCMD-Write: Wenn der Treiber write() fehlschlägt, wird dem Host derzeit kein Ergebnis über NDATA/DDATA gemeldet. Eine spätere result_status-Metrik wird in Betracht gezogen.

Topic-Schema​

Nachrichtentypen​

Lebenszyklus / Ablauf​

Datentyp-Zuordnung​

Konfiguration (src/app.properties)​

Komponenten​

Abhängigkeits-JARs​

Verifikation (Dev-Server / 2026-05-06)​

Phase 3 — Empfangsverhalten für Befehle (NCMD/DCMD)​

Topic-Validierung (parseCmdTopic)​

Rebirth-Schutznetz​

Roadmap​

Bekannte Einschränkungen​