Entwicklung eines MQTT-Clients für Metatrader 5: ein TDD-Ansatz — Teil 6

 

„Optimismus ist ein Berufsrisiko beim Programmieren; Feedback ist die Behandlung“. (Kent Beck)

Einführung

Die Methodik der testgetriebenen Entwicklung bietet viele Vorteile und hat einen großen Nachteil. Zu den Vorteilen gehört, dass wir gut definierte Einheiten und gut benannte Variablen schreiben können, um eine hohe Testabdeckung zu erreichen, die Domäne besser zu verstehen, Over-Engineering zu vermeiden und den Fokus auf die eigentliche Aufgabe zu richten. Der größte Nachteil ist eine unmittelbare Folge dieser engen Fokussierung auf die jeweilige Aufgabe, d. h., um sich nicht von der Gesamtkomplexität des Projekts einschüchtern zu lassen, lösen wir als Entwickler immer nur die kleinstmögliche Herausforderung, und zwar immer nur eine. Wenn das Genie die Person ist, die die Komplexität beseitigt, indem sie sie löst, ist der TDD-Entwickler die Person, die die Komplexität absichtlich ignoriert. 

Ja, Sie haben es erfasst: So wie wir Pferde waren, die Scheuklappen trugen, so wie der Esel, der der Karotte folgte.

Aber die Komplexität verschwindet nicht, weil wir sie ignoriert haben. Sie bleibt dort und wartet darauf, dass wir uns ihr stellen. Wenn wir den Wald ignorieren, um das Blatt genau zu betrachten, hinterlassen wir eine technische Schuld. Wir hinterlassen ständig überflüssige Funktionen, doppelte Mitglieder, unbrauchbare Tests, unnötige Klassen, unleserlichen und unzugänglichen Code, Sie wissen schon. Diese technischen Schulden, die sich während der Entwicklung ansammeln, können unserer Produktivität schaden. Das ist der Grund, warum die Überarbeitung ein integraler Bestandteil der TDD-Praxis ist. Das folgende Diagramm zeigt die typischen Schritte einer TDD-Praxis.

Abb. 01 - Die typischen Schritte einer TDD-Praxis: Rot, Grün, Überarbeitung (Quelle: IBM-Entwickler)

In den folgenden Abschnitten beschreiben wir, wie wir unsere zuvor geschriebenen Klassen überarbeitet haben und kommentieren einige Verbesserungen. Wir zeigen, wie wir unsere PUBLISH-Pakete nach diesen Verbesserungen bauen, und wie wir zu einem brauchbaren Entwurf für unsere Paketbauklassen gekommen sind. Die erste Klasse, die dem neuen Muster folgt, ist die Klasse PUBACK. Da PUBACK-Pakete das Gegenstück zu PUBLISH-Paketen mit QoS 1 sind, müssen wir uns mit dem Session State Management befassen. Unser Kunde wird eine Art Persistenzschicht benötigen, um den Zustand zu erhalten und zu aktualisieren. 

Die Persistenzschicht fällt nicht in den Anwendungsbereich der OASIS-Norm. Sie ist anwendungsspezifisch. Dabei kann es sich um eine einfache Datei im lokalen Dateisystem oder um ein vollständig verteiltes, hochverfügbares Datenbanksystem in der Cloud handeln. Für unsere Zwecke würde eine Datenbank wie ein PostgreSQL-Server ausreichen, der lokal unter Windows oder über WSL läuft. Da wir jedoch eine native Integration zwischen MQL und SQLite haben, ist dieses Single-File-No-Server-RDBMS hier die offensichtliche Wahl. SQLite ist leichtgewichtig, skalierbar, vertrauenswürdig und frei von Serverwartung. Wir können sogar eine Nur-Speicher-Datenbank verwenden, was für Tests und Fehlersuche sehr praktisch ist. 

Aber wir werden die Persistenzschicht an dieser Stelle nicht implementieren, weil wir das Schreiben und Lesen von Paketen gut getestet haben wollen, bevor wir uns mit der Verwaltung des Sitzungsstatus befassen. Wir müssen sicher sein, dass wir die verschiedenen vom MQTT-Protokoll verwendeten Datentypen korrekt kodieren und dekodieren, bevor wir zur Persistenzschicht übergehen. Um dieses Ziel zu erreichen, schreiben wir umfangreiche Unit-Tests und werden bald mit kleinen Funktionstests gegen einen echten, lokal laufenden Nachrichten-Broker (dem „Open-Source Mosquitto-Broker“ von der Eclipse Foundation) beginnen.

Um unsere PUBLISH/PUBACK-Interaktionen zu testen, werden wir eine erfundene Datenbank verwenden, eine Sammlung von Funktionen, um die kontrollierten Daten zu erzeugen, die wir für die Tests benötigen, eine Art Fixture. Wir werden sie weiter unten bei der Beschreibung der Klasse CPuback vorstellen.

In den folgenden Beschreibungen werden die Begriffe MUSS und KANN so verwendet, wie sie im OASIS-Standard verwendet werden, der sie wiederum wie in IETF RFC 2119 beschrieben verwendet.

Wenn nicht anders angegeben, stammen alle Zitate aus dem OASIS Standard.

Wie wir PUBLISH-Pakete erstellen

Bei der Überarbeitung der Klasse CPublish haben wir einige Klassenmitglieder entfernt. Außerdem haben wir die Erstellung fester und variabler Kopfzeilen in einem einzigen Schritt zusammengefasst. Diese Änderungen werden auch in anderen Control Packet-Klassen umgesetzt.

Derzeit hat unsere CPublish-Klasse die folgenden Mitglieder und Methoden.

