diff --git a/src/CHANGES.md b/src/CHANGES.md index f4c5d4ce..51723203 100644 --- 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 index ab651821..bae0df51 100644 --- 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 index a8db6909..68665d61 100644 --- 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 index f092a159..477d942a 100644 --- a/src/publisher/pubMqtt.h +++ b/src/publisher/pubMqtt.h @@ -39,6 +39,9 @@ template 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 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,6 +637,30 @@ 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; @@ -620,6 +668,8 @@ class PubMqtt { #if defined(ESP8266) WiFiEventHandler mHWifiCon, mHWifiDiscon; #endif + SemaphoreHandle_t mutex; + StaticSemaphore_t mutexBuffer; HMSYSTEM *mSys = nullptr; PubMqttIvData SendIvData; @@ -634,6 +684,8 @@ class PubMqtt { std::array mIvLastRTRpub; uint16_t mIntervalTimeout = 0; + std::queue mReceiveQueue; + // last will topic and payload must be available through lifetime of 'espMqttClient' std::array mLwtTopic; const char *mDevName = nullptr, *mVersion = nullptr;