リアクティブ組み込みシステム向けノンブロッキングMQTTクライアントの構築

MQTT-Reactiveは、LiamBindleのMQTT-Cライブラリ上に構築された軽量・ノンブロッキングのMQTT v3.1.1クライアントです。リアクティブ組み込みシステム向けに設計され、あらゆるCベースアプリケーションに統合可能なポータブルでイベント駆動の通信スタックを提供します。

この記事ではまず、リアクティブシステムを構成する要素を定義し、並行性と状態ベース振る舞いを採用したソフトウェアアーキテクチャを概説します。最後に、鉄道会社向けIoTデバイスという具体例を通じて、ステートマシンとイベント駆動パラダイムでMQTT-Reactiveをどう活用するかを解説します。

リアクティブシステムは、内部または外部イベントに応答し、イベントを処理し、次のトリガーを待つアイドル状態に戻ります。このイベント駆動パラダイムは、特に複数のアクティブオブジェクト（並行ユニット）が異なるイベントストリームを処理する場合に、予測可能で保守しやすいコードを生み出します。

ケーススタディデバイスは以下のタスクを実行しました：

1. 複数のデジタル入力の変化を検出・記録
2. 複数のアナログ信号を取得・フィルタリング・保存
3. 保存データを定期的にリモートサーバーへ送信
4. GSM経由でMQTTを使用してブローカーとデータ交換

MQTTは、GSMのような低帯域幅・高レイテンシリンクに最適な軽量なPub/Subモデルのため選択されました。

デバイスのファームウェアがリアクティブシステムとして構築されていたため、元のMQTT-Cライブラリは非同期にイベント経由で通信するよう適応が必要でした。その結果生まれたライブラリMQTT-Reactiveは、ステートマシンから呼び出せるクリーンなAPIを公開し、システムの他部分を疎結合に保ちます。

ステートマシン

MQTT-Reactiveは、図1のステートマシンで振る舞いが記述されるMqttMgrアクティブオブジェクトによって駆動されます。マシンはWaitingForNetConnectionで開始し、ネットワークリンクが利用可能になるとWaitingForSyncへ遷移し、ここでCONNECTとPUBLISHパケットをステージングできます。Syncで到着したイベントは、システムがイベントを受け入れる状態に再入るまで延期されます。

毎SyncTimeミリ秒ごとにマシンはSyncコンポジット状態へ移動し、ReceiveとSendイベントをネットワークマネージャーへポストします。例ではCONNECTとPUBLISHをカバーしていますが、SUBSCRIBEサポートへの拡張も容易です。

ステートマシンのアクションは、paramsキーワードでイベントデータを読み取り、GEN()マクロで出力イベントを生成します。例：

Connect(clientId, keepAlive)/ me->clientId = params->clientId; me->keepAlive = params->keepAlive; me->operRes = mqtt_connect(&me->client, me->clientId, NULL, NULL, 0, NULL, NULL, 0, me->keepAlive);

この遷移では、ConnectイベントがclientIdとkeepAliveパラメータを運び、それらがMqttMgrオブジェクトの属性に適用されます。同様に、イベントは他のアクターへディスパッチされます：

GEN(me->itsCollector, Receive());

デフォルトアクションは、ブローカーがCONNECTを確認したときにMQTT-Reactiveが呼び出すコールバックを設定します。コールバックはConnAcceptedまたはConnRefusedイベントを生成しなければなりません：

static void connack_response_callback(enum MQTTConnackReturnCode return_code) { if (return_code == MQTT_CONNACK_ACCEPTED) { GEN(me->itsCollector, ConnAccepted()); } else { GEN(me->itsCollector, ConnRefused(return_code)); } }

図1. MQTT-Reactiveクライアントのステートマシン（出典：VortexMakes）

モデル実装

ステートマシンは、以下のツールのいずれかを用いてプレーンなCまたはC++でコーディング可能です：RKHフレームワーク、QPフレームワーク、Yakindu Statechart Tool、Rational Rhapsody Developer。すべてStatechartsとC/C++コード生成をサポートします。

MqttMgrアクティブオブジェクトはすべてのMQTT-Reactive呼び出しをカプセル化し、システムの他部分を直接API使用から遮蔽します。並行性はイベント交換によって達成され、セマフォ、ミューテックス、遅延、イベントフラグの落とし穴を排除します。

MqttMgrの属性はその構造体に直接格納されます、例：

struct MqttMgr { struct mqtt_client client; // MQTTクライアントインスタンス LocalRecvAll localRecv; // ローカル受信バッファ /* 追加データ項目 */ };

これらのメンバはmeポインタ経由でアクセスされます：

mqtt_recvMsgError(&me->client, &me->localRecv);

表1. MqttMgr属性

全体システムには3つの主要アクターが含まれます：Collector（データ発行）、NetMgr（ネットワーク管理）、MqttMgr（MQTT管理）。

図2. IoTシステム構造のドラフト（出典：VortexMakes）

図3のシーケンス図は、MqttMgrがMQTTブローカーとセッションを開始する際のフローを示します。

図3. MQTTブローカーへの接続（出典：VortexMakes）

ネットワーク確立成功時、CollectorがConnect(clientId, keepAlive)をMqttMgrへ送信。ブローカーはConnAcceptedまたはConnRefusedで応答し、後者はMQTT v3.1.1 §3.2.2.3に準拠した拒否コードを携えます。

図4. ブローカーが接続要求を拒否（出典：VortexMakes）

パブリッシュも同様のパターンです。CollectorがPublish(data, size, topic, qos)イベントを発行。データペイロード（通常JSON）は軽量jWriteライブラリで組み立てられます。

図5. ブローカーへのデータ発行（出典：VortexMakes）

ネットワーク障害はNetMgrからReceiveFailまたはSendFailイベントで通知され、MqttMgrが適切に反応できるようにします。

図6. ネットワーク障害（出典：VortexMakes）

表2. シナリオに関わるイベント

結論

MQTT-Reactiveをアクティブオブジェクト内にカプセル化し、通信をイベントでオーケストレーションすることで、ブロッキング構造を排除し、クリーンでテスト可能な設計を実現します。ステートマシンモデルは複数の並行タスクにスケール可能で、信頼性の高いMQTT接続を必要とする最新のリアクティブ組み込みアプリケーションに適しています。