//+------------------------------------------------------------------+
//|                                                      Publish.mqh |
//|            ********* WORK IN PROGRESS **********                 |
//| **** PART OF ARTICLE https://www.mql5.com/en/articles/14391 **** |
//+------------------------------------------------------------------+
#include "IControlPacket.mqh"
//+------------------------------------------------------------------+
//|        PUBLISH VARIABLE HEADER                                   |
//+------------------------------------------------------------------+
/*
The Variable Header of the PUBLISH Packet contains the following fields in the order: Topic Name,
Packet Identifier, and Properties.
*/
//+------------------------------------------------------------------+
//| Class CPublish.                                                  |
//| Purpose: Class of MQTT Publish Control Packets.                  |
//|          Implements IControlPacket                               |
//+------------------------------------------------------------------+
class CPublish : public IControlPacket
  {
private:
   bool              IsControlPacket() {return true;}
   bool              HasWildcardChar(const string str);
protected:
   uchar             m_pubflags;
   uint              m_remlen;
   uchar             m_topname[];
   uchar             m_props[];
   uint              m_payload[];
public:
                     CPublish();
                    ~CPublish();
   //--- methods for setting Publish flags
   void              SetRetain(const bool retain);
   void              SetQoS_1(const bool QoS_1);
   void              SetQoS_2(const bool QoS_2);
   void              SetDup(const bool dup);
   //--- method for setting Topic Name
   void              SetTopicName(const string topic_name);
   //--- methods for setting Properties
   void              SetPayloadFormatIndicator(PAYLOAD_FORMAT_INDICATOR format);
   void              SetMessageExpiryInterval(uint msg_expiry_interval);
   void              SetTopicAlias(ushort topic_alias);
   void              SetResponseTopic(const string response_topic);
   void              SetCorrelationData(uchar &binary_data[]);
   void              SetUserProperty(const string key, const string val);
   void              SetSubscriptionIdentifier(uint subscript_id);
   void              SetContentType(const string content_type);
   //--- method for setting the payload
   void              SetPayload(const string payload);
   //--- method for building the final packet
   void              Build(uchar &result[]);
  };

Neben der Vereinfachung ist nun der Prozess des Setzens der Veröffentlichungsflags, der Themennamen und der Eigenschaften unabhängig voneinander, d.h. jedes dieser Elemente kann in beliebiger Reihenfolge gesetzt werden, vorausgesetzt, die Build()-Methode wird als letzte aufgerufen.
Dieser Test formalisiert dieses Verhalten. Er testet den Klassenkonstruktor mit zwei gesetzten Flags, RETAIN und QoS1, und dem erforderlichen Topic Name.

bool TEST_Ctor_Retain_QoS1_TopicName1Char()
  {
   Print(__FUNCTION__);
   CPublish *cut = new CPublish();
   uchar expected[] = {51, 6, 0, 1, 'a', 0, 1, 0}; // QoS > 0 require packet ID
   uchar result[];
   cut.SetTopicName("a");
   cut.SetRetain(true);
   cut.SetQoS_1(true);
   cut.Build(result);
   bool isTrue = AssertEqual(expected, result);
   delete(cut);
   ZeroMemory(result);
   return isTrue;
  }

Jetzt können die Methoden SetTopicName(), SetRetain() und SetQos1() in beliebiger Reihenfolge aufgerufen werden und das resultierende Paket ist immer noch gültig. Wie gesagt, dieses Verhalten wird in allen Kontrollpaketklassen reproduziert, und wir haben einen Test für jede Kombination von Veröffentlichungsflags. In den beigefügten Dateien finden Sie alle Tests.

Der feste Header des PUBLISH-Pakets

Die festen Header der PUBLISH-Pakete unterscheiden sich von allen anderen MQTT 5.0 Control Packets in der aktuellen Version des Protokolls. Sie haben drei Flags, die NICHT für eine zukünftige Verwendung reserviert sind: RETAIN-, QoS- und DUP-Flags. Im vorigen Artikel, Teil 5 dieser Serie, finden Sie einen ausführlichen Bericht über diese PUBLISH-Flags.

Abb. 02 - MQTT 5.0 PUBLISH-Paket Fixed Header RETAIN, QoS Level und DUP-Flags

Wir verwenden dasselbe Muster, um die Veröffentlichungsflags umzuschalten, aber jetzt, nach der Umstrukturierung, rufen wir SetFixedHeader() nicht mehr in jedem von ihnen auf. Zunächst definieren wir die Umschaltfunktion als booleschen Wert, der als Argument an die Funktion übergeben wird.

void CPktPublish::SetRetain(const bool retain)
  {
   retain ? m_pubflags |= RETAIN_FLAG : m_pubflags &= ~RETAIN_FLAG;
  }

Dann prüfen wir, ob der boolesche Wert wahr oder falsch ist.

void CPktPublish::SetQoS_1(const bool QoS_1)
  {
   QoS_1 ? m_pubflags |= QoS_1_FLAG : m_pubflags &= ~QoS_1_FLAG;
  }

Wenn der boolesche Wert wahr ist, führen wir eine bitweise ODER-Zuweisung zwischen dem Flag-Wert und einem uchar-Mitglied (ein Byte) durch, um das Flag zu setzen.

void CPktPublish::SetQoS_2(const bool QoS_2)
  {
   QoS_2 ? m_pubflags |= QoS_2_FLAG : m_pubflags &= ~QoS_2_FLAG;
  }

Wenn der boolesche Wert falsch ist, führen wir eine bitweise UND-Zuweisung zwischen dem Flag-Wert und demselben uchar-Mitglied durch, um das Flag zu löschen.

