Trigger a flow start by REST API or MQTT (#1648)

* Trigger flow start by Rest API

* Increase handlers

* Update

* Update

* Update

* Change max handlers

* Add debug message

* Trigger flow start by MQTT

* Update

* Remove unused function

* Remove handler_doflow + routines

* Cleanup

* MergeCheck

code/components/jomjol_mqtt/interface_mqtt.cpp

@@ -11,8 +11,7 @@
 static const char *TAG = "MQTT IF";
 
 std::map<std::string, std::function<void()>>* connectFunktionMap = NULL;  
-std::map<std::string, std::function<bool(std::string, char*, int)>>* subscribeFunktionMap = NULL;  
-
+std::map<std::string, std::function<bool(std::string, char*, int)>>* subscribeFunktionMap = NULL;
 
 int failedOnRound = -1;
  
@@ -86,7 +85,6 @@ bool MQTTPublish(std::string _key, std::string _content, int retained_flag)
 
 
 static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event) {
-    int msg_id;
     std::string topic = "";
     switch (event->event_id) {
         case MQTT_EVENT_BEFORE_CONNECT:
@@ -106,8 +104,6 @@ static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event) {
             break;
         case MQTT_EVENT_SUBSCRIBED:
             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_LOGD(TAG, "sent publish successful, msg_id=%d", msg_id);
             break;
         case MQTT_EVENT_UNSUBSCRIBED:
             ESP_LOGD(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
@@ -117,12 +113,12 @@ static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event) {
             break;
         case MQTT_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);
+            ESP_LOGD(TAG, "TOPIC=%.*s", event->topic_len, event->topic);
+            ESP_LOGD(TAG, "DATA=%.*s", event->data_len, event->data);
             topic.assign(event->topic, event->topic_len);
             if (subscribeFunktionMap != NULL) {
                 if (subscribeFunktionMap->find(topic) != subscribeFunktionMap->end()) {
-                    ESP_LOGD(TAG, "call handler function\r\n");
+                    ESP_LOGD(TAG, "call subcribe function for topic %s", topic.c_str());
                     (*subscribeFunktionMap)[topic](topic, event->data, event->data_len);
                 }
             } else {
@@ -148,6 +144,7 @@ static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event) {
     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, "Event dispatched from event loop base=%s, event_id=%d", base, event_id);
     mqtt_event_handler_cb((esp_mqtt_event_handle_t) event_data);
@@ -276,9 +273,10 @@ int MQTT_Init() {
 void MQTTdestroy_client() {
     if (client) {
         if (mqtt_connected) {
+            MQTTdestroySubscribeFunction();      
             esp_mqtt_client_disconnect(client);
             mqtt_connected = false;
-        }        
+        }
         esp_mqtt_client_stop(client);
         esp_mqtt_client_destroy(client);
         client = NULL;
@@ -286,14 +284,59 @@ void MQTTdestroy_client() {
     }
 }
 
+
 bool getMQTTisEnabled() {
     return mqtt_enabled;
 }
 
+
 bool getMQTTisConnected() {
     return mqtt_connected;
 }
 
+
+bool mqtt_handler_flow_start(std::string _topic, char* _data, int _data_len) {
+    ESP_LOGD(TAG, "Handler called: topic %s, data %.*s", _topic.c_str(), _data_len, _data);
+
+    if (_data_len > 0) {
+        MQTTCtrlFlowStart(_topic);
+    }
+
+    return ESP_OK;
+}
+
+
+void MQTTconnected(){
+    if (mqtt_connected) {
+        LogFile.WriteToFile(ESP_LOG_INFO, TAG, "Connected to broker");
+        MQTTPublish(lwt_topic, lwt_connected, true);                        // Publish "connected" to maintopic/connection
+
+        if (connectFunktionMap != NULL) {
+            for(std::map<std::string, std::function<void()>>::iterator it = connectFunktionMap->begin(); it != connectFunktionMap->end(); ++it) {
+                it->second();
+                ESP_LOGD(TAG, "call connect function %s", it->first.c_str());
+            }
+        }
+
+        /* Subcribe to topics */
+        std::function<bool(std::string topic, char* data, int data_len)> subHandler = mqtt_handler_flow_start;     
+        MQTTregisterSubscribeFunction(maintopic + "/ctrl/flow_start", subHandler);        // subcribe to maintopic/ctrl/flow_start
+
+       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);
+                LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "topic " + it->first + " subscribe successful, msg_id=" + std::to_string(msg_id));
+            }
+        }
+    }
+    
+    vTaskDelay(10000 / portTICK_PERIOD_MS);                 // Delay execution of callback routine after connection got established   
+    if (callbackOnConnected) {                              // Call onConnected callback routine --> mqtt_server
+        callbackOnConnected(maintopic, SetRetainFlag);
+    }
+}
+
+
 void MQTTregisterConnectFunction(std::string name, std::function<void()> func){
     ESP_LOGD(TAG, "MQTTregisteronnectFunction %s\r\n", name.c_str());
     if (connectFunktionMap == NULL) {
@@ -312,6 +355,7 @@ void MQTTregisterConnectFunction(std::string name, std::function<void()> func){
     }
 }
 
+
 void MQTTunregisterConnectFunction(std::string name){
     ESP_LOGD(TAG, "unregisterConnnectFunction %s\r\n", name.c_str());
     if ((connectFunktionMap != NULL) && (connectFunktionMap->find(name) != connectFunktionMap->end())) {
@@ -319,8 +363,9 @@ void MQTTunregisterConnectFunction(std::string name){
     }
 }
 
+
 void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::string, char*, int)> func){
-    ESP_LOGD(TAG, "registerSubscribeFunction %s\r\n", topic.c_str());
+    ESP_LOGD(TAG, "registerSubscribeFunction %s", topic.c_str());
     if (subscribeFunktionMap == NULL) {
         subscribeFunktionMap = new std::map<std::string, std::function<bool(std::string, char*, int)>>();
     }
@@ -331,45 +376,15 @@ void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::st
     }
 
     (*subscribeFunktionMap)[topic] = func;
