Sparkplug B

PlantPulse Edge Sparkplug B publisher (outbound) + command receiver (inbound) manual. Publishes NBIRTH / DBIRTH / DDATA / NDEATH on standard SPB 1.0 topics, and processes NCMD / DCMD sent by the host (SCADA), supporting tag writes and rebirth. Compatible with SCADA / HiveMQ / Ignition, etc.

Item	Value
Status	Phase 1 (BIRTH/DATA/DEATH) + Phase 3 (NCMD/DCMD inbound) active. Phase 2 (alias) is a toggle option
Controller package	plantpulse.app.edge.component.queue.transfer.sparkplug
Default	sparkplug.enable=true (operational active)
Command reception	Subscribes to and processes NCMD/DCMD when sparkplug.cmd.enable=true
Broker	Edge's HiveMQ (mqtt.server.* reuse)
Authentication	MQTT username/password (app.properties's mqtt.server.user / mqtt.server.password). 2026.05+ install.sh auto-generates a random password per box. The HiveMQ broker's auth.properties is auto-synced with app.properties's mqtt.* by container-entrypoint at every boot — operators need only change app.properties

2026.05+ container mode secrets

MQTT accounts are injected via MQTT_USER / MQTT_PASSWORD_FILE or MQTT_PASSWORD. When an external SCADA (Ignition / HiveMQ Edge, etc.) connects to this box via Sparkplug, the operator must look up the password from the managed secret file before use.

---

Topic schema

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

Item	Mapping
group_id	EdgeContext.site_id (e.g., SITE_00001). Can be overridden by sparkplug.group.id
edge_node_id	EdgeContext.id (e.g., EDGE_00303). Can be overridden by sparkplug.edge.node.id
device_id	opc_id (e.g., OPC_UA_Kepware) — sparkplug.device.strategy=per-opc
metric name	tag_id

---

Message types

Message	Topic	Payload core	Transmission timing
NBIRTH	spBv1.0/{g}/NBIRTH/{n}	bdSeq metric, seq=0	Immediately after connection
DBIRTH	spBv1.0/{g}/DBIRTH/{n}/{d}	All tag metrics for that OPC (initial null), seq increments	Immediately after connection, per OPC
DDATA	spBv1.0/{g}/DDATA/{n}/{d}	Single metric (tag_id, value, datatype, ts), seq increments	When one Point arrives from the queue
NDEATH	spBv1.0/{g}/NDEATH/{n}	bdSeq metric, no seq (spec)	MQTT will (abnormal termination) / explicit publish on disconnect
NCMD	spBv1.0/{g}/NCMD/{n}	Host → edge node command (e.g., Node Control/Rebirth=true)	User action / timeout / state sync by the host
DCMD	spBv1.0/{g}/DCMD/{n}/{d}	Host → edge device metric write (target metric name or alias + new value)	When host changes a specific OPC tag value as a setpoint

---

Lifecycle / Sequence

1. bdSeq — wraps from 0 to 255, persisted to disk (./data/sparkplug/bdseq default). +1 at every session start.
2. seq — wraps from 0 to 255. Starts at 0 in NBIRTH, increments by 1 at every DBIRTH/DDATA publish.
3. NDEATH = MQTT Will — registered as will at CONNECT so the broker auto-publishes on abnormal termination. On normal termination, disconnect() explicitly publishes.

---

Data type mapping

APP_TAG.data_type (free-form string) → MetricDataType (case-insensitive, trimmed).

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
Other / null	String (fallback, log.warn)

Since Point.value is stored as String, it is cast to the appropriate type above before being put into the SPB metric value. If casting fails, the metric value is published as null.

---

Configuration (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 토글 (현 무영향)

---

Components

File	Role
SparkPlugQueueTransfer.java	QueueTransfer implementation. Exposes only init/trans/close
SparkPlugSession.java	Lifecycle (CONNECT+will / NBIRTH / DBIRTH / DDATA / disconnect-NDEATH / seq / bdSeq)
SparkPlugConfig.java	Property → config object (broker reuses mqtt.server.*)
SparkPlugTopicBuilder.java	Topic string builder
SparkPlugDataTypeMapper.java	data_type → MetricDataType + value coerce
SparkPlugBdSeqStore.java	bdSeq disk persistence (0 ~ 255 wrap)
PointQueueProcessorFactory.java	Adds SparkPlugQueueTransfer to transfer_list

---

Dependency jars

Add to WebContent/WEB-INF/lib/:

jar	Role
tahu-core-1.0.14.jar	Eclipse Tahu Sparkplug B payload model + encoder
protobuf-java-3.25.5.jar	tahu's protobuf dependency
commons-compress-1.27.1.jar	tahu's commons-compress dependency

lib sync

The jars in WEB-INF/lib are materialized from Nexus at build time, so they are not committed. New dependencies must have their coordinates added in build.gradle.

---

Validation (dev server / 2026-05-06)

Capture spBv1.0/# with paho-mqtt subscriber:

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

DBIRTH metric count exactly matches the number of OPC tags in APP_TAG (OPC_UA_Kepware=11, OPC_00303=56, OPC_LS_XBM_0001=28, etc.).

Example DDATA topic:

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

Payload decoding verification tool:

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

(This script is a temporary file and is not included in build artifacts.)

---

Phase 3 — Command (NCMD/DCMD) Receive Behavior

When sparkplug.cmd.enable=true, SparkPlugSession automatically subscribes to the following topics.

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

Incoming messages are validated by the static method parseCmdTopic(String) → CmdTopic and then branched for processing.

Command	Handling
NCMD Node Control/Rebirth=true	rebirthAsync() — re-publish NBIRTH + re-publish DBIRTH for all OPCs + (when alias active) refresh aliasReverseMap (refreshAliasReverseMap())
DCMD metric write	Reverse-lookup tag_id by metric name or alias → call driver write(tagId, value) on the corresponding OPC → reflect the result in the next DDATA cycle

Topic validation (parseCmdTopic)

Input	Result
spBv1.0/Plant1/NCMD/EDGE_00303	OK (NCMD, deviceId=null)
spBv1.0/Plant1/DCMD/EDGE_00303/OPC_LS	OK (DCMD, deviceId=OPC_LS)
null / empty string / fewer than 4 segments	reject (null)
spBv1.0/Plant1/NDATA/EDGE_00303	reject (outbound type)
Other namespace (spBv2.0/...)	reject
spBv1.0/Plant1/DCMD/EDGE_00303 (4 seg, missing deviceId)	reject
6+ segments	ignore extra segments (only 4-5 used)

→ Unit tests SparkPlugCommandTopicTest guarantee the above 10 cases.

Rebirth safety net

The defect where rebirthAsync() omitted aliasReverseMap refresh after re-publishing NBIRTH/DBIRTH, causing the next DCMD to look up a stale alias, is fixed — refreshAliasReverseMap() call added (subscribeCommands() uses the same helper, DRY).

---

Roadmap

Phase	Content	Status
1 — BIRTH/DATA/DEATH	NBIRTH / DBIRTH / DDATA / NDEATH(will)	✅ Done
2 — alias optimization	Create tag_id → alias(Long) table at NBIRTH/DBIRTH, publish only aliases in DDATA (payload reduced by 30 ~ 40%). Toggle: sparkplug.alias.enable	Optional (toggle)
3 — NCMD/DCMD inbound	Process host metric write commands + Node Control/Rebirth	✅ Done (sparkplug.cmd.enable=true)
4 — STATE / operational visibility	Subscribe to STATE/{primary_host_id} → hold DDATA when primary OFFLINE. Add sparkplug section to /api/v1/edge response (connected / bdSeq / seq / devices)	TBD

---

Known limitations

• No isNull flag when DDATA value=null: Currently only Metric.setValue(null) is called. Per spec, explicitly setting isNull=true is safer → candidate for follow-up enhancement.
• Single metric per DDATA: Since the queue transfers one point at a time, batch publish optimization is not done. If frequency increases, could evolve to batched DDATA.
• Default bdSeq path is relative to working directory: On production servers, an explicit absolute path (/opt/kopens/.../data/sparkplug/bdseq) is recommended.
• No DCMD write failure response implemented: Currently, even if the driver write() fails, no result is sent to the host via NDATA/DDATA. Adding a result_status metric is under consideration for the future.