This commit is contained in:
Dmitry Borisenko
2021-10-05 19:19:14 +08:00
parent 60a292a2f5
commit 74c31e30ea
269 changed files with 0 additions and 20922 deletions

View File

@@ -1,378 +0,0 @@
#include "MqttClient.h"
#include "BufferExecute.h"
#include "Class/NotAsync.h"
#include "Global.h"
#include "Init.h"
#include "items/vLogging.h"
enum MqttBroker { MQTT_PRIMARY,
MQTT_RESERVE };
MqttBroker activeBroker = MQTT_PRIMARY;
String mqttPrefix;
String mqttRootDevice;
String mqttPass;
String mqttServer;
String mqttUser;
uint16_t mqttPort{0};
uint16_t reconnectionCounter{0};
uint16_t fallbackCounter{0};
const String getParamName(const char* param, MqttBroker broker) {
return String("mqtt") + param + (broker == MQTT_RESERVE ? "2" : "");
}
bool checkBrokerParams(MqttBroker broker) {
return !jsonReadStr(configSetupJson, getParamName("Server", broker)).isEmpty();
}
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", "WIFI", "OK");
if (mqtt.connected()) {
if (activeBroker == MQTT_RESERVE) {
// при 20 cекундных интервалах проверки, каждые 100 сек
if (fallbackCounter++ > 5) {
if (checkBrokerParams(MQTT_PRIMARY)) {
activeBroker = MQTT_PRIMARY;
fallbackCounter = 0;
mqttReconnect();
}
}
} else {
SerialPrint("I", "MQTT", "OK");
setLedStatus(LED_OFF);
}
} else {
SerialPrint("E", "MQTT", "lost connection");
if (reconnectionCounter++ > 5) {
if (activeBroker == MQTT_PRIMARY) {
if (checkBrokerParams(MQTT_RESERVE)) {
activeBroker = MQTT_RESERVE;
}
} else {
activeBroker = MQTT_PRIMARY;
}
reconnectionCounter = 0;
}
mqttConnect();
}
} else {
SerialPrint("E", "WIFI", "Lost WiFi connection");
ts.remove(WIFI_MQTT_CONNECTION_CHECK);
startAPMode();
}
},
nullptr, true);
myNotAsyncActions->add(
do_sendScenMQTT, [&](void*) {
String scen = readFile(String(DEVICE_SCENARIO_FILE), 2048);
publishInfo("scen", scen);
},
nullptr);
}
void mqttDisconnect() {
SerialPrint("I", "MQTT", "disconnect");
mqtt.disconnect();
}
void mqttReconnect() {
mqttDisconnect();
mqttConnect();
}
void mqttLoop() {
if (!isNetworkActive() || !mqtt.connected()) {
return;
}
mqtt.loop();
}
void mqttSubscribe() {
SerialPrint("I", "MQTT", "subscribe");
mqtt.subscribe(mqttPrefix.c_str());
mqtt.subscribe((mqttRootDevice + "/+/control").c_str());
mqtt.subscribe((mqttRootDevice + "/update").c_str());
if (jsonReadBool(configSetupJson, "MqttIn")) {
mqtt.subscribe((mqttPrefix + "/+/+/event").c_str());
mqtt.subscribe((mqttPrefix + "/+/+/order").c_str());
mqtt.subscribe((mqttPrefix + "/+/+/info").c_str());
}
}
bool readBrokerParams(MqttBroker broker) {
if (!checkBrokerParams(broker)) {
return false;
}
mqttServer = jsonReadStr(configSetupJson, getParamName("Server", broker));
mqttPort = jsonReadInt(configSetupJson, getParamName("Port", broker));
mqttUser = jsonReadStr(configSetupJson, getParamName("User", broker));
mqttPass = jsonReadStr(configSetupJson, getParamName("Pass", broker));
return true;
}
boolean mqttConnect() {
SerialPrint("I", "MQTT", String("use ") + (activeBroker == MQTT_PRIMARY ? "primary" : "reserve"));
if (!checkBrokerParams(activeBroker)) {
SerialPrint("E", "MQTT", "empty broker address");
return false;
}
readBrokerParams(activeBroker);
SerialPrint("I", "MQTT", "start connection");
mqttPrefix = jsonReadStr(configSetupJson, "mqttPrefix");
mqttRootDevice = mqttPrefix + "/" + chipId;
SerialPrint("I", "MQTT", "broker " + mqttServer + ":" + String(mqttPort, DEC));
SerialPrint("I", "MQTT", "topic " + mqttRootDevice);
setLedStatus(LED_FAST);
mqtt.setServer(mqttServer.c_str(), mqttPort);
bool res = false;
if (!mqtt.connected()) {
if (mqtt.connect(chipId.c_str(), mqttUser.c_str(), mqttPass.c_str())) {
SerialPrint("I", "MQTT", "connected");
setLedStatus(LED_OFF);
mqttSubscribe();
res = true;
} else {
SerialPrint("E", "MQTT", "could't connect, retry in " + String(MQTT_RECONNECT_INTERVAL / 1000) + "s");
setLedStatus(LED_FAST);
}
}
return res;
}
void mqttCallback(char* topic, uint8_t* payload, size_t length) {
String topicStr = String(topic);
//SerialPrint("I", "=>MQTT", topicStr);
String payloadStr;
payloadStr.reserve(length + 1);
for (size_t i = 0; i < length; i++) {
payloadStr += (char)payload[i];
}
//SerialPrint("I", "=>MQTT", payloadStr);
if (payloadStr.startsWith("HELLO")) {
SerialPrint("I", "MQTT", "Full update");
publishWidgets();
publishState();
choose_log_date_and_send();
}
else if (topicStr.indexOf("control") != -1) {
String key = selectFromMarkerToMarker(topicStr, "/", 3);
String order;
order += key;
order += " ";
order += payloadStr;
order += ",";
loopCmdAdd(order);
SerialPrint("I", "=>MQTT", "Msg from iotmanager app: " + key + " " + payloadStr);
}
else if (topicStr.indexOf("event") != -1) {
if (!jsonReadBool(configSetupJson, "MqttIn")) {
return;
}
if (topicStr.indexOf(chipId) == -1) {
String devId = selectFromMarkerToMarker(topicStr, "/", 2);
String key = selectFromMarkerToMarker(topicStr, "/", 3);
SerialPrint("I", "=>MQTT", "Received event from other device: '" + devId + "' " + key + " " + payloadStr);
String event = key + " " + payloadStr + ",";
eventBuf += event;
}
}
else if (topicStr.indexOf("order") != -1) {
if (!jsonReadBool(configSetupJson, "MqttIn")) {
return;
}
String devId = selectFromMarkerToMarker(topicStr, "/", 2);
String key = selectFromMarkerToMarker(topicStr, "/", 3);
SerialPrint("I", "=>MQTT", "Received direct order " + key + " " + payloadStr);
String order = key + " " + payloadStr + ",";
orderBuf += order;
}
else if (topicStr.indexOf("info") != -1) {
if (topicStr.indexOf("scen") != -1) {
writeFile(String(DEVICE_SCENARIO_FILE), payloadStr);
loadScenario();
SerialPrint("I", "=>MQTT", "Scenario received");
}
}
}
boolean publish(const String& topic, const String& data) {
if (mqtt.beginPublish(topic.c_str(), data.length(), false)) {
mqtt.print(data);
return mqtt.endPublish();
}
return false;
}
boolean publishData(const String& topic, const String& data) {
String path = mqttRootDevice + "/" + topic;
if (!publish(path, data)) {
SerialPrint("[E]", "MQTT", "on publish data");
return false;
}
return true;
}
boolean publishChart(const String& topic, const String& data) {
String path = mqttRootDevice + "/" + topic + "/status";
if (!publish(path, data)) {
SerialPrint("[E]", "MQTT", "on publish chart");
return false;
}
return true;
}
boolean publishControl(String id, String topic, String state) {
String path = mqttPrefix + "/" + id + "/" + topic + "/control";
return mqtt.publish(path.c_str(), state.c_str(), false);
}
boolean publishChart_test(const String& topic, const String& data) {
String path = mqttRootDevice + "/" + topic + "/status";
return mqtt.publish(path.c_str(), data.c_str(), false);
}
boolean publishStatus(const String& topic, const String& data) {
String path = mqttRootDevice + "/" + topic + "/status";
String json = "{}";
jsonWriteStr(json, "status", data);
return mqtt.publish(path.c_str(), json.c_str(), false);
}
boolean publishEvent(const String& topic, const String& data) {
String path = mqttRootDevice + "/" + topic + "/event";
return mqtt.publish(path.c_str(), data.c_str(), true);
}
boolean publishInfo(const String& topic, const String& data) {
String path = mqttRootDevice + "/" + topic + "/info";
return mqtt.publish(path.c_str(), data.c_str(), false);
}
#ifdef LAYOUT_IN_RAM
void publishWidgets() {
if (all_widgets != "") {
int counter = 0;
String line;
int psn_1 = 0;
int psn_2;
do {
psn_2 = all_widgets.indexOf("\r\n", psn_1); //\r\n
line = all_widgets.substring(psn_1, psn_2);
line.replace("\n", "");
line.replace("\r\n", "");
//jsonWriteStr(line, "id", String(counter));
//jsonWriteStr(line, "pageId", String(counter));
counter++;
sendMQTT("config", line);
Serial.println("[V] " + line);
psn_1 = psn_2 + 1;
} while (psn_2 + 2 < all_widgets.length());
getMemoryLoad("I after send all widgets");
}
}
#endif
#ifndef LAYOUT_IN_RAM
void publishWidgets() {
auto file = seekFile("layout.txt");
if (!file) {
SerialPrint("[E]", "MQTT", "no file layout.txt");
return;
}
while (file.available()) {
String payload = file.readStringUntil('\n');
SerialPrint("I", "MQTT", "widgets: " + payload);
publishData("config", payload);
}
file.close();
}
#endif
void publishState() {
// берет строку json и ключи превращает в топики а значения колючей в них посылает
String str = configLiveJson + "," + configStoreJson;
str.replace("{", "");
str.replace("}", "");
str += ",";
while (str.length()) {
String tmp = selectToMarker(str, ",");
String topic = selectToMarker(tmp, "\":");
topic.replace("\"", "");
String state = selectToMarkerLast(tmp, "\":");
state.replace("\"", "");
if (topic != "timenow") {
publishStatus(topic, state);
}
str = deleteBeforeDelimiter(str, ",");
}
}
const String getStateStr() {
switch (mqtt.state()) {
case -4:
return F("no respond");
break;
case -3:
return F("connection was broken");
break;
case -2:
return F("connection failed");
break;
case -1:
return F("client disconnected");
break;
case 0:
return F("client connected");
break;
case 1:
return F("doesn't support the requested version");
break;
case 2:
return F("rejected the client identifier");
break;
case 3:
return F("unable to accept the connection");
break;
case 4:
return F("wrong username/password");
break;
case 5:
return F("not authorized to connect");
break;
default:
return F("unspecified");
break;
}
}