From 8a46ba9541b5ec95d6fe48e48c0d8155b470b740 Mon Sep 17 00:00:00 2001
From: Bernhard Kirchen <schlimmchen@posteo.net>
Date: Thu, 23 May 2024 22:41:59 +0200
Subject: [PATCH] Feature: HTTP+JSON power meter: poll asynchronously

---
 include/PowerMeterHttpJson.h | 19 ++++++++-
 src/PowerMeterHttpJson.cpp   | 75 +++++++++++++++++++++++++++++-------
 2 files changed, 79 insertions(+), 15 deletions(-)

diff --git a/include/PowerMeterHttpJson.h b/include/PowerMeterHttpJson.h
index 110517c9..d226bbc0 100644
--- a/include/PowerMeterHttpJson.h
+++ b/include/PowerMeterHttpJson.h
@@ -1,9 +1,12 @@
 // SPDX-License-Identifier: GPL-2.0-or-later
 #pragma once
 
+#include <atomic>
 #include <array>
 #include <variant>
 #include <memory>
+#include <condition_variable>
+#include <mutex>
 #include <stdint.h>
 #include "HttpGetter.h"
 #include "Configuration.h"
@@ -17,8 +20,10 @@ public:
     explicit PowerMeterHttpJson(PowerMeterHttpJsonConfig const& cfg)
         : _cfg(cfg) { }
 
+    ~PowerMeterHttpJson();
+
     bool init() final;
-    void loop() final;
+    void loop() final { } // polling is performed asynchronously
     float getPowerTotal() const final;
     bool isDataValid() const final;
     void doMqttPublish() const final;
@@ -28,11 +33,21 @@ public:
     poll_result_t poll();
 
 private:
+    static void pollingLoopHelper(void* context);
+    std::atomic<bool> _taskDone;
+    void pollingLoop();
+
     PowerMeterHttpJsonConfig const _cfg;
 
-    uint32_t _lastPoll;
+    uint32_t _lastPoll = 0;
 
+    mutable std::mutex _valueMutex;
     power_values_t _powerValues;
 
     std::array<std::unique_ptr<HttpGetter>, POWERMETER_HTTP_JSON_MAX_VALUES> _httpGetters;
+
+    TaskHandle_t _taskHandle = nullptr;
+    bool _stopPolling;
+    mutable std::mutex _pollingMutex;
+    std::condition_variable _cv;
 };
diff --git a/src/PowerMeterHttpJson.cpp b/src/PowerMeterHttpJson.cpp
index c199c41c..a7164408 100644
--- a/src/PowerMeterHttpJson.cpp
+++ b/src/PowerMeterHttpJson.cpp
@@ -8,6 +8,22 @@
 #include <base64.h>
 #include <ESPmDNS.h>
 
+PowerMeterHttpJson::~PowerMeterHttpJson()
+{
+    _taskDone = false;
+
+    std::unique_lock<std::mutex> lock(_pollingMutex);
+    _stopPolling = true;
+    lock.unlock();
+
+    _cv.notify_all();
+
+    if (_taskHandle != nullptr) {
+        while (!_taskDone) { delay(10); }
+        _taskHandle = nullptr;
+    }
+}
+
 bool PowerMeterHttpJson::init()
 {
     for (uint8_t i = 0; i < POWERMETER_HTTP_JSON_MAX_VALUES; i++) {
@@ -32,24 +48,54 @@ bool PowerMeterHttpJson::init()
         return false;
     }
 
+    std::unique_lock<std::mutex> lock(_pollingMutex);
+    _stopPolling = false;
+    lock.unlock();
+
+    uint32_t constexpr stackSize = 3072;
+    xTaskCreate(PowerMeterHttpJson::pollingLoopHelper, "PM:HTTP+JSON",
+            stackSize, this, 1/*prio*/, &_taskHandle);
+
     return true;
 }
 
-void PowerMeterHttpJson::loop()
+void PowerMeterHttpJson::pollingLoopHelper(void* context)
 {
-    if ((millis() - _lastPoll) < (_cfg.PollingInterval * 1000)) {
-        return;
+    auto pInstance = static_cast<PowerMeterHttpJson*>(context);
+    pInstance->pollingLoop();
+    pInstance->_taskDone = true;
+    vTaskDelete(nullptr);
+}
+
+void PowerMeterHttpJson::pollingLoop()
+{
+    std::unique_lock<std::mutex> lock(_pollingMutex);
+
+    while (!_stopPolling) {
+        auto elapsedMillis = millis() - _lastPoll;
+        auto intervalMillis = _cfg.PollingInterval * 1000;
+        if (_lastPoll > 0 && elapsedMillis < intervalMillis) {
+            auto sleepMs = intervalMillis - elapsedMillis;
+            _cv.wait_for(lock, std::chrono::milliseconds(sleepMs),
+                    [this] { return _stopPolling; }); // releases the mutex
+            continue;
+        }
+
+        _lastPoll = millis();
+
+        lock.unlock(); // polling can take quite some time
+        auto res = poll();
+        lock.lock();
+
+        if (std::holds_alternative<String>(res)) {
+            MessageOutput.printf("[PowerMeterHttpJson] %s\r\n", std::get<String>(res).c_str());
+            continue;
+        }
+
+        MessageOutput.printf("[PowerMeterHttpJson] New total: %.2f\r\n", getPowerTotal());
+
+        gotUpdate();
     }
-
-    _lastPoll = millis();
-
-    auto res = poll();
-    if (std::holds_alternative<String>(res)) {
-        MessageOutput.printf("[PowerMeterHttpJson] %s\r\n", std::get<String>(res).c_str());
-        return;
-    }
-
-    gotUpdate();
 }
 
 PowerMeterHttpJson::poll_result_t PowerMeterHttpJson::poll()
@@ -113,6 +159,7 @@ PowerMeterHttpJson::poll_result_t PowerMeterHttpJson::poll()
         if (cfg.SignInverted) { cache[i] *= -1; }
     }
 
+    std::unique_lock<std::mutex> lock(_valueMutex);
     _powerValues = cache;
     return cache;
 }
@@ -120,6 +167,7 @@ PowerMeterHttpJson::poll_result_t PowerMeterHttpJson::poll()
 float PowerMeterHttpJson::getPowerTotal() const
 {
     float sum = 0.0;
+    std::unique_lock<std::mutex> lock(_valueMutex);
     for (auto v: _powerValues) { sum += v; }
     return sum;
 }
@@ -132,6 +180,7 @@ bool PowerMeterHttpJson::isDataValid() const
 
 void PowerMeterHttpJson::doMqttPublish() const
 {
+    std::unique_lock<std::mutex> lock(_valueMutex);
     mqttPublish("power1", _powerValues[0]);
     mqttPublish("power2", _powerValues[1]);
     mqttPublish("power3", _powerValues[2]);