-
-    if (mqtt_connected) {
-        int msg_id = esp_mqtt_client_subscribe(client, topic.c_str(), 0);
-        ESP_LOGD(TAG, "topic %s subscribe successful, msg_id=%d", topic.c_str(), msg_id);
-    }
 }
 
-void MQTTconnected(){
-    if (mqtt_connected) {
-        LogFile.WriteToFile(ESP_LOG_INFO, TAG, "Connected to broker");
-
-        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, "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);
-                LogFile.WriteToFile(ESP_LOG_INFO, TAG, "topic " + it->first + " subscribe successful, msg_id=" + std::to_string(msg_id));
-            }
-        }
-
-        if (callbackOnConnected) {
-            callbackOnConnected(maintopic, SetRetainFlag);
-        }
-    }
-}
 
 void MQTTdestroySubscribeFunction(){
     if (subscribeFunktionMap != NULL) {
         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, "topic %s unsubscribe successful, msg_id=%d", it->first.c_str(), msg_id);
+                ESP_LOGD(TAG, "topic %s unsubscribe successful, msg_id=%d", it->first.c_str(), msg_id);
             }
         }
 

code/components/jomjol_mqtt/interface_mqtt.h

@@ -26,6 +26,5 @@ void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::st
 void MQTTdestroySubscribeFunction();
 void MQTTconnected();
 
-void MQTTdisable();
 #endif //INTERFACE_MQTT_H
 #endif //#ENABLE_MQTT

code/components/jomjol_mqtt/server_mqtt.cpp

