cacto
988c574183
mqttClient.connect() блокирует на время TCP-таймаута, если брокер недоступен. Без паузы между попытками это происходило на каждой итерации loop() — веб-интерфейс и опрос датчиков/реле (не зависят от MQTT и должны работать даже без платформы) подвисали. Теперь переподключение пробуется не чаще раза в 5 секунд. В README добавлен раздел "Автономная работа без платформы" — что именно работает локально без docker compose, а что требует брокера.
115 lines
3.5 KiB
C++
115 lines
3.5 KiB
C++
#include "mqtt_link.h"
|
|
|
|
#include <ArduinoJson.h>
|
|
#include <PubSubClient.h>
|
|
#include <WiFi.h>
|
|
|
|
#include "config.h"
|
|
#include "devices.h"
|
|
#include "time_sync.h"
|
|
|
|
static WiFiClient wifiClient;
|
|
static PubSubClient mqttClient(wifiClient);
|
|
|
|
// devices/{device_id}/commands -> device_id, или "" если топик не подходит.
|
|
static String deviceIdFromCommandTopic(const String& topic) {
|
|
if (!topic.startsWith("devices/") || !topic.endsWith("/commands")) {
|
|
return "";
|
|
}
|
|
return topic.substring(8, topic.length() - 9); // strip "devices/" and "/commands"
|
|
}
|
|
|
|
void mqttPublishAck(const String& deviceId, JsonDocument& state) {
|
|
JsonDocument ackDoc;
|
|
ackDoc["state"] = state.as<JsonObject>();
|
|
|
|
char buf[256];
|
|
size_t len = serializeJson(ackDoc, buf, sizeof(buf));
|
|
|
|
String topic = "devices/" + deviceId + "/ack";
|
|
mqttClient.publish(topic.c_str(), (const uint8_t*)buf, len, false);
|
|
}
|
|
|
|
static void onMessage(char* topic, byte* payload, unsigned int length) {
|
|
String deviceId = deviceIdFromCommandTopic(String(topic));
|
|
if (deviceId.isEmpty()) {
|
|
return;
|
|
}
|
|
|
|
JsonDocument cmd;
|
|
DeserializationError err = deserializeJson(cmd, payload, length);
|
|
if (err) {
|
|
Serial.print("[mqtt] invalid command payload on ");
|
|
Serial.println(topic);
|
|
return;
|
|
}
|
|
|
|
String action = cmd["action"] | "";
|
|
double level = cmd["level"] | 0.0;
|
|
if (action.isEmpty()) {
|
|
Serial.print("[mqtt] command missing action on ");
|
|
Serial.println(topic);
|
|
return;
|
|
}
|
|
|
|
JsonDocument state;
|
|
if (devicesApplyCommand(deviceId, action, level, state)) {
|
|
mqttPublishAck(deviceId, state);
|
|
}
|
|
}
|
|
|
|
void mqttBegin() {
|
|
mqttClient.setServer(MQTT_BROKER_HOST, MQTT_BROKER_PORT);
|
|
mqttClient.setCallback(onMessage);
|
|
}
|
|
|
|
// mqttClient.connect() блокирует на время TCP-таймаута, если брокер
|
|
// недоступен. Без паузы между попытками это съедало бы почти весь loop() —
|
|
// а значит и веб-интерфейс, и опрос датчиков/реле, которые от MQTT не
|
|
// зависят и обязаны работать даже без связи с платформой (см. README,
|
|
// "Автономная работа без платформы").
|
|
static unsigned long lastConnectAttempt = 0;
|
|
static const unsigned long RECONNECT_INTERVAL_MS = 5000;
|
|
|
|
void mqttEnsureConnected() {
|
|
if (mqttClient.connected()) {
|
|
return;
|
|
}
|
|
|
|
unsigned long now = millis();
|
|
if (now - lastConnectAttempt < RECONNECT_INTERVAL_MS && lastConnectAttempt != 0) {
|
|
return;
|
|
}
|
|
lastConnectAttempt = now;
|
|
|
|
Serial.print("[mqtt] connecting to ");
|
|
Serial.print(MQTT_BROKER_HOST);
|
|
if (mqttClient.connect(MQTT_CLIENT_ID)) {
|
|
Serial.println(" connected");
|
|
mqttClient.subscribe("devices/+/commands", 1);
|
|
} else {
|
|
Serial.print(" failed, rc=");
|
|
Serial.print(mqttClient.state());
|
|
Serial.println(" (это ок, если платформа сейчас не поднята — веб-интерфейс и реле работают локально)");
|
|
}
|
|
}
|
|
|
|
void mqttLoop() {
|
|
mqttClient.loop();
|
|
}
|
|
|
|
void mqttPublishTelemetry(const String& deviceId, const String& sensorType, double value) {
|
|
JsonDocument doc;
|
|
doc["device_id"] = deviceId;
|
|
doc["zone_id"] = ZONE_ID;
|
|
doc["sensor_type"] = sensorType;
|
|
doc["value"] = value;
|
|
doc["timestamp"] = nowRFC3339();
|
|
|
|
char buf[256];
|
|
size_t len = serializeJson(doc, buf, sizeof(buf));
|
|
|
|
String topic = "devices/" + deviceId + "/telemetry";
|
|
mqttClient.publish(topic.c_str(), (const uint8_t*)buf, len, false);
|
|
}
|