feat: connect to mqtt over websocket

lib/MQTTWebSocketStreamClient/MQTTWebSocketClient.cpp

@@ -0,0 +1,83 @@
+/*
+ * This program is free software; you can use it, redistribute it
+ * and / or modify it under the terms of the GNU General Public License
+ * (GPL) as published by the Free Software Foundation; either version 3
+ * of the License or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ *  WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program, in a file called gpl.txt or license.txt.
+ *  If not, write to the Free Software Foundation Inc.,
+ *  59 Temple Place - Suite 330, Boston, MA  02111-1307 USA
+ */
+
+#include "MQTTWebSocketClient.h"
+#include "b64.h"
+
+MQTTWebSocketClient::MQTTWebSocketClient(Client &client, const char *host, uint16_t port): WebSocketClient(
+    client, host, port) {
+}
+
+int MQTTWebSocketClient::connect(const IPAddress &ip, uint16_t port) {
+    return WebSocketClient::connect(ip, port);
+}
+
+int MQTTWebSocketClient::begin(const char *aPath, const char *protocol) {
+    // start the GET request
+    beginRequest();
+    connectionKeepAlive();
+    int status = get(aPath);
+
+    if (status == 0) {
+        uint8_t randomKey[16];
+        char base64RandomKey[25];
+
+        // create a random key for the connection upgrade
+        for (int i = 0; i < static_cast<int>(sizeof(randomKey)); i++) {
+            randomKey[i] = random(0x01, 0xff);
+        }
+        memset(base64RandomKey, 0x00, sizeof(base64RandomKey));
+        b64_encode(randomKey, sizeof(randomKey), reinterpret_cast<unsigned char *>(base64RandomKey),
+                   sizeof(base64RandomKey));
+
+        // start the connection upgrade sequence
+        sendHeader("Upgrade", "websocket");
+        sendHeader("Connection", "Upgrade");
+        sendHeader("Sec-WebSocket-Key", base64RandomKey);
+        sendHeader("Sec-WebSocket-Version", "13");
+        if (protocol) {
+            // is required by Mosquitto broker
+            sendHeader("Sec-WebSocket-Protocol", protocol);
+        }
+        endRequest();
+
+        status = responseStatusCode();
+
+        if (status > 0) {
+            skipResponseHeaders();
+        }
+    }
+
+    // iRxSize = 0;
+
+    // status code of 101 means success
+    return (status == 101) ? 0 : status;
+}
+
+int MQTTWebSocketClient::begin(const String &aPath, const char *protocol) {
+    return begin(aPath.c_str(), protocol);
+}
+
+bool MQTTWebSocketClient::flush(unsigned int maxWaitMs) {
+    WebSocketClient::flush();
+    return true;
+}
+
+bool MQTTWebSocketClient::stop(unsigned int maxWaitMs) {
+    WebSocketClient::stop();
+    return true;
+}

lib/MQTTWebSocketStreamClient/MQTTWebSocketClient.h

@@ -0,0 +1,34 @@
+/*
+ * This program is free software; you can use it, redistribute it
+ * and / or modify it under the terms of the GNU General Public License
+ * (GPL) as published by the Free Software Foundation; either version 3
+ * of the License or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ *  WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program, in a file called gpl.txt or license.txt.
+ *  If not, write to the Free Software Foundation Inc.,
+ *  59 Temple Place - Suite 330, Boston, MA  02111-1307 USA
+ */
+#pragma once
+
+#include <ArduinoHttpClient.h>
+
+class MQTTWebSocketClient : public WebSocketClient {
+public:
+    MQTTWebSocketClient(Client &client, const char *host, uint16_t port);
+
+    int connect(const IPAddress &ip, uint16_t port);
+
+    int begin(const char *aPath, const char *protocol = nullptr);
+
+    int begin(const String &aPath, const char *protocol = nullptr);
+
+    bool flush(unsigned int maxWaitMs = 0);
+
+    bool stop(unsigned int maxWaitMs = 0);
+};