void CPktPublish::SetDup(const bool dup)
  {
   dup ? m_pubflags |= DUP_FLAG : m_pubflags &= ~DUP_FLAG;
  }

Auf diese Weise enthält die Variable m_pubflags alle Flags, die bei der Konfiguration des Pakets gesetzt oder gelöscht wurden. Später, wenn die Build()-Methode aufgerufen wird, führen wir erneut eine bitweise ODER-Zuweisung durch, dieses Mal zwischen den m_pubflags und dem ersten Byte des Pakets (Byte 0).

pkt[0] |= m_pubflags;

Der variable Kopf des PUBLISH-Pakets

Der variable Header des PUBLISH-Pakets enthält die folgenden Felder in dieser Reihenfolge: Topic Name, Packet Identifier und Eigenschaften.

Topic Name

Da alle Beziehungen zwischen Herausgebern und Abonnenten an den Topic Name der Publikation gebunden sind, ist dieses Feld in PUBLISH-Paketen erforderlich und darf keine Platzhalterzeichen enthalten. Beim Setzen dieses Feldes gibt es zwei Schutzbedingungen, für Platzhalterzeichen und für eine Zeichenkette mit der Länge Null, die sofort zurückkehren und den Fehler protokollieren, wenn eine dieser Bedingungen erfüllt ist.

void CPktPublish::SetTopicName(const string topic_name)
  {
   if(HasWildcardChar(topic_name) || StringLen(topic_name) == 0)
     {
      ArrayFree(m_topname);
      return;
     }
   EncodeUTF8String(topic_name, m_topname);
  }

Wenn keine der Guard-Bedingungen erfüllt ist, wird die Zeichenkette als UTF-8 kodiert und das char-Array im geschützten Element m_topname gespeichert, um beim Aufruf von Build() in das endgültige Paket aufgenommen zu werden.

Packet Identifier

Der Packet Identifier wird NICHT vom Nutzer festgelegt und ist für QoS 0 nicht erforderlich. Stattdessen wird sie automatisch bei der Build()-Methode gesetzt, wenn die erforderliche QoS > 0 ist. 

// QoS > 0 requires packet ID
   if((m_pubflags & 0x06) != 0)
     {
      SetPacketID(pkt, pkt.Size());
     }

Bei der Erstellung des endgültigen Pakets wird das Element m_pubflags durch ein bitweises UND mit dem Binärwert 0110 (0x06) geprüft. Wenn das Ergebnis ungleich Null ist, wissen wir, dass das Paket QoS_1 oder QoS_2 hat, und wir setzen den Packet Identifier.
Die Funktion SetPacketID erzeugt eine pseudozufällige Ganzzahl unter Verwendung von TimeLocal(), um den Anfangszustand zu erzeugen. Um uns das Leben beim Testen zu erleichtern, haben wir eine boolesche Variable TEST definiert. Wenn diese Variable true ist, setzt die Funktion den Wert 1 als Paket-ID.

//+------------------------------------------------------------------+
//|            SetPacketID                                           |
//+------------------------------------------------------------------+
#define TEST true

void SetPacketID(uchar& buf[], int start_idx)
  {
// MathRand - Before the first call of the function, it's necessary to call
// MathSrand to set the generator of pseudorandom numbers to the initial state.
   MathSrand((int)TimeLocal());
   int packet_id = MathRand();
   if(ArrayResize(buf, buf.Size() + 2) < 0)
     {
      printf("ERROR: failed to resize array at %s", __FUNCTION__);
      return;
     }
   buf[start_idx] = (uchar)packet_id >> 8; // MSB
   buf[start_idx + 1] = (uchar)(packet_id % 256) & 0xff; //LSB
//--- if testing, set packet ID to 1
   if(TEST)
     {
      Print("WARN: SetPacketID TEST true fixed ID = 1");
      buf[start_idx] = 0; // MSB
      buf[start_idx + 1] = 1; //LSB
     }
  }

Wie Sie sehen können, haben wir auch eine WARNUNG für den Fall der Fälle eingerichtet.

Eigenschaften

In Teil 4 dieser Artikelserie haben wir im Detail gesehen, was Eigenschaften sind und welche Rolle sie als Teil der MQTT 5.0 Erweiterungsmechanismen spielen. Im Folgenden wird beschrieben, wie wir sie implementieren, mit besonderem Augenmerk auf die verschiedenen Datentypen.

Es gibt sechs Arten der Datendarstellung, die zur Kodierung der Eigenschaftswerte in einem MQTT 5.0 Control Packet verwendet werden:

1. One Byte Integer, d.h. 8-Bit-Ganzzahlen ohne Vorzeichen.
2. Zwei Byte-Ganzzahlen, d. h. 16-Bit-Ganzzahlen ohne Vorzeichen in Big-Endian-Reihenfolge, wird auch Netzwerkreihenfolge genannt.
3. Vier Byte-Ganzzahlen, d. h. 32-Bit-Ganzzahlen ohne Vorzeichen, ebenfalls in Big-Endian-Reihenfolge.
4. Variable Byte-Ganzzahlen, die die Mindestanzahl von bis zu vier Bytes verwenden, um einen Wert zwischen 0 und 268.435.455 darzustellen.
5. Binäre Daten mit einer Länge zwischen 0 und 65.535.
6. UTF-8-kodierte Zeichenketten, die auch zur Kodierung eines Schlüssel:Wert-Paares in den Nutzereigenschaften verwendet werden können.

Die folgende Tabelle zeigt die verfügbaren PUBLISH-Eigenschaften und ihre jeweilige Datendarstellung.

