From 8f5c4878c5f89cf57aa180c9b3023da49d0cbec5 Mon Sep 17 00:00:00 2001
From: Bernhard Kirchen <schlimmchen@users.noreply.github.com>
Date: Sun, 31 Dec 2023 14:49:39 +0100
Subject: [PATCH] Fix: switch context when processing DPL MQTT requests (#572)

MQTT message callbacks are executed in the MQTT thread context. when
processing topics that control the DPL, we must avoid executing methods
that are not thread-safe. this change binds the methods to be called to
the respective parameters and executes them in the TaskScheduler
context, such that they no longer need to be thread-safe.
---
 include/MqttHandlePowerLimiter.h |  8 ++++++++
 src/MqttHandlePowerLimiter.cpp   | 32 ++++++++++++++++++++------------
 2 files changed, 28 insertions(+), 12 deletions(-)

diff --git a/include/MqttHandlePowerLimiter.h b/include/MqttHandlePowerLimiter.h
index b81917db6..d78c3f19d 100644
--- a/include/MqttHandlePowerLimiter.h
+++ b/include/MqttHandlePowerLimiter.h
@@ -4,6 +4,9 @@
 #include "Configuration.h"
 #include <espMqttClient.h>
 #include <TaskSchedulerDeclarations.h>
+#include <mutex>
+#include <deque>
+#include <functional>
 
 class MqttHandlePowerLimiterClass {
 public:
@@ -18,6 +21,11 @@ class MqttHandlePowerLimiterClass {
     uint32_t _lastPublishStats;
     uint32_t _lastPublish;
 
+    // MQTT callbacks to process updates on subscribed topics are executed in
+    // the MQTT thread's context. we use this queue to switch processing the
+    // user requests into the main loop's context (TaskScheduler context).
+    mutable std::mutex _mqttMutex;
+    std::deque<std::function<void()>> _mqttCallbacks;
 };
 
 extern MqttHandlePowerLimiterClass MqttHandlePowerLimiter;
\ No newline at end of file
diff --git a/src/MqttHandlePowerLimiter.cpp b/src/MqttHandlePowerLimiter.cpp
index 9a4a7f77e..b807fef0e 100644
--- a/src/MqttHandlePowerLimiter.cpp
+++ b/src/MqttHandlePowerLimiter.cpp
@@ -34,11 +34,21 @@ void MqttHandlePowerLimiterClass::init(Scheduler& scheduler)
 
 void MqttHandlePowerLimiterClass::loop()
 {
-    if (!MqttSettings.getConnected() ) {
+    std::unique_lock<std::mutex> mqttLock(_mqttMutex);
+
+    const CONFIG_T& config = Configuration.get();
+
+    if (!config.PowerLimiter.Enabled) {
+        _mqttCallbacks.clear();
         return;
     }
 
-    const CONFIG_T& config = Configuration.get();
+    for (auto& callback : _mqttCallbacks) { callback(); }
+    _mqttCallbacks.clear();
+
+    mqttLock.unlock();
+
+    if (!MqttSettings.getConnected() ) { return; }
 
     if ((millis() - _lastPublish) > (config.Mqtt.PublishInterval * 1000) ) {
         auto val = static_cast<unsigned>(PowerLimiter.getMode());
@@ -53,13 +63,6 @@ void MqttHandlePowerLimiterClass::loop()
 void MqttHandlePowerLimiterClass::onCmdMode(const espMqttClientTypes::MessageProperties& properties,
         const char* topic, const uint8_t* payload, size_t len, size_t index, size_t total)
 {
-    const CONFIG_T& config = Configuration.get();
-
-    // ignore messages if PowerLimiter is disabled
-    if (!config.PowerLimiter.Enabled) {
-        return;
-    }
-
     std::string strValue(reinterpret_cast<const char*>(payload), len);
     int intValue = -1;
     try {
@@ -71,19 +74,24 @@ void MqttHandlePowerLimiterClass::onCmdMode(const espMqttClientTypes::MessagePro
         return;
     }
 
+    std::lock_guard<std::mutex> mqttLock(_mqttMutex);
+
     using Mode = PowerLimiterClass::Mode;
     switch (static_cast<Mode>(intValue)) {
         case Mode::UnconditionalFullSolarPassthrough:
             MessageOutput.println("Power limiter unconditional full solar PT");
-            PowerLimiter.setMode(Mode::UnconditionalFullSolarPassthrough);
+            _mqttCallbacks.push_back(std::bind(&PowerLimiterClass::setMode,
+                        &PowerLimiter, Mode::UnconditionalFullSolarPassthrough));
             break;
         case Mode::Disabled:
             MessageOutput.println("Power limiter disabled (override)");
-            PowerLimiter.setMode(Mode::Disabled);
+            _mqttCallbacks.push_back(std::bind(&PowerLimiterClass::setMode,
+                        &PowerLimiter, Mode::Disabled));
             break;
         case Mode::Normal:
             MessageOutput.println("Power limiter normal operation");
-            PowerLimiter.setMode(Mode::Normal);
+            _mqttCallbacks.push_back(std::bind(&PowerLimiterClass::setMode,
+                        &PowerLimiter, Mode::Normal));
             break;
         default:
             MessageOutput.printf("PowerLimiter - unknown mode %d\r\n", intValue);