src/mqtt_handler.cpp - master - ellersseer/MyCheat

src/mqtt_handler.cpp
280 строк9 KB
Dmitry Zhiltsov
feat(logs): Добавить логи в WEB
20 фев 2026, 20:39
20 фев 2026, 20:39dae6bcf
#include "mqtt_handler.h"
#include "app_state.h"
#include "sensor_api.h"
#include "wifi_manager.h"
#include "debug_log.h"
#include "log_buffer.h"

#include "esp_task_wdt.h"
#include <ArduinoJson.h>

// ============================================
// MQTT Setup
// ============================================
void setupMQTT()
{
    mqttClient.setServer(mqttSettings.server, mqttSettings.port);
    mqttClient.setCallback(mqttCallback);
    mqttClient.setBufferSize(512);
    logCapture.printf("[MQTT] Configured for %s:%d\n", mqttSettings.server, mqttSettings.port);
}

// ============================================
// MQTT Callback
// ============================================
void mqttCallback(char *topic, byte *payload, unsigned int length)
{
    char message[length + 1];
    memcpy(message, payload, length);
    message[length] = '\0';

    String topicStr = String(topic);
    logCapture.printf("[MQTT] %s = %s\n", topic, message);

    // Update stats
    if (xSemaphoreTake(statsMutex, pdMS_TO_TICKS(10)) == pdTRUE)
    {
        stats.mqttMessages++;
        xSemaphoreGive(statsMutex);
    }

    // Parse topic: {topic_prefix}/set/{index}
    String prefix = String(mqttSettings.topic_prefix) + "/set/";
    if (topicStr.startsWith(prefix))
    {
        int sensorIndex = topicStr.substring(prefix.length()).toInt();

        if (sensorIndex >= 0 && sensorIndex < NUM_SENSORS)
        {
            SensorCommand_t cmd;
            cmd.type = SensorCommandType::SET_TEMPERATURE;
            cmd.sensorIndex = sensorIndex;
            cmd.temperature = atof(message);

            xQueueSend(sensorCommandQueue, &cmd, pdMS_TO_TICKS(100));

            // Publish updated state
            String stateTopic = String(mqttSettings.topic_prefix) + "/state/" + String(sensorIndex);
            mqttClient.publish(stateTopic.c_str(), String(cmd.temperature, 2).c_str(), true);
        }
    }
    // Command topic for bulk operations
    else if (topicStr == String(mqttSettings.topic_prefix) + "/cmd")
    {
        JsonDocument doc;
        DeserializationError error = deserializeJson(doc, message);

        if (!error && doc["temps"].is<JsonArray>())
        {
            JsonArray temps = doc["temps"].as<JsonArray>();

            SensorCommand_t cmd;
            cmd.type = SensorCommandType::SET_ALL_TEMPS;

            int i = 0;
            for (JsonVariant v : temps)
            {
                if (i < NUM_SENSORS)
                {
                    cmd.allTemperatures[i] = v.as<float>();
                    i++;
                }
            }
            // Fill remaining with current values
            float currentTemps[NUM_SENSORS];
            getAllTemperatures(currentTemps);
            for (; i < NUM_SENSORS; i++)
            {
                cmd.allTemperatures[i] = currentTemps[i];
            }

            xQueueSend(sensorCommandQueue, &cmd, pdMS_TO_TICKS(100));
        }
    }
}

// ============================================
// MQTT Helper Functions
// ============================================
void publishAllTemperatures()
{
    float temps[NUM_SENSORS];
    getAllTemperatures(temps);
    for (int i = 0; i < NUM_SENSORS; i++)
    {
        String topic = String(mqttSettings.topic_prefix) + "/state/" + String(i);
        mqttClient.publish(topic.c_str(), String(temps[i], 2).c_str(), true);
    }
}

