diff --git a/code/components/jomjol_mqtt/interface_mqtt.cpp b/code/components/jomjol_mqtt/interface_mqtt.cpp index 2496e111..d0b63a3d 100644 --- a/code/components/jomjol_mqtt/interface_mqtt.cpp +++ b/code/components/jomjol_mqtt/interface_mqtt.cpp @@ -11,8 +11,7 @@ static const char *TAG = "MQTT IF"; std::map>* connectFunktionMap = NULL; -std::map>* subscribeFunktionMap = NULL; - +std::map>* 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>::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 subHandler = mqtt_handler_flow_start; + MQTTregisterSubscribeFunction(maintopic + "/ctrl/flow_start", subHandler); // subcribe to maintopic/ctrl/flow_start + + if (subscribeFunktionMap != NULL) { + for(std::map>::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 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 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 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>(); } @@ -331,45 +376,15 @@ void MQTTregisterSubscribeFunction(std::string topic, std::function>::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>::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>::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); } } diff --git a/code/components/jomjol_mqtt/interface_mqtt.h b/code/components/jomjol_mqtt/interface_mqtt.h index b79cc2a7..eba7968b 100644 --- a/code/components/jomjol_mqtt/interface_mqtt.h +++ b/code/components/jomjol_mqtt/interface_mqtt.h @@ -26,6 +26,5 @@ void MQTTregisterSubscribeFunction(std::string topic, std::functionuri); + 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); } diff --git a/code/components/jomjol_tfliteclass/server_tflite.h b/code/components/jomjol_tfliteclass/server_tflite.h index 45e6815b..8ae046b9 100644 --- a/code/components/jomjol_tfliteclass/server_tflite.h +++ b/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 diff --git a/code/main/server_main.cpp b/code/main/server_main.cpp index 517ad0b7..7ee7a310 100644 --- a/code/main/server_main.cpp +++ b/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;