lib/MQTTWebSocketStreamClient/MQTTWebSocketStreamClient.cpp

@@ -0,0 +1,96 @@
+/*
+ * This program is free software; you can use it, redistribute it
+ * and / or modify it under the terms of the GNU General Public License
+ * (GPL) as published by the Free Software Foundation; either version 3
+ * of the License or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ *  WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program, in a file called gpl.txt or license.txt.
+ *  If not, write to the Free Software Foundation Inc.,
+ *  59 Temple Place - Suite 330, Boston, MA  02111-1307 USA
+ */
+
+#include "MQTTWebSocketStreamClient.h"
+
+MQTTWebSocketStreamClient::MQTTWebSocketStreamClient(MQTTWebSocketClient &webSocketClient, const char *path,
+                                                     const char *protocol) {
+    this->webSocketClient = &webSocketClient;
+    this->path = path;
+    this->protocol = protocol;
+}
+
+int MQTTWebSocketStreamClient::connect(const IPAddress ip, uint16_t port) {
+    webSocketClient->begin(path, this->protocol);
+    return 1;
+}
+
+int MQTTWebSocketStreamClient::connect(const char *host, uint16_t port) {
+    webSocketClient->begin(path, this->protocol);
+    return 1;
+}
+
+size_t MQTTWebSocketStreamClient::write(uint8_t b) {
+    if (!connected())
+        return -1;
+    return write(&b, 1);
+}
+
+size_t MQTTWebSocketStreamClient::write(const uint8_t *buf, size_t size) {
+    if (!connected())
+        return -1;
+    webSocketClient->beginMessage(TYPE_BINARY);
+    webSocketClient->write(buf, size);
+    webSocketClient->endMessage();
+    return size;
+}
+
+int MQTTWebSocketStreamClient::available() {
+    if (!connected())
+        return 0;
+    if (webSocketClient->available() == 0)
+        webSocketClient->parseMessage();
+    return webSocketClient->available();
+}
+
+int MQTTWebSocketStreamClient::read() {
+    if (!connected() || !available())
+        return -1;
+    return webSocketClient->read();
+}
+
+int MQTTWebSocketStreamClient::read(uint8_t *buf, size_t size) {
+    if (!connected() || !available())
+        return -1;
+    return webSocketClient->read(buf, size);
+}
+
+int MQTTWebSocketStreamClient::peek() {
+    if (!connected() || !available())
+        return -1;
+    return webSocketClient->peek();
+}
+
+void MQTTWebSocketStreamClient::flush() {
+    if (!connected())
+        return;
+    webSocketClient->flush();
+}
+
+void MQTTWebSocketStreamClient::stop() {
+    if (!connected())
+        return;
+    webSocketClient->stop();
+}
+
+uint8_t MQTTWebSocketStreamClient::connected() {
+    return webSocketClient->connected();
+}
+
+MQTTWebSocketStreamClient::operator bool() {
+    return webSocketClient != nullptr;
+}

lib/MQTTWebSocketStreamClient/MQTTWebSocketStreamClient.h

@@ -0,0 +1,53 @@
+/*
+ * This program is free software; you can use it, redistribute it
+ * and / or modify it under the terms of the GNU General Public License
+ * (GPL) as published by the Free Software Foundation; either version 3
+ * of the License or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ *  WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program, in a file called gpl.txt or license.txt.
+ *  If not, write to the Free Software Foundation Inc.,
+ *  59 Temple Place - Suite 330, Boston, MA  02111-1307 USA
+ */
+#pragma once
+
+#include "MQTTWebSocketClient.h"
+#include <Client.h>
+
+class MQTTWebSocketStreamClient : public Client {
+    MQTTWebSocketClient *webSocketClient;
+    const char *path;
+    const char *protocol;
+
+public:
+    MQTTWebSocketStreamClient(MQTTWebSocketClient &webSocketClient, const char *path, const char *protocol = "mqtt");
+
+    int connect(IPAddress ip, uint16_t port) override;
+
+    int connect(const char *host, uint16_t port) override;
+
+    size_t write(uint8_t b) override;
+
+    size_t write(const uint8_t *buf, size_t size) override;
+
+    int available() override;
+
+    int read() override;
+
+    int read(uint8_t *buf, size_t size) override;
+
+    int peek() override;
+
+    void flush() override;
+
+    void stop() override;
+
+    uint8_t connected() override;
+
+    explicit operator bool() override;
+};

