Non-Blocking MQTT Client für Reactive Embedded Systems bauen

MQTT-Reactive ist ein Lightweight, Non-Blocking MQTT v3.1.1 Client built on LiamBindles MQTT-C Library. Designed für Reactive Embedded Systems, liefert es einen portable, Event-Driven Communication Stack, der in jede C-basierte Applikation integriert werden kann.

In diesem Artikel definieren wir zuerst, was ein Reactive System konstituiert, dann skizzieren wir eine Software Architektur, die Concurrency und State-Based Behavior embraced. Schließlich walk through wir ein konkretes Beispiel – ein IoT Device für eine Railway Company – und zeigen, wie MQTT-Reactive via State Machine und Event-Driven Paradigm employed wird.

Reactive Systems respond zu Internal oder External Events, process das Event, und return zu Idle State awaiting den Next Trigger. Dieses Event-Driven Paradigm yields predictable, maintainable Code, besonders wenn Multiple Active Objects (Concurrent Units) distinct Event Streams handlen.

Unser Case Study Device performte folgende Tasks:

1. Detektierte und loggte Changes in mehreren Digital Inputs
2. Akquirierte, filterte und speicherte Multiple Analog Signals
3. Transmittierte gespeicherte Data periodisch zu Remote Server
4. Exchanged Data mit Broker via MQTT über GSM

MQTT wurde gewählt für sein Lightweight Publish/Subscribe Model, ideal für Low-Bandwidth, High-Latency Links wie GSM.

Weil das Device Firmware als Reactive System built war, required die Original MQTT-C Library Adaptation, um asynchron via Events zu kommunizieren. Die resultierende Library – MQTT-Reactive – exponiert ein Clean API, das von einer State Machine invoked werden kann, während der Rest des Systems decoupled bleibt.

State Machine

MQTT-Reactive wird getrieben vom MqttMgr Active Object, dessen Behavior durch die State Machine in Abbildung 1 beschrieben wird. Die Machine startet in WaitingForNetConnection; einmal Network Link verfügbar, transitioned sie zu WaitingForSync, wo CONNECT und PUBLISH Packets staged werden können. Events, die in Sync ankommen, werden deferred, bis das System einen State re-entert, der sie akzeptiert.

Alle SyncTime Millisekunden moved die Machine zum Sync Composite State, der Receive und Send Events an den Network Manager postet. Zwar deckt das Beispiel CONNECT und PUBLISH ab, aber Extension zu SUBSCRIBE ist straightforward.

Die State Machine Actions nutzen params Keyword zum Lesen von Event Data und GEN() Macro zum Generieren Outgoing Events. Beispiel:

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);

In diesem Transition trägt das Connect Event clientId und keepAlive Parameter, die auf MqttMgr Object Attributes angewendet werden. Ebenso werden Events an andere Actors dispatched via:

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

Die Default Action setzt einen Callback, den MQTT-Reactive invokt, wenn Broker ein CONNECT acknowledged. Der Callback muss entweder ConnAccepted oder ConnRefused Events generieren:

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)); } }

Abbildung 1. State Machine eines MQTT-Reactive Clients (Quelle: VortexMakes)

Model Implementation

Die State Machine kann in Plain C oder C++ coded werden using any of: RKH Framework, QP Framework, Yakindu Statechart Tool, oder Rational Rhapsody Developer. Alle unterstützen Statecharts und Code Generation für C/C++.

Das MqttMgr Active Object encapsuliert alle MQTT-Reactive Calls, shielding den Rest des Systems von Direct API Usage. Concurrency wird achieved durch Exchanging Events, eliminating Pitfalls von Semaphores, Mutexes, Delays oder Event-Flags.

Attribute von MqttMgr werden direkt in seiner Structure stored, z. B.:

struct MqttMgr { struct mqtt_client client; // MQTT Client Instance LocalRecvAll localRecv; // Local Receive Buffer /* additional data items */ };

Diese Members werden via me Pointer accessed:

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

Tabelle 1. MqttMgr Attributes

Das Overall System umfasst drei Key Actors: Collector (published Data), NetMgr (manages Network), und MqttMgr (manages MQTT).

Abbildung 2. Draft of IoT System Structure (Quelle: VortexMakes)

Das Sequence Diagram in Abbildung 3 illustriert den Flow, wenn MqttMgr eine Session mit MQTT Broker initiiert.

Abbildung 3. Connecting to an MQTT Broker (Quelle: VortexMakes)

Upon Successful Network Establishment, Collector sendet Connect(clientId, keepAlive) zu MqttMgr. Broker may reply mit ConnAccepted oder ConnRefused, letztere carrying Rejection Code per MQTT v3.1.1 §3.2.2.3.

Abbildung 4. Broker rejected eine Connection Request (Quelle: VortexMakes)

Publishing folgt ähnlichem Pattern. Collector issued Publish(data, size, topic, qos) Events. Data Payload, typisch JSON, assembled using Lightweight jWrite Library.

Abbildung 5. Publishing Data to a Broker (Quelle: VortexMakes)

Network Failures werden signalisiert durch NetMgr via ReceiveFail oder SendFail Events, enabling MqttMgr to react accordingly.

Abbildung 6. Failures in Network (Quelle: VortexMakes)

Tabelle 2. Events involved in den Scenarios

Fazit

Durch Encapsulierung von MQTT-Reactive innerhalb eines Active Object und Orchestrierung Communication via Events eliminiert der Ansatz Blocking Constructs und yields ein Clean, Testable Design. Das State-Machine Model skaliert zu Multiple Concurrent Tasks, making es Well-Suited für moderne Reactive Embedded Applications, die Reliable MQTT Connectivity benötigen.