From 8d86bb222aa4134fbf459b01e35133ba8f21b4e2 Mon Sep 17 00:00:00 2001 From: Dmitry Borisenko <67171972+IoTManagerProject@users.noreply.github.com> Date: Fri, 11 Feb 2022 00:22:59 +0100 Subject: [PATCH] mqtt --- include/Global.h | 4 - include/MqttClient.h | 2 +- include/utils/WiFiUtils.h | 1 + src/Global.cpp | 4 - src/MqttClient.cpp | 183 +++++++++++++------------------------- src/PeriodicTasks.cpp | 10 +-- src/WsServer.cpp | 3 + src/utils/WiFiUtils.cpp | 2 +- 8 files changed, 75 insertions(+), 134 deletions(-) diff --git a/include/Global.h b/include/Global.h index fb08a1e4..4a6e440c 100644 --- a/include/Global.h +++ b/include/Global.h @@ -105,10 +105,6 @@ extern String prex; extern String all_widgets; extern String scenario; -extern int mqttConnectAttempts; -extern bool changeBroker; -extern int currentBroker; - // extern DynamicJsonDocument settingsFlashJsonDoc; // extern DynamicJsonDocument paramsFlashJsonDoc; // extern DynamicJsonDocument paramsHeapJsonDoc; \ No newline at end of file diff --git a/include/MqttClient.h b/include/MqttClient.h index 92674571..30e35961 100644 --- a/include/MqttClient.h +++ b/include/MqttClient.h @@ -6,7 +6,7 @@ void mqttInit(); void selectBroker(); -void getMqttData1(); +void getMqttData(); void getMqttData2(); bool isSecondBrokerSet(); boolean mqttConnect(); diff --git a/include/utils/WiFiUtils.h b/include/utils/WiFiUtils.h index e6b6b3ff..404cee06 100644 --- a/include/utils/WiFiUtils.h +++ b/include/utils/WiFiUtils.h @@ -1,6 +1,7 @@ #pragma once #include "Global.h" +#include "MqttClient.h" boolean isNetworkActive(); void routerConnect(); diff --git a/src/Global.cpp b/src/Global.cpp index adeaf8a8..57b4d4cc 100644 --- a/src/Global.cpp +++ b/src/Global.cpp @@ -58,10 +58,6 @@ String all_widgets = ""; String scenario = ""; String mqttRootDevice = ""; -int mqttConnectAttempts = 0; -bool changeBroker = false; -int currentBroker = 1; - // DynamicJsonDocument settingsFlashJsonDoc(JSON_BUFFER_SIZE); // DynamicJsonDocument paramsFlashJsonDoc(JSON_BUFFER_SIZE); // DynamicJsonDocument paramsHeapJsonDoc(JSON_BUFFER_SIZE); \ No newline at end of file diff --git a/src/MqttClient.cpp b/src/MqttClient.cpp index ca7dc738..11c5a94a 100644 --- a/src/MqttClient.cpp +++ b/src/MqttClient.cpp @@ -1,24 +1,19 @@ #include "MqttClient.h" void mqttInit() { - myNotAsyncActions->add( - do_MQTTPARAMSCHANGED, [&](void*) { - mqttReconnect(); - }, - nullptr); - mqtt.setCallback(mqttCallback); - ts.add( WIFI_MQTT_CONNECTION_CHECK, MQTT_RECONNECT_INTERVAL, [&](void*) { if (WiFi.status() == WL_CONNECTED) { SerialPrint("I", F("WIFI"), F("OK")); if (mqtt.connected()) { - SerialPrint("I", F("MQTT"), "OK, broker No " + String(currentBroker)); + SerialPrint("I", F("MQTT"), "OK"); + jsonWriteInt_(errorsHeapJson, F("mqtt"), mqtt.state()); // setLedStatus(LED_OFF); } else { SerialPrint("E", F("MQTT"), F("✖ Connection lost")); + jsonWriteInt_(errorsHeapJson, F("mqtt"), mqtt.state()); mqttConnect(); } } else { @@ -28,13 +23,59 @@ void mqttInit() { } }, nullptr, true); +} - // myNotAsyncActions->add( - // do_sendScenMQTT, [&](void*) { - // String scen = readFile(String(DEVICE_SCENARIO_FILE), 2048); - // publishInfo("scen", scen); - // }, - // nullptr); +void mqttLoop() { + if (!isNetworkActive() || !mqtt.connected()) { + return; + } + mqtt.loop(); +} + +boolean mqttConnect() { + getMqttData(); + bool res = false; + if (mqttServer == "") { + SerialPrint("E", "MQTT", F("mqttServer empty")); + jsonWriteInt_(errorsHeapJson, F("mqtt"), 6); + return res; + } + SerialPrint("I", "MQTT", "connection started"); + + SerialPrint("I", "MQTT", "broker " + mqttServer + ":" + String(mqttPort, DEC)); + SerialPrint("I", "MQTT", "topic " + mqttRootDevice); + + // setLedStatus(LED_FAST); + + mqtt.setServer(mqttServer.c_str(), mqttPort); + + if (!mqtt.connected()) { + bool connected = false; + if (mqttUser != "" && mqttPass != "") { + connected = mqtt.connect(chipId.c_str(), mqttUser.c_str(), mqttPass.c_str()); + SerialPrint("I", F("MQTT"), F("Go to connection with login and password")); + } else if (mqttUser == "" && mqttPass == "") { + connected = mqtt.connect(chipId.c_str()); + SerialPrint("I", F("MQTT"), F("Go to connection without login and password")); + } else { + SerialPrint("E", F("MQTT"), F("✖ Login or password missed")); + jsonWriteInt_(errorsHeapJson, F("mqtt"), 7); + return res; + } + + if (connected) { + SerialPrint("I", F("MQTT"), F("✔ connected")); + jsonWriteInt_(errorsHeapJson, F("mqtt"), mqtt.state()); + // setLedStatus(LED_OFF); + mqttSubscribe(); + res = true; + } else { + SerialPrint("E", F("MQTT"), "🡆 Could't connect, retry in " + String(MQTT_RECONNECT_INTERVAL / 1000) + "s"); + jsonWriteInt_(errorsHeapJson, F("mqtt"), mqtt.state()); + // setLedStatus(LED_FAST); + } + } + return res; } void mqttDisconnect() { @@ -47,16 +88,16 @@ void mqttReconnect() { mqttConnect(); } -void mqttLoop() { - if (!isNetworkActive() || !mqtt.connected()) { - return; - } - mqtt.loop(); +void getMqttData() { + mqttServer = jsonReadStr(settingsFlashJson, F("mqttServer")); + mqttPort = jsonReadInt(settingsFlashJson, F("mqttPort")); + mqttUser = jsonReadStr(settingsFlashJson, F("mqttUser")); + mqttPass = jsonReadStr(settingsFlashJson, F("mqttPass")); } void mqttSubscribe() { SerialPrint("I", F("MQTT"), F("subscribed")); - SerialPrint("I", "MQTT", mqttRootDevice); + SerialPrint("I", F("MQTT"), mqttRootDevice); mqtt.subscribe(mqttPrefix.c_str()); mqtt.subscribe((mqttRootDevice + "/+/control").c_str()); mqtt.subscribe((mqttRootDevice + "/update").c_str()); @@ -68,105 +109,6 @@ void mqttSubscribe() { } } -void selectBroker() { - if (changeBroker) { - changeBroker = false; - if (currentBroker == 1) { - getMqttData2(); - } else if (currentBroker == 2) { - getMqttData1(); - } - } else { - if (currentBroker == 1) { - getMqttData1(); - } else if (currentBroker == 2) { - getMqttData2(); - } - } -} - -void getMqttData1() { - currentBroker = 1; - mqttServer = jsonReadStr(settingsFlashJson, F("mqttServer")); - mqttPort = jsonReadInt(settingsFlashJson, F("mqttPort")); - mqttUser = jsonReadStr(settingsFlashJson, F("mqttUser")); - mqttPass = jsonReadStr(settingsFlashJson, F("mqttPass")); - // prex = mqttPrefix + "/" + chipId; -} - -void getMqttData2() { - currentBroker = 2; - mqttServer = jsonReadStr(settingsFlashJson, F("mqttServer2")); - mqttPort = jsonReadInt(settingsFlashJson, F("mqttPort2")); - mqttUser = jsonReadStr(settingsFlashJson, F("mqttUser2")); - mqttPass = jsonReadStr(settingsFlashJson, F("mqttPass2")); - // prex = mqttPrefix + "/" + chipId; -} - -bool isSecondBrokerSet() { - bool res = true; - if (jsonReadStr(settingsFlashJson, F("mqttServer2")) == "") { - res = false; - } - if (jsonReadStr(settingsFlashJson, F("mqttPrefix2")) == "") { - res = false; - } - return res; -} - -boolean mqttConnect() { - selectBroker(); - bool res = false; - if (mqttServer == "") { - SerialPrint("E", "MQTT", F("mqttServer empty")); - return res; - } - SerialPrint("I", "MQTT", "connection started to broker No " + String(currentBroker)); - - SerialPrint("I", "MQTT", "broker " + mqttServer + ":" + String(mqttPort, DEC)); - SerialPrint("I", "MQTT", "topic " + mqttRootDevice); - // setLedStatus(LED_FAST); - mqtt.setServer(mqttServer.c_str(), mqttPort); - - if (!mqtt.connected()) { - bool connected = false; - if (mqttUser != "" && mqttPass != "") { - connected = mqtt.connect(chipId.c_str(), mqttUser.c_str(), mqttPass.c_str()); - SerialPrint("I", F("MQTT"), F("Go to connection with login and password")); - } else if (mqttUser == "" && mqttPass == "") { - connected = mqtt.connect(chipId.c_str()); - SerialPrint("I", F("MQTT"), F("Go to connection without login and password")); - } else { - SerialPrint("E", F("MQTT"), F("✖ Login or password missing")); - return res; - } - - if (connected) { - SerialPrint("I", F("MQTT"), F("✔ connected")); - // if (currentBroker == 1) jsonWriteStr(settingsFlashJson, F("warning4"), F("