Eigenschaft	Darstellung der Daten
Nutzdatenformat-Indikator	Ein Byte Ganzzahl
Ablaufintervall der Nachricht	Vier Byte Ganzzahl
Topic Alias	Zwei-Byte Ganzzahl
Topic Antwort	UTF-8 kodierte Zeichenfolge
Korrelationsdaten	Binäre Daten
Nutzer-Eigenschaft	UTF-8 kodiertes String-Paar
Abo-Identifikator	Variable Byte Ganzzahl
Typ des Inhalts	UTF-8 kodierte Zeichenfolge

Tabelle 01 - PUBLISH-Eigenschaften und ihre jeweilige Datendarstellung in MQTT 5.0

Unsere Eigenschaftsbezeichner wurden in unsere Defines.mqh-Kopfzeile aufgenommen.

//+------------------------------------------------------------------+
//|              PROPERTIES                                          |
//+------------------------------------------------------------------+
/*
The last field in the Variable Header of the CONNECT, CONNACK, PUBLISH, PUBACK, PUBREC,
PUBREL, PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK, DISCONNECT, and
AUTH packet is a set of Properties. In the CONNECT packet there is also an optional set of Properties in
the Will Properties field with the Payload
*/
#define MQTT_PROP_IDENTIFIER_PAYLOAD_FORMAT_INDICATOR          0x01 // (1) Byte                  
#define MQTT_PROP_IDENTIFIER_MESSAGE_EXPIRY_INTERVAL           0x02 // (2) Four Byte Integer     
#define MQTT_PROP_IDENTIFIER_CONTENT_TYPE                      0x03 // (3) UTF-8 Encoded String  
#define MQTT_PROP_IDENTIFIER_RESPONSE_TOPIC                    0x08 // (8) UTF-8 Encoded String  
#define MQTT_PROP_IDENTIFIER_CORRELATION_DATA                  0x09 // (9) Binary Data           
#define MQTT_PROP_IDENTIFIER_SUBSCRIPTION_IDENTIFIER           0x0B // (11) Variable Byte Integer
#define MQTT_PROP_IDENTIFIER_SESSION_EXPIRY_INTERVAL           0x11 // (17) Four Byte Integer   
.
.
. 

Nutzdatenformat-Indikator

Der Payload Format Indicator kann einen Wert von 0 oder 1 haben, d. h. rohe Bytes bzw. eine UTF-8-kodierte Zeichenkette. Falls nicht vorhanden, wird der Wert 0 angenommen (Rohbytes).

Obwohl dieses Feld direkt im m_props Member Array gesetzt werden könnte, haben wir uns für die Verwendung eines lokalen Hilfspuffers als Vermittler entschieden, um mit der Mehrheit der Eigenschaften, die eine Art von Manipulation erfordern, bevor sie in das endgültige Properties Array kopiert werden, konsistent zu sein.

void CPktPublish::SetPayloadFormatIndicator(PAYLOAD_FORMAT_INDICATOR format)
  {
   uchar aux[2];
   aux[0] = MQTT_PROP_IDENTIFIER_PAYLOAD_FORMAT_INDICATOR;
   aux[1] = (uchar)format;
   ArrayCopy(m_props, aux, m_props.Size());
  }

Obwohl es nur zwei mögliche Werte für diese Eigenschaft gibt, haben wir uns aus Gründen der Lesbarkeit dafür entschieden, ihnen einen symbolischen Wert zuzuordnen.

enum PAYLOAD_FORMAT_INDICATOR
  {
   RAW_BYTES   = 0x00,
   UTF8        = 0x01
  };

Durch die Verwendung dieses symbolischen Wertes wird der Methodenaufruf für den Endnutzer der Bibliothek eindeutig.

cut.SetPayloadFormatIndicator(RAW_BYTES);
cut.SetPayloadFormatIndicator(UTF8);

Ablaufintervall der Nachricht

Das Intervall für den Ablauf der Nachricht wird als Vier-Byte-Ganzzahl dargestellt. Es ist zu beachten, dass sich diese Darstellung von der einer variablen Byte-Ganzzahl unterscheidet. Während Letztere die minimale Anzahl von Bytes verwenden, die zur Darstellung des Wertes erforderlich sind, werden bei ersteren immer die gesamten vier Bytes verwendet.

void CPktPublish::SetMessageExpiryInterval(uint msg_expiry_interval)
  {
   uchar aux[4];
   aux[0] = MQTT_PROP_IDENTIFIER_MESSAGE_EXPIRY_INTERVAL;
   ArrayCopy(m_props, aux, m_props.Size(), 0, 1);
   EncodeFourByteInteger(msg_expiry_interval, aux);
   ArrayCopy(m_props, aux, m_props.Size());
  }

Unsere Funktion zur Kodierung der Vier-Byte-Ganzzahl folgt einem bekannten Muster von Zweierpotenzen von Rechtsverschiebungen, um die erforderliche Big-Endian-Reihenfolge (oder Netzwerkreihenfolge) zu gewährleisten.

void EncodeFourByteInteger(uint val, uchar &dest_buf[])
  {
   ArrayResize(dest_buf, 4);
   dest_buf[0] = (uchar)(val >> 24) & 0xff;
   dest_buf[1] = (uchar)(val >> 16) & 0xff;
   dest_buf[2] = (uchar)(val >> 8) & 0xff;
   dest_buf[3] = (uchar)val & 0xff;
  }

Topic Alias

Die Eigenschaft Topic Alias kann verwendet werden, um die Paketgröße zu reduzieren. Sie ist auf jede Netzwerkverbindung beschränkt und ist Teil des MQTT-Sitzungsstatus. Unsere Funktion zum Festlegen des Topic Alias kann also als Stummel (stub) betrachtet werden, so wie sie jetzt ist. Sie muss ausgefüllt werden, wenn es um den Sitzungsstatus geht.

