Protokół MQTT #3: implementacja klienta

Przyszedł czas na implementację klienta MQTT, a dokładnie publishera. Jak wiadomo z poprzedniego wpisu wykorzystamy w tym celu bibliotekę Paho MQTT Client napisaną w języku C.

⇒ Protokół MQTT: pokaż spis treści ⇐

Wprowadzenie

Na potrzeby tego wpisu wykorzystamy jedną z przykładowych aplikacji klienta umożliwiającej wysyłanie wiadomości. Ten jak i kilka innych dostępnych przykładów znajdziesz w folderze paho.mqtt.c/src/samples. Kod, po drobnych modyfikacjach, wygląda następująco:

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"

#define ADDRESS     "tcp://127.0.0.1:1883"
#define CLIENTID    "ExampleClientPub"
#define TOPIC       "MQTT Examples"
#define PAYLOAD     "Hello World!"
#define TIMEOUT     10000L

int main(int argc, char* argv[]) {

	MQTTClient client;
	MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE,
			NULL);

	MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;

	int rc;
	if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
		printf("Failed to connect, return code %d\n", rc);
		exit(EXIT_FAILURE);
	}

	MQTTClient_message pubmsg = MQTTClient_message_initializer;

	pubmsg.payload = PAYLOAD;
	pubmsg.payloadlen = strlen(PAYLOAD);
	pubmsg.qos = 1;
	pubmsg.retained = 0;

	MQTTClient_deliveryToken token;

	MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);

	printf("Waiting for up to %d seconds for publication of %s\n"
			"on topic %s for client with ClientID: %s\n",
			(int) (TIMEOUT / 1000), PAYLOAD, TOPIC, CLIENTID);

	rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);

	printf("Message with delivery token %d delivered\n", token);

	MQTTClient_disconnect(client, 10000);
	MQTTClient_destroy(&client);

	return rc;
}

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include "MQTTClient.h"

#define ADDRESS "tcp://127.0.0.1:1883"

#define CLIENTID "ExampleClientPub"

#define TOPIC "MQTT Examples"

#define PAYLOAD "Hello World!"

#define TIMEOUT 10000L

int main(int argc, char* argv[]) {

MQTTClient client;

MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE,

NULL);

MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

conn_opts.keepAliveInterval = 20;

conn_opts.cleansession = 1;

int rc;

if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {

printf("Failed to connect, return code %d\n", rc);

exit(EXIT_FAILURE);

}

MQTTClient_message pubmsg = MQTTClient_message_initializer;

pubmsg.payload = PAYLOAD;

pubmsg.payloadlen = strlen(PAYLOAD);

pubmsg.qos = 1;

pubmsg.retained = 0;

MQTTClient_deliveryToken token;

MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);

printf("Waiting for up to %d seconds for publication of %s\n"

"on topic %s for client with ClientID: %s\n",

(int) (TIMEOUT / 1000), PAYLOAD, TOPIC, CLIENTID);

rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);

printf("Message with delivery token %d delivered\n", token);

MQTTClient_disconnect(client, 10000);

MQTTClient_destroy(&client);

return rc;

}

Jak widać, w przykładzie zostało wykorzystane API standardowe, synchroniczne, dlatego dołączony został plik nagłówkowy MQTTClient.h. Oprócz tego, należy jeszcze pamiętać o podlinkowaniu biblioteki paho-mqtt3c. Kod klienta można podzielić na trzy podstawowe części: create, connect i publish/subscribe.

Create

Na początku definiujemy uchwyt do klienta:

MQTTClient client;

1	MQTTClient client;

Następnie go tworzymy za pomocą funkcji MQTTClient_create():

MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

1	MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

Jako pierwszy argument przekazujemy adres do utworzonego wcześniej uchwytu.
Kolejnym argumentem jest string zawierający adres brokera, z którym klient będzie się łączył. Powinien on mieć następujący format: „protocol://host:port”. W tym wypadku będziemy się łączyli z brokerem postawionym na localhoscie, więc makro ma postać:

#define ADDRESS     "tcp://localhost:1883"

1	#define ADDRESS "tcp://localhost:1883"

Trzecim argumentem jest string zawierający ID klienta, na podstawie którego broker będzie go identyfikował. Powinien on być unikalny dla każdego klienta. W tym przypadku też wykorzystane zostało makro:

#define CLIENTID    "ExampleClientPub"

1	#define CLIENTID "ExampleClientPub"