// ============================================
// MQTT Task
// ============================================
void mqttTask(void *pvParameters)
{
    logCapture.printf("[MQTT] Task started on Core %d\n", xPortGetCoreID());

    // Register with Task WDT for crash protection
    esp_task_wdt_add(NULL);

    setupMQTT();

    TickType_t lastReconnect = 0;
    TickType_t lastPublish = 0;

    while (true)
    {
        // Feed Task WDT at start of each iteration
        esp_task_wdt_reset();

        TickType_t now = xTaskGetTickCount();

        // Check for reconfiguration request
        if (mqttNeedsReconfigure)
        {
            logCapture.println("[MQTT] Reconfiguring with new settings...");
            mqttClient.disconnect();
            mqttConnected = false;

            // Reconfigure with new settings
            setupMQTT();
            mqttNeedsReconfigure = false;

            // Force immediate reconnection attempt
            lastReconnect = 0;

            logCapture.println("[MQTT] Reconfiguration complete");
        }

        // Check WiFi - only connect MQTT when in STA mode (not AP mode)
        bool wasWifiConnected = wifiConnected;
        wifiConnected = wifiManager.isConnected();

        // Log WiFi state changes
        if (wifiConnected && !wasWifiConnected)
        {
            DEBUG_EVENT(EVT_WIFI_CONNECT, wifiManager.getIP().toString().c_str());
        }
        else if (!wifiConnected && wasWifiConnected)
        {
            DEBUG_EVENT(EVT_WIFI_DISCONNECT, "WiFi lost");
        }

        // Check if MQTT is disabled or not configured
        if (!mqttSettings.enabled || strlen(mqttSettings.server) == 0)
        {
            if (mqttClient.connected())
            {
                mqttClient.disconnect();
            }
            mqttConnected = false;
            vTaskDelay(pdMS_TO_TICKS(1000));
            continue;
        }

        if (!wifiConnected)
        {
            mqttConnected = false;
            vTaskDelay(pdMS_TO_TICKS(1000));
            continue;
        }

        // Handle MQTT connection
        if (!mqttClient.connected())
        {
            mqttConnected = false;

            if ((now - lastReconnect) >= pdMS_TO_TICKS(MQTT_RECONNECT_INTERVAL))
            {
                lastReconnect = now;

                logCapture.printf("[MQTT] Connecting to %s:%d...\n", mqttSettings.server, mqttSettings.port);

                // Set socket timeout to 10s to prevent blocking longer than WDT timeout (30s)
                wifiClient.setTimeout(10);

                // Build will topic with defensive length check
                String willTopic = String(mqttSettings.topic_prefix) + "/status";
                if (willTopic.length() == 0 || willTopic.length() > 128)
                {
                    logCapture.println("[MQTT] Invalid will topic, skipping connect");
                    continue;
                }
                bool connected;

                if (strlen(mqttSettings.user) > 0)
                {
                    connected = mqttClient.connect(mqttSettings.client_id, mqttSettings.user, mqttSettings.password,
                                                   willTopic.c_str(), 0, true, "offline");
                }
                else
                {
                    connected = mqttClient.connect(mqttSettings.client_id,
                                                   willTopic.c_str(), 0, true, "offline");
                }

                if (connected)
                {
                    logCapture.println("[MQTT] Connected!");
                    mqttConnected = true;
                    DEBUG_EVENT(EVT_MQTT_CONNECT, mqttSettings.server);

                    // Publish online status
                    mqttClient.publish(willTopic.c_str(), "online", true);

                    // Subscribe to topics
                    String setTopic = String(mqttSettings.topic_prefix) + "/set/#";
                    mqttClient.subscribe(setTopic.c_str());

                    String cmdTopic = String(mqttSettings.topic_prefix) + "/cmd";
                    mqttClient.subscribe(cmdTopic.c_str());

                    // Publish initial states
                    publishAllTemperatures();
                }
                else
                {
                    logCapture.printf("[MQTT] Failed (rc=%d)\n", mqttClient.state());
                    DEBUG_EVENTF(EVT_MQTT_ERROR, "Connect failed rc=%d", mqttClient.state());
                }
            }
        }
        else
        {
            mqttConnected = true;
            mqttClient.loop();

            // Periodic state publish
            if ((now - lastPublish) >= pdMS_TO_TICKS(STATUS_PUBLISH_INTERVAL))
            {
                lastPublish = now;
                publishAllTemperatures();
            }
        }

        vTaskDelay(pdMS_TO_TICKS(100));
    }
}

// ============================================
// Dynamic MQTT Task Start
// ============================================
void startMqttTask()
{
    if (mqttTaskHandle != NULL)
    {
        logCapture.println("[MQTT] Task already running");
        return;
    }

    xTaskCreatePinnedToCore(
        mqttTask,
        "MQTT",
        TASK_STACK_MQTT,
        NULL,
        TASK_PRIORITY_MQTT,
        &mqttTaskHandle,
        TASK_CORE_MQTT);

    logCapture.println("[MQTT] Task started dynamically");
}