Sparkplug B 驱动 — 技术参考

PlantPulse Edge 的 Sparkplug B 驱动是 MQTT 3.1.1 client + Sparkplug B Payload Protobuf 解码器 的原生（native）Java 实现。不依赖外部 paho/tahu jar，因此以单个 jar 部署时无需额外配置即可运行。

源码：plantpulse.driver.protocol.sparkplug.*

类	职责
SparkplugBDriver	ProtocolDriver 实现类 — connect/read/write/close、alias·metric 缓存
SparkplugMqttClient	MQTT 3.1.1 自包含 client（CONNECT / SUBSCRIBE / PUBLISH / PINGREQ）
SparkplugBPayload	Sparkplug B Payload protobuf wire 编码器/解码器（varint / fixed32 / fixed64 / length-delim）

---

运行流程

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

---

选项

键	默认值	说明
group-id	default	Sparkplug Group ID。subscribe 时作为 wildcard 的一部分使用。group 不匹配的消息由 driver 过滤。
edge-node-id	+	Edge node 过滤器。为 + 时表示 group 内所有节点。
keep-alive	60	MQTT keep-alive 秒数。为 0 时禁用 ping。
connect-timeout	5000	TCP 连接超时（ms）。

Subscribe topic 按 spBv1.0/<group-id>/+/<edge-node-id>/# 模式构成（+ 为 message_type，# 为 device 部分的 wildcard）。

---

Sparkplug Datatype 映射

SparkplugBPayload.Metric.formatValue() 会按 wire datatype 转换为相应的 String。

Sparkplug datatype	存储字段	format 结果
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	原样 UTF-8
Bytes / File	bytesValue	hex string

is_null=true metric 将被求值为空字符串。

---

NBIRTH / DBIRTH 的 alias 学习

Sparkplug 高效性的核心：BIRTH 发送 metric 名称与 alias 的配对 → 之后的 NDATA/DDATA 仅发送 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")

若在缺少 NBIRTH 的状态下收到 NDATA，alias 查找将失败并被忽略（不更新 cache）。需等待 broker 重启后 publisher 重新发布 NBIRTH。

---

NDEATH / DDEATH

Edge node 或 device 终止时，broker 通过 LWT（Last Will and Testament）publish NDEATH/DDEATH。驱动仅清空 alias map，保留 cache — 最后的已知值将作为轮询结果保留（PlantPulse 的 monitor 会照常显示 lastValue UI）。

---

NCMD / DCMD 发送

write(ProtocolAddress) 会根据 address 段数自动分支。

address	publish topic	备注
edge1/SetPoint	spBv1.0/<group>/NCMD/edge1	node 级命令
edge1/dev1/SetPoint	spBv1.0/<group>/DCMD/edge1/dev1	device 级命令

phase 1 将所有命令以 Sparkplug datatype String publish — 由接收侧（Tahu / Ignition）转换为 metric 所定义的 datatype 后处理。

---

限制及后续工作

• TLS / 认证 — 当前仅支持明文 + 匿名。username/password / mTLS 选项待后续支持。
• QoS — 所有 publish/subscribe 均为 QoS 0。Sparkplug 推荐仅 NBIRTH 使用 QoS 1，NDATA/DDATA 使用 QoS 0。
• DataSet / Template metric — wire 以 raw bytes 保留（datasetRaw/templateRaw），但 decode/format 尚未实现。
• Properties / Metadata — 仅保留 raw bytes，未实现 decode。
• bdSeq — 尚未实现 NBIRTH 的 bdSeq 匹配 / NDEATH 校验。

---

测试位置

test/java/plantpulse/driver/protocol/sparkplug/SparkplugBDriverTest.java — 27 项测试。

• 初始状态 / 选项解析
• 未连接时 read/write 的安全性
• BIRTH alias 注册、NDATA 仅凭 alias 解析、group 不匹配拦截、DEATH alias 清理
• 各 datatype 的 formatValue 准确性（Int32 / Float / Double / Boolean / String / UInt32 / UInt64 / is_null）
• Payload encode→decode 往返
• MQTT remaining length / utf8 wire 工具
• safeId clientId sanitization