void CPktPublish::SetTopicAlias(ushort topic_alias)
  {
   uchar aux[2];
   aux[0] = MQTT_PROP_IDENTIFIER_TOPIC_ALIAS;
   ArrayCopy(m_props, aux, m_props.Size(), 0, 1);
   EncodeTwoByteInteger(topic_alias, aux);
   ArrayCopy(m_props, aux, m_props.Size());
  }

Unsere Funktion zur Kodierung der Zwei-Byte-Ganzzahl folgt demselben bekannten Muster, das wir bei der Kodierung von Vier-Byte-Ganzzahlen verwendet haben, d. h. Zweierpotenz-Rechtsverschiebung, um die erforderliche Big-Endian-Reihenfolge zu gewährleisten.

void EncodeTwoByteInteger(uint val, uchar &dest_buf[])
  {
   ArrayResize(dest_buf, 2);
   dest_buf[0] = (uchar)(val >> 8) & 0xff;
   dest_buf[1] = (uchar)val & 0xff;
  }

Response Topic

Die Eigenschaft Response Topic ist nicht Teil des Publish/Subscribe-Musters. Stattdessen ist sie Teil der Anfrage/Antwort-Interaktion über MQTT. Wie Sie sehen, verwendet unsere Funktion zwei Hilfspuffer, einen für den Eigenschaftsbezeichner und den anderen für die kodierte UTF-8-Zeichenfolge. Dasselbe geschieht mit anderen UTF-8-kodierten Strings, da unsere String-Encoder-Funktion keinen dritten Parameter hat, um den Startindex des Zielpuffers zu adressieren. Dieses Problem kann in den nächsten Versionen durch eine Überlastung gelöst werden.

void CPktPublish::SetResponseTopic(const string response_topic)
  {
   uchar aux[1];
   aux[0] = MQTT_PROP_IDENTIFIER_RESPONSE_TOPIC;
   ArrayCopy(m_props, aux, m_props.Size());
   uchar buf[];
   EncodeUTF8String(response_topic, buf);
   ArrayCopy(m_props, buf, m_props.Size());
  }

Korrelationsdaten

Die Eigenschaft Korrelationsdaten ist ebenfalls Teil der Anfrage/Antwort-Interaktion über MQTT und nicht Teil des Publish/Subscribe-Musters. Da es sich bei dem Wert um Binärdaten handelt, kopiert unsere Funktion einfach die als Argument übergebenen Daten in das Byte-Array m_props, nachdem der Eigenschaftsbezeichner festgelegt wurde.

void CPktPublish::SetCorrelationData(uchar &binary_data[])
  {
   uchar aux[1];
   aux[0] = MQTT_PROP_IDENTIFIER_CORRELATION_DATA;
   ArrayCopy(m_props, aux, m_props.Size());
   ArrayCopy(m_props, binary_data, m_props.Size());
  }

Nutzer-Eigenschaft

Die User Property ist die flexibelste MQTT 5.0-Eigenschaft, da sie zur Übertragung von UTF-8-kodierten Schlüssel:Wert-Paaren mit anwendungsspezifischer Semantik verwendet werden kann.

„Nicht-normativer Kommentar

Diese Eigenschaft soll eine Möglichkeit zur Übertragung von Name-Value-Tags der Anwendungsschicht bieten, deren Bedeutung und Interpretation nur den Anwendungsprogrammen bekannt ist, die für das Senden und Empfangen dieser Tags verantwortlich sind.“

Unsere Funktion verwendet drei Hilfspuffer, um diese Eigenschaft zu kodieren, da unser UTF-8-String-Encoder derzeit keinen dritten Parameter für den Startindex des Zielpuffers hat. Dieses Problem könnte in den nächsten Versionen durch eine Überlastung gelöst werden. (siehe oben Antwort Thema.)

void CPktPublish::SetUserProperty(const string key, const string val)
  {
   uchar aux[1];
   aux[0] = MQTT_PROP_IDENTIFIER_USER_PROPERTY;
   ArrayCopy(m_props, aux, m_props.Size());
   uchar key_buf[];
   EncodeUTF8String(key, key_buf);
   ArrayCopy(m_props, key_buf, m_props.Size());
   uchar val_buf[];
   EncodeUTF8String(val, val_buf);
   ArrayCopy(m_props, val_buf, m_props.Size());
  }

Abo-Identifikator

Unsere Funktion zum Einstellen der Eigenschaft Abonnementkennung beginnt mit der Überprüfung, ob das übergebene Argument zwischen 1 und 268.435.455 liegt, was die akzeptierten Werte für diese Eigenschaft sind. Ist dies nicht der Fall, drucken/protokollieren wir eine Fehlermeldung und kehren sofort zurück.

void CPktPublish::SetSubscriptionIdentifier(uint subscript_id)
  {
   if(subscript_id < 1 || subscript_id > 0xfffffff)
     {
      printf("Error: " + __FUNCTION__ +  "Subscription Identifier must be between 1 and 268,435,455");
      return;
     }
   uchar aux[1];
   aux[0] = MQTT_PROP_IDENTIFIER_SUBSCRIPTION_IDENTIFIER;
   ArrayCopy(m_props, aux, m_props.Size());
   uchar buf[];
   EncodeVariableByteInteger(subscript_id, buf);
   ArrayCopy(m_props, buf, m_props.Size());
  }

Typ des Inhalts

Der Wert der Eigenschaft Inhaltstyp wird von der Anwendung festgelegt. „MQTT führt keine Validierung der Zeichenkette durch, außer um sicherzustellen, dass es sich um eine gültige UTF-8 kodierte Zeichenkette handelt.“

