[SYS] Add Mutex to potect mqtt->publish() from concurrent… (#2034)

Protecting mqtt->publish() fixed the problem with corruption of MQTT discovery config messages #2012
This commit is contained in:
puterboy authored and GitHub committed 2024-08-29 07:10:46 -05:00
1 parent 5564ed531c
commit 830d58f98e
1 file changed
+12
+12
View File
@@ -95,6 +95,8 @@ std::queue<JsonBundle> jsonQueue;
# include <driver/adc.h> # include <driver/adc.h>
// Mutex to protect the queue // Mutex to protect the queue
SemaphoreHandle_t xQueueMutex; SemaphoreHandle_t xQueueMutex;
// Mutex to protect mqtt publish
SemaphoreHandle_t xMqttMutex;
bool blufiConnectAP = false; bool blufiConnectAP = false;
#endif #endif
@@ -718,6 +720,12 @@ void pubMQTT(const char* topic, const char* payload) {
*/ */
void pubMQTT(const char* topic, const char* payload, bool retainFlag) { void pubMQTT(const char* topic, const char* payload, bool retainFlag) {
if (SYSConfig.mqtt && !SYSConfig.offline) { if (SYSConfig.mqtt && !SYSConfig.offline) {
#ifdef ESP32
if (xSemaphoreTake(xMqttMutex, pdMS_TO_TICKS(QueueSemaphoreTimeOutTask)) == pdFALSE) {
Log.error(F("xMqttMutex not taken" CR));
return;
}
#endif
if (mqtt && mqtt->connected()) { if (mqtt && mqtt->connected()) {
SendReceiveIndicatorON(); SendReceiveIndicatorON();
Log.notice(F("[ OMG->MQTT ] topic: %s msg: %s " CR), topic, payload); Log.notice(F("[ OMG->MQTT ] topic: %s msg: %s " CR), topic, payload);
@@ -725,6 +733,9 @@ void pubMQTT(const char* topic, const char* payload, bool retainFlag) {
} else { } else {
Log.warning(F("MQTT not connected, aborting the publication" CR)); Log.warning(F("MQTT not connected, aborting the publication" CR));
} }
#ifdef ESP32
xSemaphoreGive(xMqttMutex);
#endif
} else { } else {
Log.notice(F("[ OMG->MQTT deactivated or offline] topic: %s msg: %s " CR), topic, payload); Log.notice(F("[ OMG->MQTT deactivated or offline] topic: %s msg: %s " CR), topic, payload);
} }
@@ -1286,6 +1297,7 @@ void setup() {
# endif # endif
#elif ESP32 #elif ESP32
xQueueMutex = xSemaphoreCreateMutex(); xQueueMutex = xSemaphoreCreateMutex();
xMqttMutex = xSemaphoreCreateMutex();
# if defined(ZboardM5STICKC) || defined(ZboardM5STICKCP) || defined(ZboardM5STACK) || defined(ZboardM5TOUGH) # if defined(ZboardM5STICKC) || defined(ZboardM5STICKCP) || defined(ZboardM5STACK) || defined(ZboardM5TOUGH)
setupM5(); setupM5();
# endif # endif