lib/MQTTWebSocketStreamClient/README.md

@@ -0,0 +1,4 @@
+# MQTTWebSocketStreamClient
+
+This was adopted from areve's [WebSocketStreamClient](https://github.com/areve/WebSocketStreamClient) and updated to
+work with Mosquitto broker. 

platformio.ini

@@ -37,6 +37,7 @@ lib_deps =
     https://github.com/lewisxhe/TinyGSM
     powerbroker2/ELMDuino @ ^3.3.0
     PubSubClient @ ^2.8
+    ArduinoHttpClient @ ^0.6.1
     ArduinoJson @ ^7.1.0
 ;    TinyGSM @ ^0.12.0
     vshymanskyy/StreamDebugger @ ^1.0.1

src/main.cpp

@@ -50,8 +50,6 @@ HTTPServer server(80);
 
 #define DEBUG_PORT Serial
 
-// HTTPServer server(80);
-
 // #define DUMP_AT_COMMANDS
 
 #ifdef DUMP_AT_COMMANDS
@@ -62,9 +60,7 @@ GSM gsm(debugger);
 GSM gsm(SerialAT);
 #endif
 
-PubSubClient mqttClient(gsm.client);
-
-MQTT mqtt(mqttClient);
+MQTT mqtt(gsm.client);
 
 std::atomic_bool wifiAPStarted{false};
 std::atomic_bool wifiAPInUse{false};
@@ -830,7 +826,8 @@ void outputTask(void *parameters) {
                     Settings.getMQTTHostname().c_str(),
                     Settings.getMQTTPort(),
                     Settings.getMQTTUsername().c_str(),
-                    Settings.getMQTTPassword().c_str()
+                    Settings.getMQTTPassword().c_str(),
+                    static_cast<mqttProtocol>(Settings.getMQTTProtocol())
                 );
             } else {
                 mqttSendData();

src/mqtt.cpp

@@ -15,6 +15,9 @@
  *  59 Temple Place - Suite 330, Boston, MA  02111-1307 USA
  */
 #include "mqtt.h"
+#include <ArduinoJson.h>
+#include <helper.h>
+#include <MQTTWebSocketStreamClient.h>
 
 std::string MQTT::createNodeId(const std::string &topic) {
     auto splitPos = topic.find_last_of('/');
@@ -25,15 +28,23 @@ std::string MQTT::createFieldTopic(const std::string &field) const {
     return stripChars((!identifier.empty() ? identifier : maintopic) + "_" + field);
 }
 
-MQTT::MQTT(const PubSubClient &client) {
-    mqtt = client;
+MQTT::MQTT(Client &client): wsClient(nullptr), wsStreamClient(nullptr) {
+    this->client = &client;
     mqtt.setKeepAlive(60);
     mqtt.setBufferSize(MQTT_MAX_PACKET_SIZE);
 }
 
 void MQTT::connect(const char *clientId, const char *broker, const unsigned int port, const char *username,
-                   const char *password) {
-    mqtt.setServer(broker, port);
+                   const char *password, mqttProtocol protocol) {
+    if (protocol == USE_MQTT) {
+        mqtt.setClient(*client);
+        mqtt.setServer(broker, port);
+    } else {
+        wsClient = new MQTTWebSocketClient(*client, broker, port);
+        wsStreamClient = new MQTTWebSocketStreamClient(*wsClient, "mqtt");
+        mqtt.setClient(*wsStreamClient);
+    }
+
     while (!mqtt.connected()) {
         Serial.printf("The client %s connects to the MQTT broker...", clientId);
         std::string lwtTopic = maintopic + "/" + createFieldTopic(LWT_TOPIC);