본문으로 건너뛰기

Sparkplug B Driver — 기술 레퍼런스

PlantPulse Edge 의 Sparkplug B 드라이버는 MQTT 3.1.1 client + Sparkplug B Payload Protobuf 디코더 의 자체(native) Java 구현입니다. 외부 paho/tahu jar 의존이 없어 단일 jar 배포 시 추가 설정 없이 동작합니다.

소스: plantpulse.driver.protocol.sparkplug.*

클래스책임
SparkplugBDriverProtocolDriver 구현체 — connect/read/write/close, alias·metric 캐시
SparkplugMqttClientMQTT 3.1.1 self-contained client (CONNECT / SUBSCRIBE / PUBLISH / PINGREQ)
SparkplugBPayloadSparkplug 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-iddefaultSparkplug Group ID. subscribe 시 wildcard 의 일부로 사용. group 불일치 메시지는 driver 가 거름.
edge-node-id+Edge node 필터. + 면 group 내 모든 노드.
keep-alive60MQTT keep-alive 초. 0 이면 ping 비활성.
connect-timeout5000TCP 연결 타임아웃 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 / Int32intValue (varint)signed int (Integer.toString)
UInt8 / UInt16 / UInt32intValueunsigned (& 0xFFFFFFFFL)
Int64 / DateTimelongValuesigned long
UInt64longValueLong.toUnsignedString
FloatfloatValue (fixed32)Float.toString
DoubledoubleValue (fixed64)Double.toString
BooleanbooleanValue (varint)"true" / "false"
String / Text / UUIDstringValueUTF-8 그대로
Bytes / FilebytesValuehex 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 수에 따라 자동 분기.

addresspublish topic비고
edge1/SetPointspBv1.0/<group>/NCMD/edge1node-level 명령
edge1/dev1/SetPointspBv1.0/<group>/DCMD/edge1/dev1device-level 명령

phase 1 은 모든 명령을 Sparkplug datatype String 으로 publish 합니다 — 송신 측 (Tahu / Ignition) 이 metric 의 정의된 datatype 으로 변환해 처리.


한계 및 향후 작업

  • TLS / 인증 — 현재 평문 + 익명 only. 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 round-trip
  • MQTT remaining length / utf8 wire utility
  • safeId clientId sanitization