resolve release merge conflicts

2025-12-10 21:46:55 +03:00 · 2022-11-25 16:45:02 +01:00
parent e36490c89e
commit 3815f9cf0a
1 changed files with 144 additions and 118 deletions
--- a/code/components/jomjol_mqtt/interface_mqtt.cpp
+++ b/code/components/jomjol_mqtt/interface_mqtt.cpp
@@ -4,35 +4,70 @@
 #include "esp_log.h"
 #include "mqtt_client.h"
 #include "ClassLogFile.h"
+#include "server_tflite.h"

 #define __HIDE_PASSWORD

-static const char *TAG_INTERFACEMQTT = "interface_mqtt";
+static const char *TAG = "MQTT INTERFACE";

 std::map<std::string, std::function<void()>>* connectFunktionMap = NULL;  
 std::map<std::string, std::function<bool(std::string, char*, int)>>* subscribeFunktionMap = NULL;  
-bool debugdetail = true;

-// #define CONFIG_BROKER_URL "mqtt://192.168.178.43:1883"
+
+int failedOnRound = -1;

 esp_mqtt_event_id_t esp_mmqtt_ID = MQTT_EVENT_ANY;
 // ESP_EVENT_ANY_ID

 bool mqtt_connected = false;
 esp_mqtt_client_handle_t client = NULL;
+std::string uri, client_id, lwt_topic, lwt_connected, lwt_disconnected, user, password, maintopic;
+int keepalive, SetRetainFlag;
+void (*callbackOnConnected)(std::string, int) = NULL;

-bool MQTTPublish(std::string _key, std::string _content, int retained_flag){
-  
+
+bool MQTTPublish(std::string _key, std::string _content, int retained_flag) {
    int msg_id;
    std::string zw;
+
+    if (failedOnRound == getCountFlowRounds()) { // we already failed in this round, do not retry until the next round
+        return true; // Fail quietly
+    }
+
+    LogFile.WriteHeapInfo("MQTT Publish");
+
+    if (!mqtt_connected) {
+        LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Not connected, trying to re-connect...");
+        if (!MQTT_Init()) {
+            if (!MQTT_Init()) { // Retry
+                LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Failed to init, skipping all MQTT publishings in this round!");
+                failedOnRound = getCountFlowRounds();
+                return false;
+            }
+        }
+    }    
+
    msg_id = esp_mqtt_client_publish(client, _key.c_str(), _content.c_str(), 0, 1, retained_flag);
    if (msg_id < 0) {
-        LogFile.WriteToFile("MQTT - Failed to publish '" + _key + "'!");
-        return false;
+        LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Failed to publish topic '" + _key + "', re-trying...");
+
+        msg_id = esp_mqtt_client_publish(client, _key.c_str(), _content.c_str(), 0, 1, retained_flag);
+        if (msg_id < 0) {
+            LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Failed to publish topic '" + _key + "', skipping all MQTT publishings in this round!");
+            mqtt_connected = false; // Force re-init on next call
+            failedOnRound = getCountFlowRounds();
+            return false;
+        }
    }
-    zw = "MQTT - sent publish successful in MQTTPublish, msg_id=" + std::to_string(msg_id) + ", " + _key + ", " + _content;
-    if (debugdetail) LogFile.WriteToFile(zw);
-    ESP_LOGD(TAG_INTERFACEMQTT, "sent publish successful in MQTTPublish, msg_id=%d, %s, %s", msg_id, _key.c_str(), _content.c_str());
+
+    if (_content.length() > 80) { // Truncate message if too long
+        _content.resize(80);
+        _content.append("..");
+    }
+
+    zw = "Published topic: " + _key + ", content: " + _content + " (msg_id=" + std::to_string(msg_id) + ")";
+    LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, zw);
+
    return true;
 }