@@ -214,7 +214,6 @@ esp_err_t sendDiscovery_and_static_Topics(httpd_req_t *req) {
 }
 
 void GotConnected(std::string maintopic, int retainFlag) {
-    vTaskDelay(10000 / portTICK_PERIOD_MS);     // Delay execution by 10s after connection got established   
     if (HomeassistantDiscovery) {
         MQTThomeassistantDiscovery();
     }

code/components/jomjol_tfliteclass/server_tflite.cpp

@@ -25,7 +25,6 @@
 
 ClassFlowControll tfliteflow;
 
-TaskHandle_t xHandleblink_task_doFlow = NULL;
 TaskHandle_t xHandletask_autodoFlow = NULL;
 
 bool FlowInitDone = false;
@@ -34,13 +33,13 @@ bool flowisrunning = false;
 long auto_intervall = 0;
 bool auto_isrunning = false;
 
-
 int countRounds = 0;
 
 static const char *TAG = "TFLITE SERVER";
 
 
-int getCountFlowRounds() {
+int getCountFlowRounds() 
+{
     return countRounds;
 }
 
@@ -65,19 +64,6 @@ bool isSetupModusActive() {
 
 void KillTFliteTasks()
 {
-    #ifdef DEBUG_DETAIL_ON          
-        ESP_LOGD(TAG, "Handle: xHandleblink_task_doFlow: %ld", (long) xHandleblink_task_doFlow);
-    #endif  
-    if (xHandleblink_task_doFlow != NULL)
-    {
-        TaskHandle_t xHandleblink_task_doFlowTmp = xHandleblink_task_doFlow;
-        xHandleblink_task_doFlow = NULL;
-        vTaskDelete(xHandleblink_task_doFlowTmp);
-        #ifdef DEBUG_DETAIL_ON      
-            ESP_LOGD(TAG, "Killed: xHandleblink_task_doFlow");
-        #endif
-    }
-
     #ifdef DEBUG_DETAIL_ON      
         ESP_LOGD(TAG, "Handle: xHandletask_autodoFlow: %ld", (long) xHandletask_autodoFlow);
     #endif
@@ -90,7 +76,6 @@ void KillTFliteTasks()
             ESP_LOGD(TAG, "Killed: xHandletask_autodoFlow");
         #endif
     }
-
 }
 
 
@@ -121,24 +106,9 @@ bool doflow(void)
 
     #ifdef DEBUG_DETAIL_ON      
         ESP_LOGD(TAG, "doflow - end %s", zw_time.c_str());
-    #endif    
-    return true;
-}
-
-
-void blink_task_doFlow(void *pvParameter)
-{
-    #ifdef DEBUG_DETAIL_ON          
-        ESP_LOGD(TAG, "blink_task_doFlow");
     #endif
-    if (!flowisrunning)
-    {
-        flowisrunning = true;
-        doflow();
-        flowisrunning = false;
-    }
-    vTaskDelete(NULL); //Delete this task if it exits from the loop above
-    xHandleblink_task_doFlow = NULL;
+
+    return true;
 }
 
 
@@ -167,37 +137,65 @@ esp_err_t handler_init(httpd_req_t *req)
 }
 
 
