first commit using PsychicHttp

lib/espMqttClient/src/Config.h

@@ -8,6 +8,12 @@ the LICENSE file.
 
 #pragma once
 
+#ifndef TASMOTA_SDK
+#define EMC_CLIENT_SECURE
+#else
+#undef EMC_CLIENT_SECURE
+#endif
+
 #ifndef EMC_TX_TIMEOUT
 #define EMC_TX_TIMEOUT 2000
 #endif
@@ -53,6 +59,10 @@ the LICENSE file.
 #define EMC_TASK_STACK_SIZE 5120
 #endif
 
+#ifndef EMC_MULTIPLE_CALLBACKS
+#define EMC_MULTIPLE_CALLBACKS 0
+#endif
+
 #ifndef EMC_USE_WATCHDOG
 #define EMC_USE_WATCHDOG 0
 #endif

lib/espMqttClient/src/MqttClient.h

@@ -98,7 +98,6 @@ class MqttClient {
     uint32_t        _timeout;
 
     // state is protected to allow state changes by the transport system, defined in child classes
-    // eg. to allow AsyncTCP
     enum class State {
         disconnected       = 0,
         connectingTcp1     = 1,

lib/espMqttClient/src/MqttClientSetup.h

@@ -11,6 +11,11 @@ the LICENSE file.
 
 #pragma once
 
+#if EMC_MULTIPLE_CALLBACKS
+#include <list>
+#include <utility>
+#endif
+
 #include "MqttClient.h"
 
 template <typename T>
@@ -73,36 +78,128 @@ class MqttClientSetup : public MqttClient {
     return static_cast<T&>(*this);
   }
 
-  T& onConnect(espMqttClientTypes::OnConnectCallback callback) {
+  T& onConnect(espMqttClientTypes::OnConnectCallback callback, uint32_t id = 0) {
+    #if EMC_MULTIPLE_CALLBACKS
+    _onConnectCallbacks.emplace_back(callback, id);
+    #else
+    (void) id;
     _onConnectCallback = callback;
+    #endif
     return static_cast<T&>(*this);
   }
 
-  T& onDisconnect(espMqttClientTypes::OnDisconnectCallback callback) {
+  T& onDisconnect(espMqttClientTypes::OnDisconnectCallback callback, uint32_t id = 0) {
+    #if EMC_MULTIPLE_CALLBACKS
+    _onDisconnectCallbacks.emplace_back(callback, id);
+    #else
+    (void) id;
     _onDisconnectCallback = callback;
+    #endif
     return static_cast<T&>(*this);
   }
 
-  T& onSubscribe(espMqttClientTypes::OnSubscribeCallback callback) {
+  T& onSubscribe(espMqttClientTypes::OnSubscribeCallback callback, uint32_t id = 0) {
+    #if EMC_MULTIPLE_CALLBACKS
+    _onSubscribeCallbacks.emplace_back(callback, id);
+    #else
+    (void) id;
     _onSubscribeCallback = callback;
+    #endif
     return static_cast<T&>(*this);
   }
 
-  T& onUnsubscribe(espMqttClientTypes::OnUnsubscribeCallback callback) {
+  T& onUnsubscribe(espMqttClientTypes::OnUnsubscribeCallback callback, uint32_t id = 0) {
+    #if EMC_MULTIPLE_CALLBACKS
+    _onUnsubscribeCallbacks.emplace_back(callback, id);
+    #else
+    (void) id;
     _onUnsubscribeCallback = callback;
+    #endif
     return static_cast<T&>(*this);
   }
 
-  T& onMessage(espMqttClientTypes::OnMessageCallback callback) {
+  T& onMessage(espMqttClientTypes::OnMessageCallback callback, uint32_t id = 0) {
+    #if EMC_MULTIPLE_CALLBACKS
+    _onMessageCallbacks.emplace_back(callback, id);
+    #else
+    (void) id;
     _onMessageCallback = callback;
+    #endif
     return static_cast<T&>(*this);
   }
 
-  T& onPublish(espMqttClientTypes::OnPublishCallback callback) {
+  T& onPublish(espMqttClientTypes::OnPublishCallback callback, uint32_t id = 0) {
+    #if EMC_MULTIPLE_CALLBACKS
+    _onPublishCallbacks.emplace_back(callback, id);
+    #else
+    (void) id;
     _onPublishCallback = callback;
+    #endif
     return static_cast<T&>(*this);
   }
 
+  #if EMC_MULTIPLE_CALLBACKS
+  T& removeOnConnect(uint32_t id) {
+    for (auto it = _onConnectCallbacks.begin(); it != _onConnectCallbacks.end(); ++it) {
+      if (it->second == id) {
+        _onConnectCallbacks.erase(it);
+        break;
+      }
+    }
+    return static_cast<T&>(*this);
+  }
+
+  T& removeOnDisconnect(uint32_t id) {
+    for (auto it = _onDisconnectCallbacks.begin(); it != _onDisconnectCallbacks.end(); ++it) {
+      if (it->second == id) {
+        _onDisconnectCallbacks.erase(it);
+        break;
+      }
+    }
+    return static_cast<T&>(*this);
+  }
+
+  T& removeOnSubscribe(uint32_t id) {
+    for (auto it = _onSubscribeCallbacks.begin(); it != _onSubscribeCallbacks.end(); ++it) {
+      if (it->second == id) {
+        _onSubscribeCallbacks.erase(it);
+        break;
+      }
+    }
+    return static_cast<T&>(*this);
+  }
+
+  T& removeOnUnsubscribe(uint32_t id) {
+    for (auto it = _onUnsubscribeCallbacks.begin(); it != _onUnsubscribeCallbacks.end(); ++it) {
+      if (it->second == id) {
+        _onUnsubscribeCallbacks.erase(it);
+        break;
+      }
+    }
+    return static_cast<T&>(*this);
+  }
+
+  T& removeOnMessage(uint32_t id) {
+    for (auto it = _onMessageCallbacks.begin(); it != _onMessageCallbacks.end(); ++it) {
+      if (it->second == id) {
+        _onMessageCallbacks.erase(it);
+        break;
+      }
+    }
+    return static_cast<T&>(*this);
+  }
+
+  T& removeOnPublish(uint32_t id) {
+    for (auto it = _onPublishCallbacks.begin(); it != _onPublishCallbacks.end(); ++it) {
+      if (it->second == id) {
+        _onPublishCallbacks.erase(it);
+        break;
+      }
+    }
+    return static_cast<T&>(*this);
+  }
+  #endif
+
   /*
   T& onError(espMqttClientTypes::OnErrorCallback callback) {
     _onErrorCallback = callback;
@@ -112,5 +209,37 @@ class MqttClientSetup : public MqttClient {
 
  protected:
   explicit MqttClientSetup(espMqttClientTypes::UseInternalTask useInternalTask, uint8_t priority = 1, uint8_t core = 1)
-  : MqttClient(useInternalTask, priority, core) {}
+  : MqttClient(useInternalTask, priority, core) {
+    #if EMC_MULTIPLE_CALLBACKS
+    _onConnectCallback = [this](bool sessionPresent) {
+      for (auto callback : _onConnectCallbacks) if (callback.first) callback.first(sessionPresent);
+    };
+    _onDisconnectCallback = [this](espMqttClientTypes::DisconnectReason reason) {
+      for (auto callback : _onDisconnectCallbacks) if (callback.first) callback.first(reason);
+    };
+    _onSubscribeCallback = [this](uint16_t packetId, const espMqttClientTypes::SubscribeReturncode* returncodes, size_t len) {
+      for (auto callback : _onSubscribeCallbacks) if (callback.first) callback.first(packetId, returncodes, len);
+    };
+    _onUnsubscribeCallback = [this](int16_t packetId) {
+      for (auto callback : _onUnsubscribeCallbacks) if (callback.first) callback.first(packetId);
+    };
+    _onMessageCallback = [this](const espMqttClientTypes::MessageProperties& properties, const char* topic, const uint8_t* payload, size_t len, size_t index, size_t total) {
+      for (auto callback : _onMessageCallbacks) if (callback.first) callback.first(properties, topic, payload, len, index, total);
+    };
+    _onPublishCallback = [this](uint16_t packetId) {
+      for (auto callback : _onPublishCallbacks) if (callback.first) callback.first(packetId);
+    };
+    #else
+    // empty
+    #endif
+  }
+
+  #if EMC_MULTIPLE_CALLBACKS
+  std::list<std::pair<espMqttClientTypes::OnConnectCallback, uint32_t>> _onConnectCallbacks;
+  std::list<std::pair<espMqttClientTypes::OnDisconnectCallback, uint32_t>> _onDisconnectCallbacks;
+  std::list<std::pair<espMqttClientTypes::OnSubscribeCallback, uint32_t>> _onSubscribeCallbacks;
+  std::list<std::pair<espMqttClientTypes::OnUnsubscribeCallback, uint32_t>> _onUnsubscribeCallbacks;
+  std::list<std::pair<espMqttClientTypes::OnMessageCallback, uint32_t>> _onMessageCallbacks;
+  std::list<std::pair<espMqttClientTypes::OnPublishCallback, uint32_t>> _onPublishCallbacks;
+  #endif
 };

lib/espMqttClient/src/Transport/ClientAsync.cpp

@@ -1,58 +0,0 @@
-/*
-Copyright (c) 2022 Bert Melis. All rights reserved.
-
-This work is licensed under the terms of the MIT license.  
-For a copy, see <https://opensource.org/licenses/MIT> or
-the LICENSE file.
-*/
-
-#if defined(ARDUINO_ARCH_ESP8266) || defined(ARDUINO_ARCH_ESP32)
-
-#include "ClientAsync.h"
-
-namespace espMqttClientInternals {
-
-ClientAsync::ClientAsync()
-: client()
-, availableData(0)
-, bufData(nullptr) {
-  // empty
-}
-
-bool ClientAsync::connect(IPAddress ip, uint16_t port) {
-  return client.connect(ip, port);
-}
-
-bool ClientAsync::connect(const char* host, uint16_t port) {
-  return client.connect(host, port);
-}
-
-size_t ClientAsync::write(const uint8_t* buf, size_t size) {
-  return client.write(reinterpret_cast<const char*>(buf), size);
-}
-
-int ClientAsync::read(uint8_t* buf, size_t size) {
-  size_t willRead = std::min(size, availableData);
-  memcpy(buf, bufData, std::min(size, availableData));
-  if (availableData > size) {
-    emc_log_w("Buffer is smaller than available data: %zu - %zu", size, availableData);
-  }
-  availableData = 0;
-  return willRead;
-}
-
-void ClientAsync::stop() {
-  client.close(false);
-}
-
-bool ClientAsync::connected() {
-  return client.connected();
-}
-
-bool ClientAsync::disconnected() {
-  return client.disconnected();
-}
-
-}  // namespace espMqttClientInternals
-
-#endif

lib/espMqttClient/src/Transport/ClientAsync.h

@@ -1,44 +0,0 @@
-/*
-Copyright (c) 2022 Bert Melis. All rights reserved.
-
-This work is licensed under the terms of the MIT license.  
-For a copy, see <https://opensource.org/licenses/MIT> or
-the LICENSE file.
-*/
-
-#if defined(ARDUINO_ARCH_ESP8266) || defined(ARDUINO_ARCH_ESP32)
-
-#pragma once
-
-#if defined(ARDUINO_ARCH_ESP32)
-  #include "freertos/FreeRTOS.h"
-  #include <AsyncTCP.h>
-#elif defined(ARDUINO_ARCH_ESP8266)
-  #include <ESPAsyncTCP.h>
-#endif
-
-#include "Transport.h"
-#include "../Config.h"
-#include "../Logging.h"
-
-namespace espMqttClientInternals {
-
-class ClientAsync : public Transport {
- public:
-  ClientAsync();
-  bool connect(IPAddress ip, uint16_t port) override;
-  bool connect(const char* host, uint16_t port) override;
-  size_t write(const uint8_t* buf, size_t size) override;
-  int read(uint8_t* buf, size_t size) override;
-  void stop() override;
-  bool connected() override;
-  bool disconnected() override;
-
-  AsyncClient client;
-  size_t availableData;
-  uint8_t* bufData;
-};
-
-}  // namespace espMqttClientInternals
-
-#endif

lib/espMqttClient/src/Transport/ClientSecureSync.h

@@ -10,7 +10,12 @@ the LICENSE file.
 
 #if defined(ARDUINO_ARCH_ESP8266) || defined(ARDUINO_ARCH_ESP32)
 
+#include "../Config.h"
+#if defined(EMC_CLIENT_SECURE)
 #include <WiFiClientSecure.h>  // includes IPAddress
+#else
+#include <WiFiClient.h>
+#endif
 
 #include "Transport.h"
 
@@ -26,7 +31,11 @@ class ClientSecureSync : public Transport {
   void stop() override;
   bool connected() override;
   bool disconnected() override;
+#if defined(EMC_CLIENT_SECURE)
   WiFiClientSecure client;
+#else
+  WiFiClient client;
+#endif
 };
 
 }  // namespace espMqttClientInternals

lib/espMqttClient/src/espMqttClient.cpp

@@ -78,27 +78,37 @@ espMqttClientSecure::espMqttClientSecure(uint8_t priority, uint8_t core)
 }
 
 espMqttClientSecure& espMqttClientSecure::setInsecure() {
+#if defined(EMC_CLIENT_SECURE)
   _client.client.setInsecure();
+#endif
   return *this;
 }
 
 espMqttClientSecure& espMqttClientSecure::setCACert(const char* rootCA) {
+#if defined(EMC_CLIENT_SECURE)
   _client.client.setCACert(rootCA);
+#endif
   return *this;
 }
 
 espMqttClientSecure& espMqttClientSecure::setCertificate(const char* clientCa) {
+#if defined(EMC_CLIENT_SECURE)
   _client.client.setCertificate(clientCa);
+#endif
   return *this;
 }
 
 espMqttClientSecure& espMqttClientSecure::setPrivateKey(const char* privateKey) {
+#if defined(EMC_CLIENT_SECURE)
   _client.client.setPrivateKey(privateKey);
+#endif
   return *this;
 }
 
 espMqttClientSecure& espMqttClientSecure::setPreSharedKey(const char* pskIdent, const char* psKey) {
+#if defined(EMC_CLIENT_SECURE)
   _client.client.setPreSharedKey(pskIdent, psKey);
+#endif
   return *this;
 }
 

lib/espMqttClient/src/espMqttClientAsync.cpp

@@ -1,61 +0,0 @@
-/*
-Copyright (c) 2022 Bert Melis. All rights reserved.
-
-This work is licensed under the terms of the MIT license.  
-For a copy, see <https://opensource.org/licenses/MIT> or
-the LICENSE file.
-*/
-
-#if defined(ARDUINO_ARCH_ESP8266) || defined(ARDUINO_ARCH_ESP32)
-
-#include "espMqttClientAsync.h"
-
-espMqttClientAsync::espMqttClientAsync()
-: MqttClientSetup(espMqttClientTypes::UseInternalTask::NO)
-, _clientAsync() {
-  _transport = &_clientAsync;
-  _clientAsync.client.onConnect(onConnectCb, this);
-  _clientAsync.client.onDisconnect(onDisconnectCb, this);
-  _clientAsync.client.onData(onDataCb, this);
-  _clientAsync.client.onPoll(onPollCb, this);
-}
-
-bool espMqttClientAsync::connect() {
-  bool ret = MqttClient::connect();
-  loop();
-  return ret;
-}
-
-void espMqttClientAsync::_setupClient(espMqttClientAsync* c) {
-  (void)c;
-}
-
-void espMqttClientAsync::onConnectCb(void* a, AsyncClient* c) {
-  c->setNoDelay(true);
-  espMqttClientAsync* client = reinterpret_cast<espMqttClientAsync*>(a);
-  client->_state = MqttClient::State::connectingTcp2;
-  client->loop();
-}
-
-void espMqttClientAsync::onDataCb(void* a, AsyncClient* c, void* data, size_t len) {
-  (void)c;
-  espMqttClientAsync* client = reinterpret_cast<espMqttClientAsync*>(a);
-  client->_clientAsync.bufData = reinterpret_cast<uint8_t*>(data);
-  client->_clientAsync.availableData = len;
-  client->loop();
-}
-
-void espMqttClientAsync::onDisconnectCb(void* a, AsyncClient* c) {
-  (void)c;
-  espMqttClientAsync* client = reinterpret_cast<espMqttClientAsync*>(a);
-  client->_state = MqttClient::State::disconnectingTcp2;
-  client->loop();
-}
-
-void espMqttClientAsync::onPollCb(void* a, AsyncClient* c) {
-  (void)c;
-  espMqttClientAsync* client = reinterpret_cast<espMqttClientAsync*>(a);
-  client->loop();
-}
-
-#endif

lib/espMqttClient/src/espMqttClientAsync.h

@@ -1,36 +0,0 @@
-/*
-Copyright (c) 2022 Bert Melis. All rights reserved.
-
-API is based on the original work of Marvin Roger:
-https://github.com/marvinroger/async-mqtt-client
-
-This work is licensed under the terms of the MIT license.  
-For a copy, see <https://opensource.org/licenses/MIT> or
-the LICENSE file.
-*/
-
-#pragma once
-
-#if defined(ARDUINO_ARCH_ESP8266) || defined(ARDUINO_ARCH_ESP32)
-
-#include "Transport/ClientAsync.h"
-
-#include "MqttClientSetup.h"
-
-class espMqttClientAsync : public MqttClientSetup<espMqttClientAsync> {
- public:
-  espMqttClientAsync();
-  bool connect();
-
- protected:
-  espMqttClientInternals::ClientAsync _clientAsync;
-  static void _setupClient(espMqttClientAsync* c);
-  static void _disconnectClient(espMqttClientAsync* c);
-
-  static void onConnectCb(void* a, AsyncClient* c);
-  static void onDataCb(void* a, AsyncClient* c, void* data, size_t len);
-  static void onDisconnectCb(void* a, AsyncClient* c);
-  static void onPollCb(void* a, AsyncClient* c);
-};
-
-#endif