0.8.124

* improved MqTT `OnMessage` (threadsafe)
lumapu · Jun 6, 2024 · 034cbd5 · 034cbd5
1 parent 8d4ca49
commit 034cbd5
Show file tree

Hide file tree

Showing 4 changed files with 60 additions and 5 deletions.
diff --git a/src/CHANGES.md b/src/CHANGES.md
@@ -1,10 +1,13 @@
 # Development Changes
 
+## 0.8.124 - 2024-06-06
+* improved MqTT `OnMessage` (threadsafe)
+
 ## 0.8.123 - 2024-05-30
 * fix ESP8266, ESP32 static IP #1643 #1608
 * update MqTT library which enhances stability #1646
-* merge PR: MQTT JSON Payload pro Kanal und total, auswählbar #1541
-* add option to publish mqtt as json
+* merge PR: MqTT JSON Payload pro Kanal und total, auswählbar #1541
+* add option to publish MqTT as json
 * publish rssi not on ch0 any more, published on `topic/rssi`
 * add total power to index page (if multiple inverters are configured)
 * show device name in html title #1639

diff --git a/src/defines.h b/src/defines.h
@@ -13,7 +13,7 @@
 //-------------------------------------
 #define VERSION_MAJOR       0
 #define VERSION_MINOR       8
-#define VERSION_PATCH       123
+#define VERSION_PATCH       124
 //-------------------------------------
 typedef struct {
     uint8_t ch;

diff --git a/src/plugins/MaxPower.h b/src/plugins/MaxPower.h
@@ -50,7 +50,7 @@ class MaxPower {
                 if((mValues[i].first + mMaxDiff) >= *mTs)
                     val += mValues[i].second;
                 else if(mValues[i].first > 0)
-                    return mLast; // old data
+                    break; // old data
             }
             if(val > mLast)
                 mLast = val;

diff --git a/src/publisher/pubMqtt.h b/src/publisher/pubMqtt.h
@@ -39,6 +39,9 @@ template<class HMSYSTEM>
 class PubMqtt {
     public:
         PubMqtt() : SendIvData() {
+            mutex = xSemaphoreCreateBinaryStatic(&mutexBuffer);
+            xSemaphoreGive(mutex);
+
             mLastIvState.fill(InverterStatus::OFF);
             mIvLastRTRpub.fill(0);
 
@@ -50,7 +53,9 @@ class PubMqtt {
             mSendAlarm.fill(false);
         }
 
-        ~PubMqtt() { }
+        ~PubMqtt() {
+            vSemaphoreDelete(mutex);
+        }
 
         void setup(IApp *app, cfgMqtt_t *cfg_mqtt, const char *devName, const char *version, HMSYSTEM *sys, uint32_t *utcTs, uint32_t *uptime) {
             mApp             = app;
@@ -96,6 +101,17 @@ class PubMqtt {
         }
 
         void loop() {
+            std::queue<message_s> queue;
+            xSemaphoreTake(mutex, portMAX_DELAY);
+            std::swap(queue, mReceiveQueue);
+            xSemaphoreGive(mutex);
+
+            while (!queue.empty()) {
+                message_s *entry = &queue.front();
+                handleMessage(entry->topic, entry->payload, entry->len, entry->index, entry->total);
+                queue.pop();
+            }
+
             SendIvData.loop();
 
             #if defined(ESP8266)
@@ -301,6 +317,14 @@ class PubMqtt {
         void onMessage(const espMqttClientTypes::MessageProperties& properties, const char* topic, const uint8_t* payload, size_t len, size_t index, size_t total) {
             if(len == 0)
                 return;
+
+            xSemaphoreTake(mutex, portMAX_DELAY);
+            mReceiveQueue.push(message_s(topic, payload, len, index, total));
+            xSemaphoreGive(mutex);
+
+        }
+
+        inline void handleMessage(const char* topic, const uint8_t* payload, size_t len, size_t index, size_t total) {
             DPRINT(DBG_INFO, mqttStr[MQTT_STR_GOT_TOPIC]);
             DBGPRINTLN(String(topic));
             if(NULL == mSubscriptionCb)
@@ -613,13 +637,39 @@ class PubMqtt {
     private:
         enum {MQTT_STATUS_OFFLINE = 0, MQTT_STATUS_PARTIAL, MQTT_STATUS_ONLINE};
 
+        struct message_s {
+            char* topic;
+            uint8_t* payload;
+            size_t len;
+            size_t index;
+            size_t total;
+
+            message_s(const char* topic, const uint8_t* payload, size_t len, size_t index, size_t total) {
+                this->topic = new char[strlen(topic) + 1];
+                this->payload = new uint8_t[len];
+
+                memcpy(this->topic, topic, strlen(topic));
+                memcpy(this->payload, payload, len);
+                this->len = len;
+                this->index = index;
+                this->total = total;
+            }
+
+            ~message_s() {
+                delete[] this->topic;
+                delete[] this->payload;
+            }
+        };
+
     private:
         espMqttClient mClient;
         cfgMqtt_t *mCfgMqtt = nullptr;
         IApp *mApp;
         #if defined(ESP8266)
         WiFiEventHandler mHWifiCon, mHWifiDiscon;
         #endif
+        SemaphoreHandle_t mutex;
+        StaticSemaphore_t mutexBuffer;
 
         HMSYSTEM *mSys = nullptr;
         PubMqttIvData<HMSYSTEM> SendIvData;
@@ -634,6 +684,8 @@ class PubMqtt {
         std::array<uint32_t, MAX_NUM_INVERTERS> mIvLastRTRpub;
         uint16_t mIntervalTimeout = 0;
 
+        std::queue<message_s> mReceiveQueue;
+
         // last will topic and payload must be available through lifetime of 'espMqttClient'
         std::array<char, (MQTT_TOPIC_LEN + 5)> mLwtTopic;
         const char *mDevName = nullptr, *mVersion = nullptr;