void CPktPublish::SetContentType(const string content_type)
  {
   uchar aux[1];
   aux[0] = MQTT_PROP_IDENTIFIER_CONTENT_TYPE;
   ArrayCopy(m_props, aux, m_props.Size());
   uchar buf[];
   EncodeUTF8String(content_type, buf);
   ArrayCopy(m_props, buf, m_props.Size());
  };

Nutzlast

Das letzte Feld im Header der PUBLISH-Variable ist die Nutzlast (payload), wie es richtig heißt. Eine Nutzlast der Länge Null ist gültig. Unsere Funktion ist nichts weiter als ein Wrapper um unseren UTF-8-String-Encoder, der dem gleichen Muster folgt, indem er einen Hilfspuffer verwendet, der dann in das m_payload-Mitglied kopiert wird.

void CPktPublish::SetPayload(const string payload)
  {
   uchar aux[];
   EncodeUTF8String(payload, aux);
   ArrayCopy(m_payload, aux, m_props.Size());
  }

Die endgültige Build-Methode

Der Zweck der Methode Build() besteht darin, den Fixed Header, den Topic Name, den Packet Identifier, die Properties und die Payload im endgültigen Paket zusammenzuführen, wobei sowohl die Property(ies) Length als auch die Packet Remaining Length als variable Byte-Integer kodiert werden.

Zunächst wird geprüft, ob der obligatorische Topic Name vorhanden ist. Wenn die Länge Null ist, wird der Fehler gedruckt/protokolliert und sofort zurückgegeben.

void CPktPublish::Build(uchar &pkt[])
  {
   if(m_topname.Size() == 0)
     {
      printf("Error: " + __FUNCTION__ + " topic name is mandatory");
      return;
     }
   ArrayResize(pkt, 2);

Dann setzen wir das erste Byte des Fixed Headers mit dem Typ des Kontrollpakets und den entsprechenden PUBLISH-Flags.

// pkt type with publish flags
   pkt[0] = (uchar)PUBLISH << 4;
   pkt[0] |= m_pubflags;

Dann kopieren wir das Array m_topname in das endgültige Paket und setzen/kopieren den Packet Identifier, wenn QoS > 0 ist.

// topic name
   ArrayCopy(pkt, m_topname, pkt.Size());
// QoS > 0 require packet ID
   if((m_pubflags & 0x06) != 0)
     {
      SetPacketID(pkt, pkt.Size());
     }

Als Nächstes kodieren wir die Länge der Eigenschaft(en) als variable Byte-Ganzzahl.

// properties length
   uchar buf[];
   EncodeVariableByteInteger(m_props.Size(), buf);
   ArrayCopy(pkt, buf, pkt.Size());

Wir kopieren die Eigenschaften und die Nutzlast aus ihren Klassenmitgliedern in das endgültige Paket-Array.

// properties
   ArrayCopy(pkt, m_props, pkt.Size());
// payload
   ArrayCopy(pkt, m_payload, pkt.Size());

Zum Schluss wird die verbleibende Länge des Pakets als variable Byte-Ganzzahl kodiert.

// remaining length
   m_remlen += pkt.Size() - 2;
   uchar aux[];
   EncodeVariableByteInteger(m_remlen, aux);
   ArrayCopy(pkt, aux, 1);
  }

Das PUBACK-Kontrollpaket

Wie wir oben bei der Implementierung unserer Klasse CPublish gesehen haben, benötigt jedes PUBLISH-Paket mit QoS 1 einen Packet Identifier ungleich Null. Diese Paket-ID wird in dem entsprechenden PUBACK-Paket zurückgegeben. Anhand dieser ID kann unser Client erkennen, ob das zuvor gesendete PUBLISH-Paket zugestellt wurde oder ob ein Fehler aufgetreten ist. Unabhängig davon, ob die Zustellung erfolgreich war oder nicht, ist das PUBACK der Auslöser, mit dem wir den Sitzungsstatus aktualisieren. Wir aktualisieren den Session State (Sitzungsstatus) auf der Grundlage des/der Reason Codes.

Das PUBACK-Paket gibt einen von neun Reason Codes (Begründungscode) zurück.

SUCCESS - Mit der Nachricht ist alles in Ordnung. Sie wurde angenommen, und die Veröffentlichung ist im Gange. „SUCCESS“ (Erfolg) bedeutet hier, dass der Empfänger den Besitz der Nachricht akzeptiert hat. Dies ist der einzige Reason Code, der implizit sein kann, d.h. er ist der einzige Reason Code, der weggelassen werden kann. Ein PUBACK mit nur einer Paket-ID MUSS als erfolgreiche QoS-1-Zustellung interpretiert werden.

„Der Client oder Server, der das PUBACK-Paket sendet, MUSS einen der PUBACK Reason Codes [MQTT-3.4.2-1] verwenden. Der Reason Code und die Property Length können weggelassen werden, wenn der Reason Code 0x00 (Success) ist und es keine Properties gibt.“

NO MATCHING SUBSCRIBERS (Keine Abonnenten) - Mit der Nachricht ist alles in Ordnung. Er wurde angenommen und die Veröffentlichung ist im Gange, aber niemand hat seinen Topic Name abonniert. Dieser Reason Code wird nur vom Nachrichten-Broker gesendet und ist optional, d.h. der Broker KANN diesen Reason Code anstelle von SUCCESS senden.

UNSPECIFIED ERROR - Die Nachricht wurde abgelehnt, aber der Herausgeber möchte den Grund nicht preisgeben, oder keiner der anderen, spezifischeren Reason Codes ist geeignet, den Grund zu beschreiben.

