Een MQTT-client implementeren voor reactieve systemen

MQTT-Reactive is een MQTT v3.1.1-client afgeleid van LiamBindle's MQTT-C-bibliotheek. Het doel van MQTT-Reactive is om een ​​draagbare en niet-blokkerende MQTT-client te bieden die is geschreven in C om te worden gebruikt in reactieve embedded systemen. Allereerst wordt in dit artikel uitgelegd wat een reactief systeem is. Vervolgens wordt beschreven hoe een geschikte softwarestructuur voor dat soort systeem kan worden ontworpen. Ten slotte laat het artikel zien hoe de MQTT-reactieve bibliotheek in een reactief systeem kan worden gebruikt met behulp van een toestandsmachine en het gebeurtenisgestuurde paradigma. Om dat te doen, gebruikt het artikel een echt IoT-apparaat als een demonstratief voorbeeld van waaruit het artikel de softwarestructuur en het op toestanden gebaseerde gedrag uitlegt met behulp van UML-diagrammen zoals toestandsmachine, interactie en structuur. Het artikel geeft ook richtlijnen om de MQTT-Reactive client van het IoT-apparaat in C-taal te implementeren.

Veel embedded systemen zijn reactief, d.w.z. ze reageren op interne of externe gebeurtenissen. Zodra deze reacties zijn voltooid, gaat de software terug om te wachten op de volgende gebeurtenis. Daarom worden gebeurtenisgestuurde systemen ook wel reactieve systemen genoemd.

Gebeurtenisgestuurd programmeren, of reactief programmeren, is een van de meest geschikte programmeerparadigma's om flexibele, voorspelbare en onderhoudbare software voor reactieve systemen te realiseren. In dit paradigma wordt de stroom van het programma bepaald door gebeurtenissen. Vaak bestaat de structuur van de reactieve software uit verschillende gelijktijdige eenheden, ook wel actieve objecten genoemd, die wachten en verschillende soorten gebeurtenissen verwerken. Elk actief object heeft een controledraad en een gebeurteniswachtrij waarmee het zijn binnenkomende gebeurtenissen verwerkt. In reactieve systemen hebben de actieve objecten typisch statusgebaseerd gedrag gedefinieerd in een statusdiagram.

Om te onderzoeken hoe de MQTT-reactieve bibliotheek kan worden gebruikt in een reactief systeem met meerdere en gelijktijdige taken en met zowel een statusmachine als het gebeurtenisgestuurde paradigma, gebruiken we een IoT-apparaat als voorbeeld.

Het idee om het MQTT-protocol te gebruiken werd geboren terwijl een IoT-apparaat werd ontwikkeld voor een spoorwegmaatschappij. Dit apparaat was een duidelijk reactief systeem dat in staat was om:

1. detecteer en bewaar wijzigingen van verschillende digitale ingangen
2. verkrijg, filter en sla meerdere analoge signalen op
3. Stuur regelmatig opgeslagen informatie naar een externe server
4. informatie verzenden en ontvangen via het MQTT-protocol via het GSM-netwerk

MQTT werd gekozen omdat het een lichtgewicht op uitgever-abonnee gebaseerd berichtenprotocol is dat vaak wordt gebruikt in IoT- en netwerktoepassingen waar verbindingen met hoge latentie en lage gegevenssnelheid worden verwacht, zoals de GSM-netwerken.

De MQTT-mogelijkheid voor het genoemde IoT-apparaat werd bereikt door een aangepaste versie van LiamBindle's MQTT-C te gebruiken. Omdat de software van dat apparaat was ontworpen als reactieve software, moest MQTT-C worden aangepast om het met de rest van het systeem te communiceren door asynchrone gebeurtenissen uit te wisselen. Deze gebeurtenissen werden gebruikt voor het ontvangen en verzenden van verkeer over het netwerk en voor het verbinden en publiceren van gevoelige informatie naar een server. De resulterende softwarebibliotheek heette MQTT-Reactive.

Statusmachine

