Sparkplug B ドライバ — 技術リファレンス
PlantPulse Edge の Sparkplug B ドライバは、MQTT 3.1.1 client + Sparkplug B Payload Protobuf デコーダをネイティブ Java で実装したものです。外部の paho/tahu jar に依存しないため、単一 jar での配布時も追加設定なしで動作します。
ソース: plantpulse.driver.protocol.sparkplug.*
| クラス | 責務 |
|---|---|
SparkplugBDriver | ProtocolDriver 実装 — connect/read/write/close、alias・metric キャッシュ |
SparkplugMqttClient | MQTT 3.1.1 self-contained client (CONNECT / SUBSCRIBE / PUBLISH / PINGREQ) |
SparkplugBPayload | Sparkplug B Payload protobuf wire encoder/decoder (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 が一致しないメッセージはドライバが除外する。 |
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 lookup が失敗し無視されます (cache は更新されない)。broker 再起動後、publisher が NBIRTH を再発行するまで待機します。
NDEATH / DDEATH
Edge node または device が終了すると、broker が LWT (Last Will and Testament) として NDEATH/DDEATH を publish します。ドライバは alias map のみをクリアし、cache は保持します — 最後の known value がポーリング結果として保全されます (PlantPulse の monitor が lastValue UI をそのまま表示)。
NCMD / DCMD の送信
write(ProtocolAddress) は address segment 数に応じて自動的に分岐します。
| 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-only resolve、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