IMPLEMENTATION SPECIFIC ERROR (implementierungsspezifischer Fehler) - Mit der Nachricht ist alles in Ordnung, aber der Herausgeber möchte sie nicht veröffentlichen. Der Standard bietet keine weiteren Details über die Semantik dieses Reason Codes, aber wir können daraus schließen, dass der Grund für die Nichtveröffentlichung nicht in den Anwendungsbereich des Protokolls fällt, d.h. anwendungsspezifisch ist.

NOT AUTHORIZED (nicht autorisiert) - Selbsterklärend.

TOPIC NAME INVALID (Topic Name invalid) - Alles ist in Ordnung mit der Nachricht, einschließlich des Topic Name, der eine wohlgeformte, gut kodierte UTF-8-Zeichenfolge ist, aber der Herausgeber, sei es der Kunde oder der Nachrichten-Broker, akzeptiert diesen Topic Name nicht. Auch hier können wir daraus schließen, dass der Grund für die Nichtveröffentlichung anwendungsspezifisch ist.

PACKET IDENTIFIER IN USE (Paket Identifikator bereits verwendet) - Alles ist in Ordnung mit der Nachricht, aber es gibt eine mögliche Unstimmigkeit im Sitzungsstatus zwischen dem Client und dem Nachrichten-Broker, weil die Paket-ID, die wir in PUBLISH gesendet haben, bereits in Gebrauch ist.

QUOTA EXCEEDED (Quantum überschritten) - Selbsterklärend. Auch hier gilt, dass der Grund für die Ablehnung nicht in den Rahmen des Protokolls fällt. Sie ist anwendungsspezifisch.

PAYLOAD FORMAT INVALID (ungültigen Format der Payload) - Alles ist in Ordnung mit der Nachricht, aber die Eigenschaft Payload Format Indicator, die wir in unserem PUBLISH gesendet haben, unterscheidet sich von dem tatsächlichen Payload-Format.

Neben dem Reason Code kann das PUBACK-Paket auch einen Reason String und eine User Property enthalten.

Reason String ist eine menschenlesbare UTF-8-kodierte Zeichenfolge, die bei der Diagnose helfen soll. Sie ist nicht dazu bestimmt, vom Empfänger entschlüsselt zu werden. Stattdessen dienen sie als Träger zusätzlicher Informationen, die protokolliert, gedruckt, an Berichte angehängt werden können usw. Es ist anzumerken, dass jeder konforme Server oder Client den Reason String nicht sendet, wenn seine Einbeziehung die Paketgröße über die zum Zeitpunkt der Verbindung (CONNECT-Paket) angegebene maximale Paketgröße hinaus erhöht.

Das PUBACK kann auch eine beliebige Anzahl von Schlüssel:Wert-Paaren enthalten, die als Nutzereigenschaft(en) kodiert sind. Diese Paare können verwendet werden, um zusätzliche Informationen über den Fehler zu liefern und sind auch anwendungsspezifisch. Das heißt, das Protokoll definiert ihre Semantik nicht. 

Unser Client „MUSS das PUBLISH-Paket als ‚unbestätigt‘ behandeln, bis er das entsprechende PUBACK-Paket vom Empfänger erhalten hat.“

Die Klasse CPuback

Unsere Klasse CPuback folgt demselben Bauplan wie die Klasse CPublish. Sie implementiert auch die Schnittstelle IControlPacket, die als unsere Stub-Root für die Objekthierarchie steht.

Ein PUBACK-Paket wird als Antwort auf PUBLISH-Pakete mit QoS 1 gesendet. Sein zwei Byte langer fester Header enthält nur die Kennung des Kontrollpakets auf dem ersten Byte und die verbleibende Länge des Pakets auf dem zweiten Byte. Seine Bit-Flags sind in dieser Version des Protokolls alle auf RESERVIERT gesetzt.

Abb. 03 - Struktur des festen Headers eines MQTT-5.0-PUBACK-Pakets

„Der variable Header des PUBACK-Pakets enthält die folgenden Felder in dieser Reihenfolge: Packet Identifier aus dem PUBLISH-Paket, das bestätigt wird, PUBACK Reason Code, Property Length und die Properties“.

Abb. 04 - Struktur des Variablen-Kopfes eines MQTT-5.0-PUBACK-Pakets

Bis jetzt haben wir unseren Kunden nur als Sender betrachtet; von nun an müssen wir auch die Rolle des Empfängers in Betracht ziehen. Der Grund dafür ist:

„Das Übermittlungsprotokoll ist symmetrisch, [...] der Client und der Server können jeweils die Rolle des Senders oder des Empfängers übernehmen.“

Wir müssen einen Test für eine Funktion schreiben, die die Kennung des zu bestätigenden Pakets ermittelt:

1. aus dem zurückgesendeten Paket, das der Nachrichten-Broker beim Empfang eines PUBACK,
2. oder von unserem Persistenzsystem beim Senden eines PUBACK.

Ein PUBLISH-Paket mit QoS 1 hat keine Bedeutung ohne sein entsprechendes PUBACK, das wiederum eine Art von Persistenz erfordert, um die Paket-ID des entsprechenden PUBLISH-Pakets zu speichern. Aber obwohl wir bereits wissen, dass wir irgendwann eine echte Datenbank als Persistenzschicht brauchen werden, brauchen wir sie im Moment noch nicht. Um unsere Funktion zu testen und weiterzuentwickeln, brauchen wir etwas, das wie eine Datenbank funktioniert, etwas, das bei einer Abfrage die Kennung der PUBLISH-Pakete zurückgibt, die noch nicht quittiert wurden. Um Überraschungen zu vermeiden, erstellen wir eine einzige Funktion namens GetPendingPublishIDs(ushort &result[]) und speichern sie in einer Datei namens DB.mqh.