MQTT-Reactive werd gebruikt via een statusmachine zoals weergegeven in figuur 1, die het basisgedrag van een MQTT-Reactive-client modelleert. Het was een actief object genaamd MqttMgr (MQTT Manager). De acties van de statusmachine in figuur 1 laten zien hoe de MQTT-reactieve bibliotheek zou kunnen worden gebruikt vanaf een statusmachine. Hoewel de C-taal werd gebruikt als de actietaal in figuur 1, kan elke computer of formele taal worden gebruikt.

klik voor grotere afbeelding

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

De statusmachine in afbeelding 1 begint in de status WaitingForNetConnection. Nadat een netwerkverbinding met een server tot stand is gebracht, ontvangt de WaitingForNetConnection de gebeurtenis Activate en gaat de statusmachine over naar de status WaitingForSync. Alleen in deze status kunnen de MQTT-berichten van de statusmachine naar de broker worden verzonden, zoals CONNECT of PUBLISH via respectievelijk de Connect- en Publish-gebeurtenissen. De Sync-status gebruikt het speciale mechanisme van een UML voor het uitstellen van de Publish-gebeurtenis die wordt gespecificeerd door het defer-sleutelwoord dat is opgenomen in het interne compartiment van de Sync-status. Als de publicatie-gebeurtenis plaatsvindt terwijl Sync de huidige status heeft, wordt deze opgeslagen (uitgesteld) voor toekomstige verwerking totdat de SM in een staat komt waarin de publicatie-gebeurtenis niet in de lijst met uitgestelde gebeurtenissen staat, zoals WaitingForSync of WaitingForNetConnection. Bij het betreden van dergelijke statussen, zal de statusmachine automatisch alle opgeslagen publicatiegebeurtenissen oproepen en deze gebeurtenis vervolgens consumeren of verwijderen volgens de doelstatus van de overgang.

Elke SyncTime milliseconden gaat de statusmachine over naar de Sync-composietstatus, die het daadwerkelijke verzenden en ontvangen van verkeer van het netwerk doet door ontvangst- en verzendgebeurtenissen naar de netwerkbeheerder te posten. Het is een gelijktijdige entiteit die zich bezighoudt met netwerkproblemen.

Hoewel het geïntroduceerde MqttMgr alleen de CONNECT- en PUBLISH-pakketten ondersteunt, zou het het SUBSCRIBE-pakket kunnen ondersteunen met vrij eenvoudige wijzigingen.

De statusmachine-acties hebben toegang tot de parameters van de verbruikte gebeurtenis met behulp van het params-sleutelwoord. In de volgende overgang heeft de gebeurtenis Connect bijvoorbeeld twee parameters, clientId en keepAlive, waarvan de waarden worden gebruikt om de kenmerken van het overeenkomstige MqttMgr-object bij te werken:

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 dit voorbeeld is de gebeurtenis Connect(clientId, keepAlive) de trigger van de overgang en maakt de aanroep mqtt_connect() deel uit van de actie die als resultaat wordt uitgevoerd. Met andere woorden, wanneer het MqttMgr-object een Connect(clientId, keepAlive) -gebeurtenis ontvangt met de parameters 'publishing_client' en '400', Connect("publishing_client", 400), worden de clientId- en keepAlive-attributen van MqttMgr bijgewerkt met de waarden ' publishing_client' en '400' bijgevolg.

Om gebeurtenissen te maken en te verzenden, gebruiken de acties van de statusmachine de GEN()-macro. De volgende instructie verzendt bijvoorbeeld een Receive-gebeurtenis naar het Collector-object, waarnaar wordt verwezen als het kenmerk van een MqttMgr-object  door de Collector-aanwijzer:

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

Het eerste argument van de GEN()-instructie is het object dat de gebeurtenis ontvangt, terwijl het tweede argument de gebeurtenis is die wordt verzonden, inclusief gebeurtenisargumenten (als die er zijn). De argumenten moeten overeenkomen met de gebeurtenisparameters. De volgende instructie genereert bijvoorbeeld een ConnRefused(code)-gebeurtenis en verzendt deze naar het Collector-object en geeft de code door die door de broker is geretourneerd als een gebeurtenisparameter:

