MQTT5 support.

Author: dustin

src/MQTTClient.cpp

@@ -56,7 +56,7 @@ inline lwmqtt_err_t lwmqtt_arduino_network_write(void *ref, uint8_t *buffer, siz
 }
 
 static void MQTTClientHandler(lwmqtt_client_t * /*client*/, void *ref, lwmqtt_string_t topic,
-                              lwmqtt_message_t message) {
+                              lwmqtt_message_t message, lwmqtt_serialized_properties_t props) {
   // get callback
   auto cb = (MQTTClientCallback *)ref;
 
@@ -72,7 +72,7 @@ static void MQTTClientHandler(lwmqtt_client_t * /*client*/, void *ref, lwmqtt_st
 
   // call the advanced callback and return if available
   if (cb->advanced != nullptr) {
-    cb->advanced(cb->client, terminated_topic, (char *)message.payload, (int)message.payload_len);
+      cb->advanced(cb->client, terminated_topic, (char *)message.payload, (int)message.payload_len, props);
     return;
   }
 
@@ -118,7 +118,7 @@ MQTTClient::~MQTTClient() {
   free(this->writeBuf);
 }
 
-void MQTTClient::begin(const char hostname[], int port, Client &client) {
+void MQTTClient::begin(const char hostname[], int port, Client &client, lwmqtt_protocol_t protocol) {
   // set hostname and port
   this->setHost(hostname, port);
 
@@ -127,6 +127,7 @@ void MQTTClient::begin(const char hostname[], int port, Client &client) {
 
   // initialize client
   lwmqtt_init(&this->client, this->writeBuf, this->bufSize, this->readBuf, this->bufSize);
+  lwmqtt_set_protocol(&this->client, protocol);
 
   // set timers
   lwmqtt_set_timers(&this->client, &this->timer1, &this->timer2, lwmqtt_arduino_timer_set, lwmqtt_arduino_timer_get);
@@ -222,7 +223,8 @@ void MQTTClient::setOptions(int keepAlive, bool cleanSession, int timeout) {
   this->timeout = (uint32_t)timeout;
 }
 
-bool MQTTClient::publish(const char topic[], const char payload[], int length, bool retained, int qos) {
+bool MQTTClient::publish(const char topic[], const char payload[], int length, bool retained, int qos,
+                         lwmqtt_properties_t props) {
   // return immediately if not connected
   if (!this->connected()) {
     return false;
@@ -236,7 +238,7 @@ bool MQTTClient::publish(const char topic[], const char payload[], int length, b
   message.qos = lwmqtt_qos_t(qos);
 
   // publish message
-  this->_lastError = lwmqtt_publish(&this->client, lwmqtt_string(topic), message, this->timeout);
+  this->_lastError = lwmqtt_publish(&this->client, lwmqtt_string(topic), message, props, this->timeout);
   if (this->_lastError != LWMQTT_SUCCESS) {
     // close connection
     this->close();
@@ -294,14 +296,15 @@ bool MQTTClient::connect(const char clientId[], const char username[], const cha
   return true;
 }
 
-bool MQTTClient::subscribe(const char topic[], int qos) {
+bool MQTTClient::subscribe(const char topic[], int qos, lwmqtt_properties_t props) {
   // return immediately if not connected
   if (!this->connected()) {
     return false;
   }
 
   // subscribe to topic
-  this->_lastError = lwmqtt_subscribe_one(&this->client, lwmqtt_string(topic), (lwmqtt_qos_t)qos, this->timeout);
+  lwmqtt_sub_options_t subopts = {.qos = (lwmqtt_qos_t)qos};
+  this->_lastError = lwmqtt_subscribe_one(&this->client, lwmqtt_string(topic), subopts, props, this->timeout);
   if (this->_lastError != LWMQTT_SUCCESS) {
     // close connection
     this->close();
@@ -312,14 +315,14 @@ bool MQTTClient::subscribe(const char topic[], int qos) {
   return true;
 }
 
-bool MQTTClient::unsubscribe(const char topic[]) {
+bool MQTTClient::unsubscribe(const char topic[], lwmqtt_properties_t props) {
   // return immediately if not connected
   if (!this->connected()) {
     return false;
   }
 
   // unsubscribe from topic
-  this->_lastError = lwmqtt_unsubscribe_one(&this->client, lwmqtt_string(topic), this->timeout);
+  this->_lastError = lwmqtt_unsubscribe_one(&this->client, lwmqtt_string(topic), props, this->timeout);
   if (this->_lastError != LWMQTT_SUCCESS) {
     // close connection
     this->close();
@@ -368,14 +371,14 @@ bool MQTTClient::connected() {
   return this->netClient != nullptr && this->netClient->connected() == 1 && this->_connected;
 }
 
-bool MQTTClient::disconnect() {
+bool MQTTClient::disconnect(uint8_t reason, lwmqtt_properties_t props) {
   // return immediately if not connected anymore
   if (!this->connected()) {
     return false;
   }
 
   // cleanly disconnect
-  this->_lastError = lwmqtt_disconnect(&this->client, this->timeout);
+  this->_lastError = lwmqtt_disconnect(&this->client, reason, props, this->timeout);
 
   // close
   this->close();

src/MQTTClient.h

@@ -23,7 +23,8 @@ typedef struct {
 class MQTTClient;
 
 typedef void (*MQTTClientCallbackSimple)(String &topic, String &payload);
-typedef void (*MQTTClientCallbackAdvanced)(MQTTClient *client, char topic[], char bytes[], int length);
+typedef void (*MQTTClientCallbackAdvanced)(MQTTClient *client, char topic[], char bytes[], int length,
+                                           lwmqtt_serialized_properties_t props);
 
 typedef struct {
   MQTTClient *client = nullptr;
@@ -50,7 +51,7 @@ class MQTTClient {
   lwmqtt_arduino_network_t network = {nullptr};
   lwmqtt_arduino_timer_t timer1 = {0, nullptr};
   lwmqtt_arduino_timer_t timer2 = {0, nullptr};
-  lwmqtt_client_t client = {0};
+    lwmqtt_client_t client = {LWMQTT_MQTT311, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
 
   bool _connected = false;
   lwmqtt_return_code_t _returnCode = (lwmqtt_return_code_t)0;
@@ -61,8 +62,9 @@ class MQTTClient {
 
   ~MQTTClient();
 
-  void begin(const char hostname[], Client &client) { this->begin(hostname, 1883, client); }
-  void begin(const char hostname[], int port, Client &client);
+  void begin(const char hostname[], Client &client) { this->begin(hostname, 1883, client, LWMQTT_MQTT311); }
+  void begin(const char hostname[], int port, Client &client) { this->begin(hostname, port, client, LWMQTT_MQTT311); }
+  void begin(const char hostname[], int port, Client &client, lwmqtt_protocol_t protocol);
 
   void onMessage(MQTTClientCallbackSimple cb);
   void onMessageAdvanced(MQTTClientCallbackAdvanced cb);
@@ -91,6 +93,10 @@ class MQTTClient {
   bool publish(const String &topic, const String &payload, bool retained, int qos) {
     return this->publish(topic.c_str(), payload.c_str(), retained, qos);
   }
+  bool publish(const String &topic, const String &payload, bool retained, int qos, lwmqtt_properties_t props) {
+    return this->publish(topic.c_str(), payload.c_str(), payload.length(), retained, qos, props);
+  }
+
   bool publish(const char topic[], const String &payload) { return this->publish(topic, payload.c_str()); }
   bool publish(const char topic[], const String &payload, bool retained, int qos) {
     return this->publish(topic, payload.c_str(), retained, qos);
@@ -104,23 +110,41 @@ class MQTTClient {
   bool publish(const char topic[], const char payload[], int length) {
     return this->publish(topic, payload, length, false, 0);
   }
-  bool publish(const char topic[], const char payload[], int length, bool retained, int qos);
+  bool publish(const char topic[], const char payload[], int length, bool retained, int qos) {
+    lwmqtt_properties_t props = lwmqtt_empty_props;
+    return this->publish(topic, payload, length, retained, qos, props);
+  }
+
+  bool publish(const char topic[], const char payload[], int length, bool retained, int qos, lwmqtt_properties_t props);
 
   bool subscribe(const String &topic) { return this->subscribe(topic.c_str()); }
   bool subscribe(const String &topic, int qos) { return this->subscribe(topic.c_str(), qos); }
   bool subscribe(const char topic[]) { return this->subscribe(topic, 0); }
-  bool subscribe(const char topic[], int qos);
+  bool subscribe(const char topic[], int qos) {
+    lwmqtt_properties_t props = lwmqtt_empty_props;
+    return this->subscribe(topic, qos, props);
+  }
+  bool subscribe(const char topic[], int qos, lwmqtt_properties_t props);
 
   bool unsubscribe(const String &topic) { return this->unsubscribe(topic.c_str()); }
-  bool unsubscribe(const char topic[]);
+  bool unsubscribe(const char topic[]) {
+    lwmqtt_properties_t props = lwmqtt_empty_props;
+    return this->unsubscribe(topic, props);
+  }
+  bool unsubscribe(const char topic[], lwmqtt_properties_t props);
 
   bool loop();
   bool connected();
 
   lwmqtt_err_t lastError() { return this->_lastError; }
   lwmqtt_return_code_t returnCode() { return this->_returnCode; }
 
-  bool disconnect();
+  bool disconnect() {
+    lwmqtt_properties_t props = lwmqtt_empty_props;
+    return this->disconnect(0, props);
+  }
+
+  bool disconnect(uint8_t reason, lwmqtt_properties_t props);
 
  private:
   void close();