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 自己完結クライアント (CONNECT / SUBSCRIBE / PUBLISH / PINGREQ) |
SparkplugBPayload | Sparkplug B Payload protobuf ワイヤエンコーダ/デコーダ (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 時にワイルドカードの一部として使用。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 部分のワイルドカード)。
Sparkplug Datatype マッピング
SparkplugBPayload.Metric.formatValue() がワイヤデータ型ごとに適切な 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 ルックアップが失敗し無視されます (キャッシュ更新なし)。broker 再起動後に publisher が NBIRTH を再発行するまで待機。
NDEATH / DDEATH
Edge node または device が終了すると、broker が LWT (Last Will and Testament) で NDEATH/DDEATH を publish。ドライバは alias map のみクリアし、キャッシュは維持 — 最後の既知値がポーリング結果として保持されます (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 — ワイヤは 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 解決、group 不一致ブロック、DEATH alias クリア
- datatype ごとの formatValue 正確性 (Int32 / Float / Double / Boolean / String / UInt32 / UInt64 / is_null)
- Payload encode→decode round-trip
- MQTT remaining length / utf8 ワイヤユーティリティ
- safeId clientId サニタイゼーション