GEN(me->itsCollector, ConRefused(code));

Het idee om het params-sleutelwoord te gebruiken om toegang te krijgen tot de parameters van de verbruikte gebeurtenis en de GEN()-macro om gebeurtenissen uit acties te genereren, werd voor louter illustratieve doeleinden overgenomen van de codegenerator van Rational Rhapsody Developer.

De standaardactie van de statusmachine in afbeelding 1 stelt de callback in die wordt aangeroepen door MQTT-Reactive wanneer een verbindingsacceptatie wordt ontvangen van de makelaar. Deze callback moet worden geïmplementeerd in MqttMgr-code. Deze callback moet ConnAccepted of ConnRefused(code)-gebeurtenissen genereren om naar het Collector-object te worden verzonden, zoals hieronder wordt weergegeven.

statisch  nietig connack_response_callback (enum MQTTConnackReturnCode return_code){    /*...*/    if (return_code ==MQTT_CONNACK_ACCEPTED)    {        GEN(me->itsCollector, ConnAccepted()); }    anders {        GEN(me->itsCollector, ConnRefused(return_code)); }}

Modelimplementatie

Het model in figuur 1 kan worden geïmplementeerd in C of C++ met behulp van uw favoriete softwaretool of alleen uw eigen implementatie van een state-machine. Er zijn verschillende tools op internet beschikbaar om dat te doen, zoals onder andere RKH-framework, QP-framework, Yakindu Statechart Tool of Rational Rhapsody Developer. Ze ondersteunen allemaal Statecharts- en C/C++-talen. Sommige bevatten bovendien een tool om een ​​Statechart-diagram te tekenen en daaruit code te genereren.

Deze toestandsmachine werd uitgevoerd vanuit een actief object genaamd MqttMgr (MQTT Manager), dat zorgde voor strikte inkapseling van de MQTT-reactieve code en het was de enige entiteit die een MQTT-reactieve functie mocht aanroepen of toegang mocht krijgen tot MQTT-reactieve gegevens. De andere gelijktijdige entiteiten in het systeem en eventuele ISR's konden MQTT-Reactive alleen indirect gebruiken door gebeurtenissen uit te wisselen met MqttMgr. Het gebruik van dit mechanisme om gelijktijdige entiteiten te synchroniseren en om gegevens onderling te delen, vermijdt het omgaan met de gevaren van traditionele blokkeringsmechanismen zoals semaforen, mutex, vertragingen of gebeurtenismarkeringen. Deze mechanismen kunnen onverwachte storingen veroorzaken die moeilijk en vervelend zijn om te diagnosticeren en op te lossen.

Het actieve object MqttMgr kapselt zijn attributen in als een set gegevensitems. Een gegevensitem duidt een variabele aan met een naam en een type, waarbij het type eigenlijk een gegevenstype is. Een gegevensitem voor het MqttMgr-object wordt toegewezen aan een lid van de objectstructuur. De naam en het type van het lid zijn dezelfde als die van de gegevens van het object. Het clientkenmerk van het objecttype MqttMgr is bijvoorbeeld ingesloten door waarde als een gegevenslid in de MqttMgr-structuur:

struct MqttMgr {    /* ... */ struct mqtt_client klant; /* kenmerk cliënt */ LocalRecvAlle localRecv; /* kenmerk localRecv */ };

De gegevens van het MqttMgr-object worden rechtstreeks geopend en gewijzigd zonder gebruik te maken van accessor- of mutatorbewerkingen. Client en localRecv zijn bijvoorbeeld toegankelijk via de me-aanwijzer, die verwijst naar een instantie van MqttMgr.

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

De MqttMgr heeft de lijst met attributen getoond in Tabel 1.

Tabel 1. MqttMgr-kenmerken

De structuur in figuur 2 is nuttig om de relaties tussen de betrokken actoren in het oog te houden. Dit zijn:het Collector-object, dat informatie naar de makelaar wil sturen; het object NetMgr, dat zich bezighoudt met het netwerk; en het MqttMgr-object.

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

