メインコンテンツまでスキップ

MQTTクライアントドライバ

PlantPulse Edge の MQTTクライアントドライバは、Eclipse Paho mqttv3 1.2.5 ライブラリを ラップしたJava実装です。外部IIoTブローカー（HiveMQ / Mosquitto / EMQX / AWS IoT Core / Azure IoT Hub など）のトピックをsubscribeしてメッセージをインメモリキャッシュに保持し、 タグのポーリングサイクルごとにキャッシュ値を返します。writeは即時publish。

ソース: plantpulse.driver.protocol.mqtt_client.*

クラス責務
MQTTClientDriverProtocolDriver 実装 — connect/read/write/close、トピックキャッシュ、lazy subscribe

ライブラリ: lib/org.eclipse.paho.client.mqttv3-1.2.5.jar

Sparkplug B との違い​

PlantPulse は2種類のMQTTベースドライバを別々に提供します。

ドライバpayload使用場面
MQTT_CLIENT（このページ）任意のString/JSON/binary一般的なIIoTブローカー、ユーザー定義トピック、ASCII / JSON payload
SPARKPLUG_BSparkplug B Payload（Protobuf）Cirrus Link / Tahu / Ignition 標準、NBIRTH/DBIRTH alias

Sparkplug B は独自のワイヤーコーデックとalias学習ロジックが含まれているため、一般的なMQTTのみで十分な環境では このドライバの方が軽量で、標準のPaho APIだけで十分です。

動作フロー​

broker ──PUBLISH──> Paho MqttClient ──MqttCallback.messageArrived(topic, payload)──> Driver
                                                                                       │
                                                                                       └─ lastValueByTopic.put(topic, new String(payload))

PlantPulse 폴러 ──read(addr)──> lastValueByTopic[addr]   (in-memory hit)
                  write(addr, value) ──Paho.publish(topic, value, qos, retain=false)──> broker

readの最初の呼び出しでキャッシュミスの場合、lazy subscribe（QoSオプション値）後に空文字列を返します — 次のサイクルから 実際の値が入ります。 tag_map に登録されたトピックは、接続直後に preSubscribeFromTagMap() で 一括subscribeされ、最初のサイクルで即座に値が届く場合もあります（メッセージのpublish頻度によります）。

オプション​

キーデフォルト値説明
username（なし）ブローカー認証ユーザー名
password（なし）ブローカー認証パスワード
tlsfalsetrue → ssl://（デフォルト8883）、false → tcp://（デフォルト1883）
qos0subscribe / publish QoS（0 / 1 / 2）。範囲外の値は自動的にクランプ。
keep-alive60keep-alive周期（秒） — MqttConnectOptions.setKeepAliveInterval
clean-sessiontrueMqttConnectOptions.setCleanSession
client-id（自動）明示的なclient id。未指定時は PP-<opc_id>-<random6>（≤23文字）を自動生成

MqttConnectOptions.setAutomaticReconnect(true) が常に有効で、一時的なブローカー切断後に 自動再接続されます（Pahoのデフォルトbackoffを適用）。

Addressマッピング​

address動作
factory/line1/temp単一トピック — このトピックの最後のメッセージ
device/+/statuswildcard（1段階） — マッチするトピックの最後のメッセージ（最も最近到着）
factory/#wildcard（マルチ） — 最後のセグメント、すべてのサブトピック
"" / nullemptyを返す

wildcard注意: Pahoのcallbackは、実際に到着したトピックごとに別エントリをキャッシュせず、 到着したトピック自体がキーになります。つまり device/+/status でsubscribeしても lastValueByTopic には device/sensor01/status、device/sensor02/status が個別に入り、 ユーザーがwildcardトピック自体でreadするとキャッシュにはエントリがありません（wildcard自体は publishの宛先ではないため）。wildcardパターンを登録する場合は、device/+/status のように登録した後、 readの結果として最後に到着したトピックの値を別途追跡する必要があります — 一般的な用途ではトピックごとに 明示的登録を推奨します。

write（publish）​

client.publish(topic, value.getBytes(), qos, /*retain=*/ false);
  • value のバイトがそのままpayloadとして送信されます（UTF-8デフォルトcharset）。
  • retain=false で固定されており、変更できません — ブローカーが最後のメッセージを保持しない ため、後から購読を開始したクライアントは次の書き込みが発生するまで値を受け取れません。 retainedメッセージが必要なコンシューマには、ブローカー側のブリッジや別のpublisherを設置してください。
  • QoSはオプション値のまま使用。QoS 1/2 の場合、Pahoが内部的にackまで再送信します。

接続ライフサイクル​

メソッド動作
connect()options適用 → broker URL組み立て → MqttClient.connect() → tag_map 一括subscribe
read()キャッシュミス時にlazy subscribe → キャッシュヒットなら値を返す
write()publish（retain=false）
close()disconnect() + close()（best-effort、例外無視） → キャッシュクリア
connectionLost(cause)callbackで setConnected(false) — Pahoが自動再接続を試行

制限事項および今後の作業​

  • mTLS / クライアント証明書 — 現在はusername/passwordのみ。AWS IoT のようなX.509 mTLS環境は別途 truststore / keystore設定が必要（Paho MqttConnectOptions.setSocketFactory の公開は未実装）。
  • retained / willメッセージ — 現在 retain=false 固定、LWT未設定。
  • wildcardキャッシュ分離 — wildcardパターン自体でreadすると結果が空（到着トピックごとのエントリ）。
  • JSONパス抽出 — payloadがJSONでもドライバはraw Stringのみ返します。JSONPath抽出は format / 数式段階で処理が必要（または別途パーサーノード）。
  • payloadエンコーディング — 現在はプラットフォームデフォルトcharsetを使用。binary payloadはUTF-8で壊れる可能性あり → 将来的に payload-encoding オプション（utf8 / iso-8859-1 / base64）の公開を検討。

テスト場所​

test/java/plantpulse/driver/protocol/mqtt_client/MQTTClientDriverTest.java — 31テスト。

  • 初期状態 / メタ（isConnected/isWriteSupported/isExternal/getDriverSource）
  • 未接続時のread/write/null/empty安全性
  • getDebugBaseUrl — tcp/ssl、明示ポート、デフォルトポート（1883/8883）
  • オプションパース — username/password/tls/keep-alive/clean-session/qos/client-id、不正値→デフォルト
  • 静的ヘルパー — clampQos、defaultClientId（長さ/random/null）、nullIfEmpty、parseInt、parseBool
  • 定数検証（DEFAULT_PORT、DEFAULT_PORT_TLS、DEFAULT_KEEP_ALIVE_SEC、DEFAULT_QOS）

実際のブローカー依存の統合テストは別途 — 本単体テストは外部ブローカーなしで動作します。