void GetPendingPublishIDs(ushort &result[])
  {
   ArrayResize(result, 3);
   result[0] = 1;
   result[1] = 255; // one byte
   result[2] = 65535; // two bytes
  }

Mit unserer „Persistenzschicht“ können wir uns auf die eigentliche Aufgabe konzentrieren: eine Funktion zu schreiben, die, wenn sie ein vom Nachrichten-Broker gesendetes PUBACK-Byte-Array (Paket) erhält, die Kennung des bestätigten PUBLISH abfragt und mit den in unserer Persistenzschicht gespeicherten ausstehenden PUBLISH-IDs vergleicht. Wenn es eine ID-Übereinstimmung gibt, wird ‚True‘ zurückgegeben. Später, bei der Implementierung des Protokolls Operational Behavior, werden wir diese übereinstimmende ID aus dem echten Speicher freigeben.

In Anbetracht der obigen PUBACK-Variablen-Header-Struktur müssen wir jetzt nur noch die ersten beiden Bytes lesen, um die ID des Pakets zu erhalten, das bestätigt wird.

ushort CPuback::GetPacketID(uchar &pkt[])
  {
   return (pkt[0] * 256) + pkt[1];
  }

Wir erinnern uns, dass die Paketkennung als Zwei-Byte-Ganzzahl in Big-Endian- (oder Netzwerk-) Reihenfolge kodiert ist, wobei das höchstwertige Byte (MSB) zuerst erscheint. Zur Kodierung wurde eine bitweise Linksverschiebung verwendet (<<). Um sie zu dekodieren, multiplizieren wir den Wert des höchstwertigen Bytes mit 256 und addieren das niederwertige Byte.

Die obige Funktion ist für den Moment ausreichend. Später, wenn wir gegen einen echten Broker im offenen Netz testen, müssen wir uns möglicherweise mit Endianness-Problemen befassen, aber wir werden an dieser Stelle nicht darauf testen. Gehen wir weiter auf unsere attraktive Karotte zu, die Aufgabe, die vor uns liegt.

bool CPuback::IsPendingPkt(uchar &pkt[])
{
   ushort pending_ids[];
   GetPendingPublishIDs(pending_ids);
   ushort packet_id = GetPacketID(pkt);
   for(uint i = 0; i < pending_ids.Size(); i++)
     {
      if(pending_ids[i] == packet_id)
        {
         return true;
        }
     }
   return false;
}

Die obige Funktion erhält ein Byte-Array als Argument. Dieses Byte-Array ist der variable Header des PUBACK-Pakets. Es speichert dann in einer lokalen Variablen (pending_ids) ein Array von Paketkennungen aus unserem Speicher/Datenbank, die noch nicht bestätigt wurden. Schließlich liest er die Paket-ID aus dem vom Broker gesendeten Byte-Array und vergleicht sie mit dem Array der ausstehenden IDs. Wenn sich das Paket im Array befindet, gibt unsere Funktion „true“ zurück und wir können die ID freigeben.

Mit der gleichen Logik können wir die Paketkennungen von PUBREC, PUBREL und PUBCOMP für PUBLISH mit QoS 2 freigeben. Außerdem werden wir später unsere pseudo-Funktion-in-einer-Datei „Persistenzschicht“ durch eine echte Datenbank ersetzen, aber die Hauptlogik der Funktion wird beibehalten. An diesem Punkt könnte ein anderer Entwickler an der Persistenzschicht arbeiten, während wir unsere Paketklassen völlig unabhängig entwickeln.

Wir müssen auch in der Lage sein, den/die Reason Code(s) aus dem Header der Variable PUBACK zu lesen. Da dieses Feld eine feste Position und Größe hat, müssen wir nur dieses spezielle Byte lesen.

uchar CPuback::GetReasonCode(uchar &pkt[])
  {
   return pkt[2];
  }

Da wir nur auf der Empfängerseite unseres Clients arbeiten - d.h. noch keine PUBACKs senden - reichen die obigen Funktionen für unsere nächsten Funktionstests aus. Und jetzt gegen einen echten Broker.

Schlussfolgerung

Kontinuierliche Überarbeitung ist Teil der TDD-Praxis. Ziel ist nicht nur ein voll funktionsfähiger, sondern auch ein sauberer Code: einzelne Verantwortungseinheiten und Funktionen (hier Klassen und Methoden), lesbare Bezeichner (Klassen-, Methoden- und Variablennamen) und Vermeidung von Redundanz („wiederhole dich nicht“). Es handelt sich um einen Prozess, nicht um eine Aufgabe in einem Schritt. Wir wissen also schon jetzt, dass wir kontinuierlich Überarbeitung (refactoring) betreiben werden, bis wir einen voll funktionsfähigen MQTT 5.0-Client haben.

Jetzt sind wir bereit, unseren ersten Funktionstest mit einem echten MQTT-Broker zu schreiben, um zu sehen, ob unsere CONNECT-, CONNACK-, PUBLISH- und PUBACK-Pakete wie erwartet funktionieren. 

PUBACK-Pakete sind das Gegenstück zu den PUBLISH-Paketen mit QoS 1. PUBLISH-Pakete mit QoS 2 benötigen PUBREC-, PUBCOMP- und PUBREL-Pakete als Gegenstück. Sie sind das Thema unseres nächsten Artikels.

Wenn Sie MQL5 gut verstehen und zur Entwicklung dieses quelloffenen MQTT-Clients beitragen können, schreiben Sie bitte eine Nachricht in den Kommentaren unten oder in unserem Community Chat.