@@ -43,162 +78,147 @@ static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event)
    std::string topic = "";
    switch (event->event_id) {
        case MQTT_EVENT_BEFORE_CONNECT:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_BEFORE_CONNECT");
+            ESP_LOGD(TAG, "MQTT_EVENT_BEFORE_CONNECT");
            break;
        case MQTT_EVENT_CONNECTED:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_CONNECTED");
+            ESP_LOGD(TAG, "MQTT_EVENT_CONNECTED");
            mqtt_connected = true;
            MQTTconnected();
            break;
        case MQTT_EVENT_DISCONNECTED:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_DISCONNECTED");
+            ESP_LOGD(TAG, "MQTT_EVENT_DISCONNECTED");
+            LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Disconnected! Going to re-connect...");
+            mqtt_connected = false; // Force re-init on next call
+            esp_mqtt_client_reconnect(client);
            break;
        case MQTT_EVENT_SUBSCRIBED:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
+            ESP_LOGD(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
            msg_id = esp_mqtt_client_publish(client, "/topic/qos0", "data", 0, 0, 0);
-            ESP_LOGI(TAG_INTERFACEMQTT, "sent publish successful, msg_id=%d", msg_id);
+            ESP_LOGD(TAG, "sent publish successful, msg_id=%d", msg_id);
            break;
        case MQTT_EVENT_UNSUBSCRIBED:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
+            ESP_LOGD(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
            break;
        case MQTT_EVENT_PUBLISHED:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
+            ESP_LOGD(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
            break;
        case MQTT_EVENT_DATA:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_DATA");
-            ESP_LOGI(TAG_INTERFACEMQTT, "TOPIC=%.*s\r\n", event->topic_len, event->topic);
-            ESP_LOGI(TAG_INTERFACEMQTT, "DATA=%.*s\r\n", event->data_len, event->data);
+            ESP_LOGD(TAG, "MQTT_EVENT_DATA");
+            ESP_LOGD(TAG, "TOPIC=%.*s\r\n", event->topic_len, event->topic);
+            ESP_LOGD(TAG, "DATA=%.*s\r\n", event->data_len, event->data);
            topic.assign(event->topic, event->topic_len);
            if (subscribeFunktionMap != NULL) {
                if (subscribeFunktionMap->find(topic) != subscribeFunktionMap->end()) {
-                    ESP_LOGD(TAG_INTERFACEMQTT, "call handler function\r\n");
+                    ESP_LOGD(TAG, "call handler function\r\n");
                    (*subscribeFunktionMap)[topic](topic, event->data, event->data_len);
                }
            } else {
-                ESP_LOGW(TAG_INTERFACEMQTT, "no handler available\r\n");
+                ESP_LOGW(TAG, "no handler available\r\n");
            }
            break;
        case MQTT_EVENT_ERROR:
-            ESP_LOGI(TAG_INTERFACEMQTT, "MQTT_EVENT_ERROR");
+            ESP_LOGD(TAG, "MQTT_EVENT_ERROR");
+            mqtt_connected = false; // Force re-init on next call
            break;
        default:
-            ESP_LOGI(TAG_INTERFACEMQTT, "Other event id:%d", event->event_id);
+            ESP_LOGD(TAG, "Other event id:%d", event->event_id);
            break;
    }
    return ESP_OK;
 }

 static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) {
-    ESP_LOGD(TAG_INTERFACEMQTT, "Event dispatched from event loop base=%s, event_id=%d", base, event_id);
+    ESP_LOGD(TAG, "Event dispatched from event loop base=%s, event_id=%d", base, event_id);
    mqtt_event_handler_cb((esp_mqtt_event_handle_t) event_data);
 }


-bool MQTTInit(std::string _mqttURI, std::string _clientid, std::string _user, std::string _password, std::string _LWTContext, int _keepalive){
-    std::string _zwmessage = "connection lost";
+void MQTT_Configure(std::string _mqttURI, std::string _clientid, std::string _user, std::string _password,
+        std::string _maintopic, std::string _lwt, std::string _lwt_connected, std::string _lwt_disconnected,
+        int _keepalive, int _SetRetainFlag, void *_callbackOnConnected){
+#ifdef __HIDE_PASSWORD
+    LogFile.WriteToFile(ESP_LOG_INFO, TAG, "URI: " + _mqttURI + ", clientname: " + _clientid + 
+            ", user: " + _user + ", password: XXXXXXXX, maintopic: " + _maintopic + ", last-will-topic: " + _maintopic + "/" + _lwt + ", keepAlive: " + std::to_string(_keepalive)); 
+#else
+    LogFile.WriteToFile(ESP_LOG_INFO, TAG, "URI: " + _mqttURI + ", clientname: " + _clientid + 
+            ", user: " + _user + ", password: " + _password + ", maintopic: " + _maintopic + ", last-will-topic: " + _maintopic + "/" + _lwt + ", keepAlive: " + std::to_string(_keepalive)); 
+#endif

-    int _lzw = _zwmessage.length();
-
-/*    LWTContext = _LWTContext;
-
-    mqtt_cfg.uri = _mqttURI.c_str();
-    mqtt_cfg.client_id = _clientid.c_str();
-    mqtt_cfg.lwt_topic = _LWTContext.c_str();
-    mqtt_cfg.lwt_msg = _zwmessage.c_str();
-    mqtt_cfg.lwt_retain = 1;
-    mqtt_cfg.lwt_msg_len = _lzw;
-    mqtt_cfg.keepalive = _keepalive;
-*/
-
-    esp_mqtt_client_config_t mqtt_cfg = {
-        .uri = _mqttURI.c_str(),
-        .client_id = _clientid.c_str(),
-        .lwt_topic = _LWTContext.c_str(),
-        .lwt_msg = _zwmessage.c_str(),
-        .lwt_retain = 1,
-        .lwt_msg_len = _lzw,
-        .keepalive = _keepalive
-    };
-
-    LogFile.WriteToFile("MQTT - Init");
+    uri = _mqttURI;
+    client_id = _clientid;
+    lwt_topic = _maintopic + "/" + _lwt;
+    lwt_connected = _lwt_connected;
+    lwt_disconnected = _lwt_disconnected;
+    keepalive = _keepalive;
+    SetRetainFlag = _SetRetainFlag;
+    maintopic = _maintopic;
+    callbackOnConnected = ( void (*)(std::string, int) )(_callbackOnConnected);

    if (_user.length() && _password.length()){
-        mqtt_cfg.username = _user.c_str();
-        mqtt_cfg.password = _password.c_str();
+        user = _user;
+        password = _password;
+    }
+}

-#ifdef __HIDE_PASSWORD
-        ESP_LOGI(TAG_INTERFACEMQTT, "Connect to MQTT: %s, XXXXXXXX", mqtt_cfg.username);
-#else
-        ESP_LOGI(TAG_INTERFACEMQTT, "Connect to MQTT: %s, %s", mqtt_cfg.username, mqtt_cfg.password);
-#endif        
+bool MQTT_Init() {
+    esp_err_t ret;
+    LogFile.WriteToFile(ESP_LOG_INFO, TAG, std::string("Init"));
+
+    MQTTdestroy_client();
+
+    std::string lw = lwt_disconnected;
+
+    esp_mqtt_client_config_t mqtt_cfg = {
+        .uri = uri.c_str(),
+        .client_id = client_id.c_str(),
+        .lwt_topic = lwt_topic.c_str(),
+        .lwt_msg = lw.c_str(),
+        .lwt_retain = 1,
+        .lwt_msg_len = (int)(lw.length()),
+        .keepalive = keepalive
    };

-    MQTTdestroy();
+    if (user.length() && password.length()){
+        mqtt_cfg.username = user.c_str();
+        mqtt_cfg.password = password.c_str();
+    };
+
+    LogFile.WriteHeapInfo("MQTT Client Init");
    client = esp_mqtt_client_init(&mqtt_cfg);
    if (client)
    {
-        if (esp_mqtt_client_register_event(client, esp_mmqtt_ID, mqtt_event_handler, client) != ESP_OK)
+        ret = esp_mqtt_client_register_event(client, esp_mmqtt_ID, mqtt_event_handler, client);
+        if (ret != ESP_OK)
        {
-            LogFile.WriteToFile("MQTT - Could not register event!");
-            return false;
-        }
-        if (esp_mqtt_client_start(client) != ESP_OK)
-        {
-            LogFile.WriteToFile("MQTT - Could not start client!");
+            LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Could not register event (ret=" + std::to_string(ret) + ")!");
            return false;
        }

-       /* if(!MQTTPublish(_LWTContext, "", 1))
+        LogFile.WriteHeapInfo("MQTT Client Start");
+        ret = esp_mqtt_client_start(client);
+        if (ret != ESP_OK)
        {
-            LogFile.WriteToFile("MQTT - Could not publish LWT!");
-            return false;
-        }*/
+            LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Could not start client (ret=" + std::to_string(ret) + "), retrying...");
+            ret = esp_mqtt_client_start(client);
+            if (ret != ESP_OK)
+            {
+                LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Could not start client (ret=" + std::to_string(ret) + ")!");
+                return false;
+            }
+        }
    }
    else
    {
-        LogFile.WriteToFile("MQTT - Could not Init client!");
+        LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Could not init client!");
        return false;
    }

-    LogFile.WriteToFile("MQTT - Init successful");
+    LogFile.WriteToFile(ESP_LOG_INFO, TAG, "Init successful");
    return true;
 }

-/*
-void MQTTReConnect(){
-    std::string _zwmessage = "connection lost";
-    int _lzw = _zwmessage.length();

->>>>>>> Stashed changes
-    client = esp_mqtt_client_init(&mqtt_cfg);
-    if (client)
-    {
-        if (esp_mqtt_client_register_event(client, esp_mmqtt_ID, mqtt_event_handler, client) != ESP_OK)
-            LogFile.WriteToFile("MQTT - Could not register event!");
-        if (esp_mqtt_client_start(client) != ESP_OK)
-            LogFile.WriteToFile("MQTT - Could not start client!");
-
-<<<<<<< Updated upstream
-        if(MQTTPublish(_LWTContext, "", 1)) {
-            LogFile.WriteToFile("MQTT - Client init successful");
-        }
-=======
-        if (mqtt_connected)
-            MQTTPublish(LWTContext, "", 1);
-        else
-            LogFile.WriteToFile("Problem with (Re)Connection not successful!");
-
->>>>>>> Stashed changes
-    }
-    else
-    {
-        LogFile.WriteToFile("MQTT - Could not Init client!");
-    }
-
-}
-*/
-
-void MQTTdestroy() {
+void MQTTdestroy_client() {
    if (client != NULL) {
        esp_mqtt_client_stop(client);
        esp_mqtt_client_destroy(client);
@@ -210,13 +230,13 @@ bool MQTTisConnected() {
 }

 void MQTTregisterConnectFunction(std::string name, std::function<void()> func){
-    ESP_LOGD(TAG_INTERFACEMQTT, "MQTTregisteronnectFunction %s\r\n", name.c_str());
+    ESP_LOGD(TAG, "MQTTregisteronnectFunction %s\r\n", name.c_str());
    if (connectFunktionMap == NULL) {
        connectFunktionMap = new std::map<std::string, std::function<void()>>();
    }

    if ((*connectFunktionMap)[name] != NULL) {
-        ESP_LOGW(TAG_INTERFACEMQTT, "connect function %s already registred", name.c_str());
+        ESP_LOGW(TAG, "connect function %s already registred", name.c_str());
        return;
    }

@@ -228,20 +248,20 @@ void MQTTregisterConnectFunction(std::string name, std::function<void()> func){
 }

 void MQTTunregisterConnectFunction(std::string name){
-    ESP_LOGD(TAG_INTERFACEMQTT, "MQTTregisteronnectFunction %s\r\n", name.c_str());
+    ESP_LOGD(TAG, "unregisterConnnectFunction %s\r\n", name.c_str());
    if ((connectFunktionMap != NULL) && (connectFunktionMap->find(name) != connectFunktionMap->end())) {
        connectFunktionMap->erase(name);
    }
 }

 void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::string, char*, int)> func){
-    ESP_LOGD(TAG_INTERFACEMQTT, "MQTTregisterSubscribeFunction %s\r\n", topic.c_str());
+    ESP_LOGD(TAG, "registerSubscribeFunction %s\r\n", topic.c_str());
    if (subscribeFunktionMap == NULL) {
        subscribeFunktionMap = new std::map<std::string, std::function<bool(std::string, char*, int)>>();
    }

    if ((*subscribeFunktionMap)[topic] != NULL) {
-        ESP_LOGW(TAG_INTERFACEMQTT, "topic %s already registred for subscription", topic.c_str());
+        ESP_LOGW(TAG, "topic %s already registered for subscription", topic.c_str());
        return;
    }

@@ -249,27 +269,33 @@ void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::st

    if (mqtt_connected) {
        int msg_id = esp_mqtt_client_subscribe(client, topic.c_str(), 0);
-        ESP_LOGD(TAG_INTERFACEMQTT, "topic %s subscribe successful, msg_id=%d", topic.c_str(), msg_id);
+        ESP_LOGD(TAG, "topic %s subscribe successful, msg_id=%d", topic.c_str(), msg_id);
    }
 }

 void MQTTconnected(){
    if (mqtt_connected) {
-        LogFile.WriteToFile("MQTT - Connected");
+        LogFile.WriteToFile(ESP_LOG_INFO, TAG, "Connected");
+
+        MQTTPublish(lwt_topic, lwt_connected, true);
+
        if (connectFunktionMap != NULL) {
            for(std::map<std::string, std::function<void()>>::iterator it = connectFunktionMap->begin(); it != connectFunktionMap->end(); ++it) {
                it->second();
-                ESP_LOGD(TAG_INTERFACEMQTT, "call connect function %s", it->first.c_str());
+                ESP_LOGD(TAG, "call connect function %s", it->first.c_str());
            }
        }

       if (subscribeFunktionMap != NULL) {
            for(std::map<std::string, std::function<bool(std::string, char*, int)>>::iterator it = subscribeFunktionMap->begin(); it != subscribeFunktionMap->end(); ++it) {
                int msg_id = esp_mqtt_client_subscribe(client, it->first.c_str(), 0);
-                ESP_LOGD(TAG_INTERFACEMQTT, "topic %s subscribe successful, msg_id=%d", it->first.c_str(), msg_id);
-                LogFile.WriteToFile("MQTT - topic " + it->first + " subscribe successful, msg_id=" + std::to_string(msg_id));
+                LogFile.WriteToFile(ESP_LOG_INFO, TAG, "topic " + it->first + " subscribe successful, msg_id=" + std::to_string(msg_id));
            }
        }
+
+        if (callbackOnConnected) {
+            callbackOnConnected(maintopic, SetRetainFlag);
+        }
    }
 }

@@ -278,7 +304,7 @@ void MQTTdestroySubscribeFunction(){
        if (mqtt_connected) {
            for(std::map<std::string, std::function<bool(std::string, char*, int)>>::iterator it = subscribeFunktionMap->begin(); it != subscribeFunktionMap->end(); ++it) {
                int msg_id = esp_mqtt_client_unsubscribe(client, it->first.c_str());
-                ESP_LOGI(TAG_INTERFACEMQTT, "topic %s unsubscribe successful, msg_id=%d", it->first.c_str(), msg_id);
+                ESP_LOGI(TAG, "topic %s unsubscribe successful, msg_id=%d", it->first.c_str(), msg_id);
            }
        }

@@ -286,4 +312,4 @@ void MQTTdestroySubscribeFunction(){
        delete subscribeFunktionMap;
        subscribeFunktionMap = NULL;
    }
-}
+}