W kolejnym argumencie mamy możliwość przekazania jednego z trzech bibliotecznych makr. W tym przykładzie wykorzystane zostało MQTTCLIENT_PERSISTENCE_NONE, co powoduje, że w przypadku gdy klient padnie to aktualny stan przesyłanych wiadomości zostanie utracony i wiadomości mogą nie zostać doręczone, nawet przy QoS ustawionym na wartość 1 lub 2. W przypadku wyboru MQTTCLIENT_PERSISTENCE_DEFAULT wszystkie wiadomości, które są aktualnie przesyłane są także przechowywane na dysku. Dzięki czemu, w sytuacji gdy klient padnie (aplikacja się zawiesi), możliwe będzie jego odtworzenie. Wiadomości, które były w trakcie przesyłania także zostaną odczytane i klient będzie mógł rozpocząć swoją pracę w kontekście, w którym nastąpiła awaria. Ostatni argument jest zależny od poprzedniego i w związku z tym, że w poprzednim wybraliśmy MQTTCLIENT_PERSISTENCE_NONE to w tym należy wpisać NULL. Szczegółowe informacje na temat dwóch ostatnich argumentów można znaleźć w dokumentacji biblioteki.

Connect

Aby nawiązać połączenie z brokerem, klient musi mu przesłać zdefiniowany w standardzie pakiet CONNECT. Warto dodać, że wiele przesyłanych w pakiecie informacji, jak np. login czy hasło jest opcjonalnych. Jak wyglądają poszczególne ramki zobaczyć można w dokumentacji protokołu. Nie musimy jednak jej dogłębnie analizować, ponieważ biblioteka udostępnia nam gotowe funkcje i struktury, umożliwiające wprowadzenie konkretnych ustawień. Do nawiązania połączenia wykorzystywana jest funkcja MQTTClient_connect():

MQTTClient_connect(client, &conn_opts);

1	MQTTClient_connect(client, &conn_opts);

Jak widać, przyjmuje ona dwa argumenty. Pierwszym jest uchwyt naszego klienta, a drugim jest adres do struktury, w której przechowywane są wszystkie ustawienia.

Struktura jest na początku inicjalizowana standardowymi ustawieniami:

MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

1	MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

Jest to dość istotne, ponieważ dla elementów struktury nie są zdefiniowane wartości domyślne, więc w przypadku zmiennych automatycznych przyjmują one wartości losowe, a w przypadku zmiennych statycznych wypełnione są zerami, co będzie powodowało niepoprawną pracę klienta.

Po zainicjalizowaniu zmiennej, ustawiane są interesujące nas pola:

conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;

1 2	conn_opts.keepAliveInterval = 20; conn_opts.cleansession = 1;

Za pomocą pola keepAliveInterval ustawiamy okres czasu (w sekundach), co jaki klient ma wysyłać do brokera wiadomość pingującą. Broker na każdą z nich powinien odpowiedzieć wiadomością potwierdzającą. Taki mechanizm umożliwia obu stronom określenie czy strona przeciwna jest ciagle osiągalna.

Pole cleansession może przyjmować wartości 0 lub 1 i służy do kontroli zachowania klienta i brokera w czasie łączenia i rozłączania. Gdy cleansession jest ustawione na wartość 1 to uruchamiana jest czysta sesja. Wszystkie informacje z poprzedniej sesji zostaną wyczyszczone oraz broker nie będzie przechowywał dla klienta informacji z aktualnej sesji. W przypadku ustawienia cleansession na 0 zostanie odtworzona poprzednia sesja, a jeśli taka nie istnieje to zostanie utworzona nowa.

Struktura ma jeszcze wiele innych pól. Możliwe jest np. przesłania loginu i hasła na potrzeby uwierzytelnienia. Należy jednak mieć na uwadze, że jeśli nie korzystamy z protokołu SSL to dane przesyłane są jako tekst jawny. Dokładny opis wszystkich pól struktury można znaleźć tutaj.

Po przesłaniu przez klienta pakietu CONNECT broker zawsze odpowiada pakietem odpowiedzi CONNACK, w którym zawarty jest odpowiedni kod. W związku z tym, funkcja MQTTClient_connect() także zwraca różne wartości: albo flagę sukcesu albo kod błędu.

Publish

Przechodzimy teraz do wysyłania wiadomości 🙂 W tym celu wykorzystywana jest funkcja MQTTClient_publishMessage() posiadająca cztery parametry:

MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);

1	MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);

Pierwszym przekazywanym argumenetem, podobnie jak w przypadku MQTTClient_connect() jest uchwyt klienta. Drugim jest string zawierający nazwę tematu. W tym przypadku wykorzystane zostało makro:

#define TOPIC       "MQTT Examples"

1	#define TOPIC "MQTT Examples"

Kolejnym jest adres do struktury przechowującej treść wiadomości oraz różne jej właściwości, a ostatnim – adres do tzw. tokena dostarczenia, który jest wykorzystywany do przechowywania informacji o tym czy wiadomość została dostarczona z sukcesem.

Struktura, podobnie jak w przypadku conn_opts, jest na początku inicjalizowana standardowymi ustawieniami, po czym ustawiane są interesujące nas pola, np.:

MQTTClient_message pubmsg = MQTTClient_message_initializer;
pubmsg.payload = PAYLOAD;
pubmsg.payloadlen = strlen(PAYLOAD);
pubmsg.qos = 1;
pubmsg.retained = 0;

MQTTClient_message pubmsg = MQTTClient_message_initializer;

pubmsg.payload = PAYLOAD;

pubmsg.payloadlen = strlen(PAYLOAD);

pubmsg.qos = 1;

pubmsg.retained = 0;

Do pola payload przekazywany jest adres do stringa zawierającego treść wiadomości. W tym przypadku wykorzystane zostało makro:

#define PAYLOAD     "Hello World!"

1	#define PAYLOAD "Hello World!"

Do pola payloadlen przekazywana jest długość wiadomości, do qos wartość quality of service, o którym dokładniej pisałem w pierwszym wpisie tej serii. Pole retained może przyjmować wartość 0 lub 1. Ustawienie jedynki będzie informowało broker, że ma przechowywać kopię tej wiadomości. Dokładny opis wszystkich pól struktury można znaleźć tutaj.

Gdy wysyłamy jakąś wiadomość z QoS1 lub QoS2 i chcemy mieć pewność, że wiadomość została dostarczona to w związku z tym, że klient działa w trybie synchronicznym konieczne jest jeszcze wywołanie MQTTClient_waitForCompletion(), której argumentami są uchwyt do klienta, token oraz timeout. Funkcja blokuje wykonywanie programu do momentu dostarczenia wiadomości lub upłynięcia timeoutu.

Na końcu programu następuje rozłączenie i usunięcie klienta (zwolnienie pamięci):

MQTTClient_disconnect(client, 10000);
MQTTClient_destroy(&client);

1 2	MQTTClient_disconnect(client, 10000); MQTTClient_destroy(&client);

Drugim argumentem przekazywanym do funkcji MQTTClient_disconnect() jest wartość timeoutu, który jest wykorzystywany po to, aby umożliwić klientowi ewentualne dokończenie przesyłania wiadomości.

Czas na testy!

No i to wszystko. Czas na kompilację i testy. Na początku uruchamiamy program brokera i klienta subskrybującego., tak jak to robiliśmy w pierwszym wpisie. Jedyne o czym należy pamiętać to o wpisaniu prawidłowej nazwy tematu i adresu IP brokera. Przypominam, że jeżeli klient i broker są uruchomione na tej samej maszynie to nie trzeba nawet ustawiać flagi -h, wystarczy taki zapis:

mosquitto_sub -t "MQTT Examples"

1	mosquitto_sub -t "MQTT Examples"

Gdy wszystko jest gotowe, można uruchomić napisany program 🙂 Wiadomość powinna wyświetlić się w konsoli subscribera:

Podsumowanie

No i pierwszy program mamy za sobą 🙂 Napisanie innego klienta nie powinno stanowić większego problemu. Jak wspominałem wcześniej, w pobranym folderze, dostępnych jest jeszcze kilka innych przykładów, zarówno z API synchronicznym jak i asynchronicznym. Warto sobie je jeszcze dodatkowo przejrzeć 🙂 W razie potrzeby, całe dokumentacje biblioteki dla obu API, wraz z opisem wszystkich funkcji oraz zmiennych są dostępne tutaj oraz tutaj.

To na razie tyle, jeśli chodzi o protokół MQTT. W przyszłości planuję jeszcze wpisy, w których przedstawię jak uruchomić bibliotekę na mikrokontrolerach, np. STM32, ale to dopiero za jakiś czas.