Support for RadioHead's RHReliableDatagram on ESP32

.gitignore

@@ -1 +1,8 @@
 .idea/
+.pio
+.vscode/.browse.c_cpp.db*
+.vscode/c_cpp_properties.json
+.vscode/extensions.json
+.vscode/launch.json
+.vscode/ipch
+.history

MqttSnClient.h

@@ -23,7 +23,7 @@ struct topic_registration {
     uint8_t granted_qos;
 };
 
-#ifdef ESP8266
+#if defined(ESP8266) || defined(ESP32)
 #include <functional>
 #define MQTT_SN_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, uint16_t, bool retain)> callback
 #else
@@ -76,6 +76,7 @@ class MqttSnClient {
 
     void notify_socket_disconnected() {
         this->socket_disconnected = true;
+        this->mqttsn_connected = false;
     }
 
     void set_await_message(message_type msg_type) {
@@ -105,13 +106,13 @@ class MqttSnClient {
         this->callback = callback;
     }
 
-    bool connect(device_address *address, const char *client_id, uint16_t duration) {
-        // global verwalten
-        uint8_t retries = 2;
+    bool connect(device_address *address, const char *client_id, uint16_t duration, uint8_t retries = 2, uint8_t timeout = 5, uint8_t retry_pause = 10) {
+
+        socket_disconnected = false;
         memcpy(&this->gw_address, address, sizeof(device_address));
 
         for (uint8_t tries = 0; tries < retries; tries++) {
-            system.set_heartbeat(5 * 1000);
+            system.set_heartbeat(timeout * 1000);
             mqttSnMessageHandler.send_connect(address, client_id, duration);
             this->set_await_message(MQTTSN_CONNACK);
             while (!mqttsn_connected) {
@@ -129,7 +130,7 @@ class MqttSnClient {
                     return false;
                 }
             }
-            system.sleep((tries + 1) * 10 * 1000);
+            system.sleep((tries + 1) * retry_pause * 1000);
         }
         // timeout - take another gateway
         return false;
@@ -275,6 +276,7 @@ class MqttSnClient {
     }
 
     void notify_socket_connected() {
+        Serial.println("\nSocket disconnected\n");
         this->socket_disconnected = false;
     }
 

RHDatagramSocket.h

@@ -7,6 +7,8 @@
 #ifndef ARDUINO_MQTT_SN_CLIENT_RHDATAGRAMSOCKET_H
 #define ARDUINO_MQTT_SN_CLIENT_RHDATAGRAMSOCKET_H
 
+#include <RHDatagram.h>
+
 #include "MqttSnMessageHandler.h"
 #include "SocketInterface.h"
 

RHReliableDatagramSocket.h

@@ -0,0 +1,130 @@
+/*
+* The MIT License (MIT)
+*
+* Copyright (C) 2018 Gabriel Nikol
+*/
+
+#ifndef ARDUINO_MQTT_SN_CLIENT_RHDRELIABLEATAGRAMSOCKET_H
+#define ARDUINO_MQTT_SN_CLIENT_RHRELIABLEDATAGRAMSOCKET_H
+
+#include <RHReliableDatagram.h>
+
+#include "MqttSnMessageHandler.h"
+#include "SocketInterface.h"
+
+class RHReliableDatagramSocket : SocketInterface {
+private:
+    RHReliableDatagram &rhDatagram;
+
+    device_address own_address;
+    device_address broadcast_address;
+
+    bool reliable = true;
+public:
+    RHReliableDatagramSocket(RHReliableDatagram &rhDatagram) : rhDatagram(rhDatagram) {};
+
+    bool begin() override {
+        own_address = device_address(rhDatagram.thisAddress(), 0, 0, 0, 0, 0);
+        broadcast_address = device_address(RH_BROADCAST_ADDRESS, 0, 0, 0, 0, 0);
+        return rhDatagram.init();
+    }
+
+    device_address *getBroadcastAddress() override {
+        return &broadcast_address;
+    }
+
+    device_address *getAddress() override {
+        return &own_address;
+    }
+
+    uint8_t getMaximumMessageLength() override {
+        return RH_MAX_MESSAGE_LEN;
+    }
+
+    bool send(device_address *destination, uint8_t *bytes, uint16_t bytes_len) override {
+        return send(destination, bytes, bytes_len, UINT8_MAX);
+    }
+
+    bool send(device_address *destination, uint8_t *bytes, uint16_t bytes_len, uint8_t signal_strength) {
+        bool sendOk;
+
+        if (reliable) {
+            sendOk = rhDatagram.sendtoWait(bytes, bytes_len, destination->bytes[0]);
+        }
+        else {
+            sendOk = rhDatagram.sendto(bytes, bytes_len, destination->bytes[0]);
+        }
+        rhDatagram.waitPacketSent();
+        rhDatagram.available(); // put it back to receive mode
+        return sendOk;
+    }
+
+
+
+    bool receive(device_address *source, uint8_t *bytes, uint16_t bytes_len) {
+        uint8_t len = bytes_len;
+        if (reliable) {
+            bool ok = rhDatagram.recvfromAck(bytes, &len, &(source->bytes[0]));
+            return ok;
+        }
+        else {
+            return rhDatagram.recvfrom(bytes, &len, &(source->bytes[0]));
+        }
+    }
+
+
+    /**
+     * Sets whether or not to use RadioHead's "reliable" acknowledged protocol, or its 
+     * normal one.  The default is "not reliable"
+     */
+    void setReliableProtocol(bool isReliable) { this->reliable = isReliable; }
+
+    bool isReliable() { return reliable; }
+
+
+    bool loop() override {
+
+        if (rhDatagram.available()) {
+
+            device_address receive_address;
+            memset(&receive_address, 0x0, sizeof(device_address));
+
+            uint8_t receive_buffer[RH_MAX_MESSAGE_LEN];
+            memset(&receive_buffer, 0x0, RH_MAX_MESSAGE_LEN);
+
+            uint8_t receive_buffer_len = sizeof(receive_buffer);
+
+            if (receive(&receive_address, receive_buffer, receive_buffer_len)) {
+
+                if (mqttSnMessageHandler != nullptr) {
+                    mqttSnMessageHandler->receiveData(&receive_address, receive_buffer);
+                }
+
+                if (transmissionProtocolUartBridge != nullptr) {
+                    transmissionProtocolUartBridge->receiveData(&receive_address, receive_buffer, receive_buffer_len);
+                }
+
+            }
+        }
+
+        return true;
+    }
+
+    template<class RHReliableDatagramSocket>
+    void setMqttSnMessageHandler(
+            MqttSnMessageHandler<RHReliableDatagramSocket> *mqttSnMessageHandler) {
+        this->mqttSnMessageHandler = mqttSnMessageHandler;
+    };
+
+    template<class RHReliableDatagramSocket>
+    void setTransmissionProtocolUartBridge(
+            TransmissionProtocolUartBridge<RHReliableDatagramSocket> *transmissionProtocolUartBridge) {
+        this->transmissionProtocolUartBridge = transmissionProtocolUartBridge;
+    };
+
+private:
+    MqttSnMessageHandler<RHReliableDatagramSocket> *mqttSnMessageHandler = nullptr;
+    TransmissionProtocolUartBridge<RHReliableDatagramSocket> *transmissionProtocolUartBridge = nullptr;
+};
+
+#endif //ARDUINO_MQTT_SN_CLIENT_RHRELIABLEDATAGRAMSOCKET_H

System.cpp

@@ -36,7 +36,7 @@ void System::sleep(uint32_t duration) {
 }
 
 void System::exit() {
-#if defined(ESP8266)
+#if defined(ESP8266) || defined(ESP32)
     ESP.restart();
 #elif defined(ARDUINO_AVR_UNO) || defined(ARDUINO_AVR_YUN) || defined(ARDUINO_AVR_MEGA2560)
     asm volatile ("  jmp 0");

TransmissionProtocolUartBridge.h

@@ -15,7 +15,7 @@
 #include <Stream.h>
 
 
-#if defined(ESP8266)
+#if defined(ESP8266) || defined(ESP32)
 #include <cerrno>
 #include <climits>
 #endif
@@ -178,7 +178,7 @@ class TransmissionProtocolUartBridge {
     }
 
     void resetChip() {
-#if defined(ESP8266)
+#if defined(ESP8266) || defined(ESP32)
         stream->print(F("OK RESET\n"));
         delay(500);
         ESP.restart();

global_defines.cpp

@@ -0,0 +1,14 @@
+#include <Arduino.h>
+
+#include "global_defines.h"
+
+void printDeviceAddress(device_address *address) {
+    for (uint8_t i = 0; i < sizeof(device_address); i++) {
+        if (i == sizeof(device_address) - 1) {
+            Serial.print(address->bytes[i]);
+        } else {
+            Serial.print(address->bytes[i]);
+            Serial.print(", ");
+        }
+    }
+}

global_defines.h

@@ -31,15 +31,7 @@ struct device_address {
     }
 };
 
-void printDeviceAddress(device_address *address) {
-    for (uint8_t i = 0; i < sizeof(device_address); i++) {
-        if (i == sizeof(device_address) - 1) {
-            Serial.print(address->bytes[i]);
-        } else {
-            Serial.print(address->bytes[i]);
-            Serial.print(", ");
-        }
-    }
-}
+
+extern void printDeviceAddress(device_address *address);
 
 #endif //ARDUINO_MQTTSN_CLIENT_GLOBAL_DEFINES_H

mqttsn_messages.h

@@ -195,7 +195,7 @@ struct msg_willtopic : public message_header {
     char will_topic[252];
 
     msg_willtopic(const char *willtopic, int8_t qos, bool retain) {
-        memset(this, 0, sizeof(this));
+        memset(this, 0, sizeof(*this));
         length = 3 + strlen(willtopic) + 1;
         type = MQTTSN_WILLTOPIC;
         if (retain) {
@@ -215,7 +215,7 @@ struct msg_willmsg : public message_header {
     uint8_t willmsg[253];
 
     msg_willmsg(const uint8_t *s_data, uint8_t s_data_len) {
-        memset(this, 0, sizeof(this));
+        memset(this, 0, sizeof(*this));
         this->length = ((uint8_t) 2) + s_data_len;
         memcpy(&willmsg, s_data, s_data_len);
     }
@@ -279,7 +279,7 @@ struct msg_publish : public message_header {
 
     msg_publish(bool dup, int8_t qos, bool retain, bool short_topic, uint16_t topic_id, uint16_t msg_id,
                 const uint8_t *s_data, uint8_t s_data_len) : topic_id(topic_id), message_id(msg_id) {
-        memset(this, 0, sizeof(this));
+        memset(this, 0, sizeof(*this));
         this->length = ((uint8_t) 7) + s_data_len;
         this->type = MQTTSN_PUBLISH;
         this->flags = 0x0;
@@ -349,7 +349,7 @@ struct msg_subscribe_shorttopic : public msg_subscribe {
     uint16_t topic_id;
 
     msg_subscribe_shorttopic(bool short_topic, uint16_t topic_id, uint16_t msg_id, uint8_t qos, bool dup) {
-        memset(this, 0, sizeof(this));
+        memset(this, 0, sizeof(*this));
 
         this->length = 7;
         this->type = MQTTSN_SUBSCRIBE;
@@ -382,7 +382,7 @@ struct msg_subscribe_topicname : public msg_subscribe {
     char topic_name[250];
 
     msg_subscribe_topicname(const char *topic_name, uint16_t msg_id, uint8_t qos, bool dup) {
-        memset(this, 0, sizeof(this));
+        memset(this, 0, sizeof(*this));
         this->length = (uint8_t) (5 + strlen(topic_name) + 1);
         this->type = MQTTSN_SUBSCRIBE;
         if (dup) {

platformio.ini

@@ -0,0 +1,34 @@
+; PlatformIO Project Configuration File
+;
+;   Build options: build flags, source filter
+;   Upload options: custom upload port, speed and extra flags
+;   Library options: dependencies, extra library storages
+;   Advanced options: extra scripting
+;
+; Please visit documentation for the other options and examples
+; https://docs.platformio.org/page/projectconf.html
+
+[platformio]
+src_dir = .
+
+
+
+[common_env_data]
+build_flags = 
+   -Wall
+   -Wno-reorder
+   -Wno-parentheses
+
+[env:heltec_wifi_kit_32]
+platform = espressif32
+board = heltec_wifi_kit_32
+framework = arduino
+
+build_flags =
+    ${common_env_data.build_flags}
+
+;lib_deps =
+;    RadioHead
+
+src_filter=+<*.h> +<*.cpp> -<.git/> -<.svn/> -<examples/> -<test/> 
+monitor_speed=115200