Een non-blocking MQTT-client bouwen voor reactieve embedded-systemen

MQTT-Reactive is een lichtgewicht, non-blocking MQTT v3.1.1-client gebouwd op LiamBindle's MQTT-C-bibliotheek. Ontworpen voor reactieve embedded-systemen, levert het een portable, event-gedreven communicatiestack die geïntegreerd kan worden in elke C-gebaseerde applicatie.

In dit artikel definiëren we eerst wat een reactief systeem constitutieert, schetsen we dan een software-architectuur die concurrentie en state-gebaseerd gedrag omarmt. Ten slotte lopen we door een concreet voorbeeld - een IoT-apparaat voor een spoorwegbedrijf - wat laat zien hoe MQTT-Reactive via een state-machine en event-gedreven paradigma te gebruiken.

Reactieve systemen reageren op interne of externe events, verwerken het event, en keeren terug naar een idle-state in afwachting van de volgende trigger. Dit event-gedreven paradigma levert voorspelbare, onderhoudbare code, vooral wanneer meerdere actieve objecten (concurrente eenheden) afzonderlijke event-stromen afhandelen.

Onze case-study-apparaat voerde de volgende taken uit:

1. Wijzigingen in diverse digitale inputs detecteren en loggen
2. Meerdere analoge signalen acquireren, filteren en opslaan
3. Opgeslagen data periodiek naar een remote server transmitteren
4. Data uitwisselen met een broker via MQTT over GSM

MQTT werd gekozen voor zijn lichtgewicht, publish/subscribe-model, ideaal voor lage-bandbreedte, hoge-latentie links zoals GSM.

Omdat de firmware van het apparaat als een reactief systeem gebouwd was, vereiste de originele MQTT-C-bibliotheek aanpassing om asynchroon via events te communiceren. De resulterende bibliotheek - MQTT-Reactive - blootlegt een schone API die vanuit een state-machine aangeroepen kan worden terwijl de rest van het systeem gedecoupled blijft.

State machine

MQTT-Reactive wordt gedreven door het MqttMgr actieve object, wiens gedrag beschreven wordt door de state-machine in Figuur 1. De machine start in WaitingForNetConnection; zodra een netwerklink beschikbaar is, gaat het over naar WaitingForSync, waar CONNECT- en PUBLISH-packets gestaged kunnen worden. Events die in Sync aankomen, worden uitgesteld totdat het systeem een state herbetreedt die ze accepteert.

Elke SyncTime milliseconden beweegt de machine naar de Sync composite-state, die Receive en Send events naar de netwerkmanager post. Hoewel het voorbeeld CONNECT en PUBLISH dekt, is het uitbreiden van ondersteuning naar SUBSCRIBE eenvoudig.

De state-machine's acties gebruiken het params keyword om event-data te lezen en de GEN() macro om uitgaande events te genereren. Bijvoorbeeld:

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 deze transitie draagt het Connect-event clientId en keepAlive parameters die toegepast worden op het MqttMgr-object's attributen. Evenzo worden events verstuurd naar andere actoren via:

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

De standaard-actie stelt een callback in die MQTT-Reactive aanroept wanneer de broker een CONNECT bevestigt. De callback moet ofwel ConnAccepted ofwel ConnRefused events genereren:

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

Figuur 1. State machine van een MQTT-Reactive client (Bron: VortexMakes)

Model-implementatie

De state-machine kan in plain C of C++ gecodeerd worden met een van de volgende tools: RKH-framework, QP-framework, Yakindu Statechart Tool, of Rational Rhapsody Developer. Allemaal ondersteunen Statecharts en code-generatie voor C/C++.

Het MqttMgr actieve object inkapselt alle MQTT-Reactive-aanroepen, wat de rest van het systeem beschermt tegen direct API-gebruik. Concurrentie wordt bereikt door events uit te wisselen, wat de vallen van semaphores, mutexes, vertragingen of event-flags elimineert.

Attributen van MqttMgr worden direct in zijn structuur opgeslagen, bijv.:

struct MqttMgr { struct mqtt_client client; // MQTT client instance LocalRecvAll localRecv; // lokale ontvangstbuffer /* additionele data-items */ };

Deze leden worden benaderd via de me pointer:

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

Tabel 1. MqttMgr-attributen

Het overall-systeem omvat drie sleutelactoren: Collector (publiceert data), NetMgr (beheert het netwerk) en MqttMgr (beheert MQTT).

Figuur 2. Schets van IoT-systeemstructuur (Bron: VortexMakes)

Het sequentiediagram in Figuur 3 illustreert de flow wanneer MqttMgr een sessie initieert met een MQTT-broker.

Figuur 3. Verbinden met een MQTT-broker (Bron: VortexMakes)

Bij succesvolle netwerkvestiging stuurt Collector Connect(clientId, keepAlive) naar MqttMgr. De broker kan antwoorden met ConnAccepted of ConnRefused, de laatste dragend een rejectiecode volgens MQTT v3.1.1 §3.2.2.3.

Figuur 4. Broker weigert een verbindingsverzoek (Bron: VortexMakes)

Publiceren volgt een vergelijkbaar patroon. Collector geeft Publish(data, size, topic, qos) events uit. De data-payload, typisch JSON, wordt samengesteld met de lichtgewicht jWrite-bibliotheek.

Figuur 5. Data publiceren naar een broker (Bron: VortexMakes)

Netwerkfouten worden gesignaleerd door NetMgr via ReceiveFail of SendFail events, wat MqttMgr in staat stelt om dienovereenkomstig te reageren.

Figuur 6. Fouten in netwerk (Bron: VortexMakes)

Tabel 2. Events betrokken in de scenario's

Conclusie

Door MQTT-Reactive in te kapselen in een actief object en communicatie te orchestratie via events, elimineert de benadering blokkerende constructies en levert een schone, testbare ontwerp. Het state-machine-model schaalt naar meerdere gelijktijdige taken, wat het goed geschikt maakt voor moderne reactieve embedded-toepassingen die betrouwbare MQTT-connectiviteit vereisen.