Подключено к основному брокеру

")); - // if (currentBroker == 2) jsonWriteStr(settingsFlashJson, F("warning4"), F("

Подключено к резервному брокеру

")); - // setLedStatus(LED_OFF); - mqttSubscribe(); - res = true; - } else { - mqttConnectAttempts++; - SerialPrint("E", F("MQTT"), "🡆 Attempt No: " + String(mqttConnectAttempts) + " could't connect, retry in " + String(MQTT_RECONNECT_INTERVAL / 1000) + "s"); - // setLedStatus(LED_FAST); - // jsonWriteStr(settingsFlashJson, F("warning4"), F("

Не подключено брокеру

")); - if (mqttConnectAttempts >= CHANGE_BROKER_AFTER) { - mqttConnectAttempts = 0; - if (isSecondBrokerSet()) { - changeBroker = true; - SerialPrint("E", F("MQTT"), "✖ Broker fully missed (" + String(CHANGE_BROKER_AFTER) + " attempts passed), try connect to another one"); - } else { - SerialPrint("E", F("MQTT"), F("Secound broker not seted")); - } - } - } - } - return res; -} - void mqttCallback(char* topic, uint8_t* payload, size_t length) { String topicStr = String(topic); // SerialPrint("I", "=>MQTT", topicStr); @@ -404,3 +346,6 @@ const String getStateStr() { break; } } + +// 6 сервер не задан +// 7 Логин или пароль отсутствует diff --git a/src/PeriodicTasks.cpp b/src/PeriodicTasks.cpp index 4f679439..b167841d 100644 --- a/src/PeriodicTasks.cpp +++ b/src/PeriodicTasks.cpp @@ -22,18 +22,18 @@ void periodicTasksInit() { void printGlobalVarSize() { size_t settingsFlashJsonSize = settingsFlashJson.length(); - SerialPrint(F("i"), F("settingsFlashJson"), String(settingsFlashJsonSize)); + // SerialPrint(F("i"), F("settingsFlashJson"), String(settingsFlashJsonSize)); size_t errorsHeapJsonSize = errorsHeapJson.length(); - SerialPrint(F("i"), F("settingsFlashJson"), String(errorsHeapJsonSize)); + // SerialPrint(F("i"), F("settingsFlashJson"), String(errorsHeapJsonSize)); size_t paramsFlashJsonSize = paramsFlashJson.length(); - SerialPrint(F("i"), F("settingsFlashJson"), String(paramsFlashJsonSize)); + // SerialPrint(F("i"), F("settingsFlashJson"), String(paramsFlashJsonSize)); size_t paramsHeapJsonSize = paramsHeapJson.length(); - SerialPrint(F("i"), F("settingsFlashJson"), String(paramsHeapJsonSize)); + // SerialPrint(F("i"), F("settingsFlashJson"), String(paramsHeapJsonSize)); size_t halfBuffer = JSON_BUFFER_SIZE / 2; if (settingsFlashJsonSize > halfBuffer || errorsHeapJsonSize > halfBuffer || paramsFlashJsonSize > halfBuffer || paramsHeapJsonSize > halfBuffer) { - SerialPrint(F("EE"), F("Json"), "Insufficient buffer size!!!"); + SerialPrint(F("EE"), F("Json"), F("Insufficient buffer size!!!")); jsonWriteInt(errorsHeapJson, "jsbuf", 1); } } \ No newline at end of file diff --git a/src/WsServer.cpp b/src/WsServer.cpp index eb0519b2..2dd31319 100644 --- a/src/WsServer.cpp +++ b/src/WsServer.cpp @@ -106,6 +106,9 @@ void webSocketEvent(uint8_t num, WStype_t type, uint8_t* payload, size_t length) if (headerStr == "/reboot|") { ESP.restart(); } + if (headerStr == "/mqtt|") { + mqttReconnect(); + } } break; diff --git a/src/utils/WiFiUtils.cpp b/src/utils/WiFiUtils.cpp index 0c12eeae..3b4374e2 100644 --- a/src/utils/WiFiUtils.cpp +++ b/src/utils/WiFiUtils.cpp @@ -41,7 +41,7 @@ void routerConnect() { SerialPrint("i", "WIFI", "http://" + WiFi.localIP().toString()); jsonWriteStr(settingsFlashJson, "ip", WiFi.localIP().toString()); - // mqttInit(); + mqttInit(); } SerialPrint("i", F("WIFI"), F("Network Init")); }