-esp_err_t handler_doflow(httpd_req_t *req)
-{
+esp_err_t handler_flow_start(httpd_req_t *req) {
+
     #ifdef DEBUG_DETAIL_ON          
-        LogFile.WriteHeapInfo("handler_doflow - Start");       
+    LogFile.WriteHeapInfo("handler_flow_start - Start");       
     #endif
 
-    ESP_LOGD(TAG, "handler_doFlow uri: %s", req->uri);
+    ESP_LOGD(TAG, "handler_flow_start uri: %s", req->uri);
 
-    if (flowisrunning)
-    {
-        const char* resp_str = "doFlow is already running and cannot be started again";
-        httpd_resp_send(req, resp_str, strlen(resp_str));       
-        return 2;
+    if (auto_isrunning) {
+        xTaskAbortDelay(xHandletask_autodoFlow); // Delay will be aborted if task is in blocked (waiting) state. If task is already running, no action
+        LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Flow start triggered by REST API /flow_start");
+        const char* resp_str = "The flow is going to be started immediately or is already running";
+        httpd_resp_send(req, resp_str, strlen(resp_str));  
     }
-    else
-    {
-        xTaskCreate(&blink_task_doFlow, "blink_doFlow", configMINIMAL_STACK_SIZE * 64, NULL, tskIDLE_PRIORITY+1, &xHandleblink_task_doFlow);
+    else {
+        LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Flow start triggered by REST API, but flow is not active!");
+        const char* resp_str = "WARNING: Flow start triggered by REST API, but flow is not active";
+        httpd_resp_send(req, resp_str, strlen(resp_str));  
     }
-    const char* resp_str = "doFlow started - takes about 60 seconds";
-    httpd_resp_send(req, resp_str, strlen(resp_str));  
+
     /* Respond with an empty chunk to signal HTTP response completion */
-    httpd_resp_send_chunk(req, NULL, 0);       
+    httpd_resp_send_chunk(req, NULL, 0);    
 
     #ifdef DEBUG_DETAIL_ON   
-        LogFile.WriteHeapInfo("handler_doflow - Done");       
+        LogFile.WriteHeapInfo("handler_flow_start - Done");       
     #endif
 
     return ESP_OK;
 }
 
 
+#ifdef ENABLE_MQTT
+esp_err_t MQTTCtrlFlowStart(std::string _topic) {
+
+    #ifdef DEBUG_DETAIL_ON          
+        LogFile.WriteHeapInfo("MQTTCtrlFlowStart - Start");       
+    #endif
+
+    ESP_LOGD(TAG, "MQTTCtrlFlowStart: topic %s", _topic.c_str());
+
+    if (auto_isrunning) 
+    {
+        xTaskAbortDelay(xHandletask_autodoFlow); // Delay will be aborted if task is in blocked (waiting) state. If task is already running, no action
+        LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Flow start triggered by MQTT topic " + _topic);
+    }
+    else 
+    {
+        LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Flow start triggered by MQTT topic " + _topic + ", but flow is not active!");
+    }  
+
+    #ifdef DEBUG_DETAIL_ON   
+        LogFile.WriteHeapInfo("MQTTCtrlFlowStart - Done");       
+    #endif
+
+    return ESP_OK;
+}
+#endif //ENABLE_MQTT
+
+
 esp_err_t handler_json(httpd_req_t *req)
 {
     #ifdef DEBUG_DETAIL_ON       
@@ -230,6 +228,7 @@ esp_err_t handler_json(httpd_req_t *req)
     #ifdef DEBUG_DETAIL_ON       
         LogFile.WriteHeapInfo("handler_JSON - Done");   
     #endif
+
     return ESP_OK;
 }
 
@@ -240,10 +239,6 @@ esp_err_t handler_wasserzaehler(httpd_req_t *req)
         LogFile.WriteHeapInfo("handler water counter - Start");    
     #endif
 
-							 
-															   
-	  
-
     if (FlowInitDone) 
     {
         bool _rawValue = false;
@@ -252,9 +247,7 @@ esp_err_t handler_wasserzaehler(httpd_req_t *req)
         std::string _type = "value";
         string zw;
 
-        ESP_LOGD(TAG, "handler water counter uri: %s", req->uri);
-
-															   
+        ESP_LOGD(TAG, "handler water counter uri: %s", req->uri);														   
 
         char _query[100];
         char _size[10];
@@ -386,10 +379,6 @@ esp_err_t handler_wasserzaehler(httpd_req_t *req)
         return ESP_ERR_NOT_FOUND;
     }
 
-
-																		
-										   
-
     #ifdef DEBUG_DETAIL_ON       
         LogFile.WriteHeapInfo("handler_wasserzaehler - Done");   
     #endif
@@ -699,9 +688,9 @@ esp_err_t handler_rssi(httpd_req_t *req)
 esp_err_t handler_uptime(httpd_req_t *req)
 {
 
-#ifdef DEBUG_DETAIL_ON       
-    LogFile.WriteHeapInfo("handler_uptime - Start");       
-#endif
+    #ifdef DEBUG_DETAIL_ON       
+        LogFile.WriteHeapInfo("handler_uptime - Start");       
+    #endif
     
     std::string formatedUptime = getFormatedUptime(false);
 
@@ -710,9 +699,9 @@ esp_err_t handler_uptime(httpd_req_t *req)
     /* Respond with an empty chunk to signal HTTP response completion */
     httpd_resp_send_chunk(req, NULL, 0);      
 
-#ifdef DEBUG_DETAIL_ON       
-    LogFile.WriteHeapInfo("handler_uptime - End");       
-#endif
+    #ifdef DEBUG_DETAIL_ON       
+        LogFile.WriteHeapInfo("handler_uptime - End");       
+    #endif
 
     return ESP_OK;
 }
@@ -862,7 +851,6 @@ void TFliteDoAutoStart()
     xReturned = xTaskCreate(&task_autodoFlow, "task_autodoFlow", configMINIMAL_STACK_SIZE * 35, NULL, tskIDLE_PRIORITY+1, &xHandletask_autodoFlow);
     if( xReturned != pdPASS )
     {
-
        //Memory: 64 --> 48 --> 35 --> 25
        ESP_LOGD(TAG, "ERROR task_autodoFlow konnte nicht erzeugt werden!");
     }
@@ -901,15 +889,15 @@ void register_server_tflite_uri(httpd_handle_t server)
     camuri.user_ctx  = (void*) "Prevalue";    
     httpd_register_uri_handler(server, &camuri);
 
-    camuri.uri       = "/doflow";
-    camuri.handler   = handler_doflow;
-    camuri.user_ctx  = (void*) "Light Off"; 
-    httpd_register_uri_handler(server, &camuri);  
+    camuri.uri       = "/flow_start";
+    camuri.handler   = handler_flow_start;
+    camuri.user_ctx  = (void*) "Flow Start"; 
+    httpd_register_uri_handler(server, &camuri);
 
     camuri.uri       = "/statusflow.html";
     camuri.handler   = handler_statusflow;
     camuri.user_ctx  = (void*) "Light Off"; 
-    httpd_register_uri_handler(server, &camuri);  
+    httpd_register_uri_handler(server, &camuri);
 
     camuri.uri       = "/statusflow";
     camuri.handler   = handler_statusflow;
@@ -925,13 +913,13 @@ void register_server_tflite_uri(httpd_handle_t server)
     camuri.uri       = "/cpu_temperature";
     camuri.handler   = handler_cputemp;
     camuri.user_ctx  = (void*) "Light Off"; 
-    httpd_register_uri_handler(server, &camuri);  
+    httpd_register_uri_handler(server, &camuri);
 
     // Legacy API => New: "/rssi"
     camuri.uri       = "/rssi.html";
     camuri.handler   = handler_rssi;
     camuri.user_ctx  = (void*) "Light Off"; 
-    httpd_register_uri_handler(server, &camuri);  
+    httpd_register_uri_handler(server, &camuri);
 
     camuri.uri       = "/rssi";
     camuri.handler   = handler_rssi;
@@ -946,7 +934,7 @@ void register_server_tflite_uri(httpd_handle_t server)
     camuri.uri       = "/editflow";
     camuri.handler   = handler_editflow;
     camuri.user_ctx  = (void*) "EditFlow"; 
-    httpd_register_uri_handler(server, &camuri);     
+    httpd_register_uri_handler(server, &camuri);   
 
     // Legacy API => New: "/value"
     camuri.uri       = "/value.html";
@@ -957,17 +945,16 @@ void register_server_tflite_uri(httpd_handle_t server)
     camuri.uri       = "/value";
     camuri.handler   = handler_wasserzaehler;
     camuri.user_ctx  = (void*) "Value"; 
-    httpd_register_uri_handler(server, &camuri);  
+    httpd_register_uri_handler(server, &camuri);
 
     // Legacy API => New: "/value"
     camuri.uri       = "/wasserzaehler.html";
     camuri.handler   = handler_wasserzaehler;
     camuri.user_ctx  = (void*) "Wasserzaehler"; 
-    httpd_register_uri_handler(server, &camuri);  
+    httpd_register_uri_handler(server, &camuri);
 
     camuri.uri       = "/json";
     camuri.handler   = handler_json;
     camuri.user_ctx  = (void*) "JSON"; 
-    httpd_register_uri_handler(server, &camuri);     
-
+    httpd_register_uri_handler(server, &camuri);
 }

code/components/jomjol_tfliteclass/server_tflite.h

@@ -12,24 +12,20 @@
 
 //#include "ClassControllCamera.h"
 
+extern ClassFlowControll tfliteflow;
 void register_server_tflite_uri(httpd_handle_t server);
 
 void KillTFliteTasks();
-
 void TFliteDoAutoStart();
-
 bool isSetupModusActive();
-
-#ifdef ENABLE_MQTT
-std::string GetMQTTMainTopic();
-#endif //ENABLE_MQTT
-
 int getCountFlowRounds();
 
 esp_err_t GetJPG(std::string _filename, httpd_req_t *req);
-
 esp_err_t GetRawJPG(httpd_req_t *req);
 
-extern ClassFlowControll tfliteflow;
+#ifdef ENABLE_MQTT
+std::string GetMQTTMainTopic();
+esp_err_t MQTTCtrlFlowStart(std::string);
+#endif //ENABLE_MQTT
 
 #endif //SERVERTFLITE_H

code/main/server_main.cpp

@@ -461,12 +461,12 @@ httpd_handle_t start_webserver(void)
     config.server_port = 80;
     config.ctrl_port = 32768;
     config.max_open_sockets = 5; //20210921 --> previously 7   
-    config.max_uri_handlers = 37; // previously 24, 20220511: 35             
+    config.max_uri_handlers = 38; // previously 24, 20220511: 35, 20221220: 37             
     config.max_resp_headers = 8;                        
     config.backlog_conn = 5;                        
     config.lru_purge_enable = true; // this cuts old connections if new ones are needed.               
     config.recv_wait_timeout = 5; // default: 5 20210924 --> previously 30              
-    config.send_wait_timeout = 5; // default: 5 20210924 --> previously 30                   
+    config.send_wait_timeout = 5; // default: 5 20210924 --> previously 30                    
     config.global_user_ctx = NULL;                        
     config.global_user_ctx_free_fn = NULL;                
     config.global_transport_ctx = NULL;