Het sequentiediagram in figuur 3 laat zien hoe het MqttMgr-object interageert met de rest van het systeem wanneer het nodig is om een ​​sessie met een MQTT-server te openen. In dit diagram worden de MqttMgr-status en de uitgewisselde asynchrone berichten weergegeven tussen Collector-, MqttMgr- en NetMgr-actoren.

Figuur 3. Verbinding maken met een MQTT-makelaar (Bron:VortexMakes)

Nadat een netwerkverbinding tot stand is gebracht door het NetMgr-object met een broker, moet het eerste pakket dat van MqttMgr naar een MQTT-server wordt verzonden, een CONNECT-pakket zijn. De Collector-acteur verzendt dus een Connect(clientId, keepAlive) -gebeurtenis naar de MqttMgr-acteur. Deze gebeurtenis moet de client-ID en het keep-alive-tijdsinterval bevatten. Als de server het verbindingsverzoek accepteert, stuurt de MqttMgr-acteur een ConnAccepted-gebeurtenis naar de Collector-acteur om deze situatie te melden. Vanaf dat moment kan de Collector-acteur informatieberichten naar die makelaar publiceren.

Als de server het verbindingsverzoek afwijst, stuurt de MqttMgr-acteur een ConnRefused-gebeurtenis naar de Collector-acteur. Deze gebeurtenis brengt een code met zich mee die de oorzaak van de afwijzing aangeeft, zoals weergegeven in Afbeelding 4. Zie MQTT v3.1.1-sectie 3.2.2.3.

Figuur 4. Broker wijst een verbindingsverzoek af (Bron:VortexMakes)

Figuur 5 toont de interactiestroom wanneer een bericht wordt gepubliceerd. Om dat te doen, verzendt de Collector-acteur een Publish (data, size, topic, qos) -gebeurtenis, die de te publiceren informatie bevat (data), de lengte van de informatie in bytes (grootte), de onderwerpnaam waarnaar de informatie wordt gepubliceerd (onderwerp) en het niveau van zekerheid om dit bericht te leveren (qos). In het eerder genoemde IoT-apparaat was de gepubliceerde informatie geformatteerd met behulp van de JSON-specificatie. Het is een open standaardindeling die gegevensobjecten bevat met attribuut-waardeparen in voor mensen leesbare tekst. Dit formaat werd bereikt met jWrite, een eenvoudige en lichtgewicht bibliotheek geschreven in C.

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

Figuur 6 toont een scenario waarin de ontvangst en verzending van MQTT-berichten naar het netwerk mislukt. Als de netwerkbeheerder geen verkeer van het netwerk kan ontvangen, stuurt hij een ReceiveFail naar MqttMgr-acteur. Evenzo, als de netwerkbeheerder geen gegevens naar het netwerk kan verzenden, stuurt deze een SendFail naar MqttMgr-acteur.

Figuur 6. Storingen in netwerk (Bron:VortexMakes)

Tabel 2 geeft een overzicht van de betrokken gebeurtenissen in de getoonde scenario's.

Tabel 2. Evenementen

Conclusie

Door de gevaren van traditionele blokkeringsmechanismen te vermijden - zoals semaforen, mutex, vertragingen of gebeurtenisvlaggen - laten de MQTT-reactieve bibliotheek, de statusmachine en de software-architectuur die in dit artikel worden voorgesteld, reactieve embedded systemen toe om een ​​MQTT-client in een roman te implementeren manier. Dit wordt bereikt door MQTT-reactieve code in te kapselen in een gelijktijdigheidseenheid genaamd actief object , waarvan het op toestanden gebaseerde gedrag is gedefinieerd in de voorgestelde toestandsmachine. Dit actieve object communiceert met de rest van het systeem door asynchrone gebeurtenissen uit te wisselen die worden gebruikt:niet alleen voor het ontvangen en verzenden van verkeer via het netwerk, maar ook voor het verbinden en publiceren van informatie naar een server voor Internet of Things-toepassingen.