diff --git a/code/components/jomjol_mqtt/interface_mqtt.cpp b/code/components/jomjol_mqtt/interface_mqtt.cpp index 1789d252..935b122d 100644 --- 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>* connectFunktionMap = NULL; std::map>* 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 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>(); } 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 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 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>(); } 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>::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>::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>::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; } -} \ No newline at end of file +}