yituo_video/src/mqtt_client_wrapper.cpp

// mqtt_client_wrapper.cpp
#include "mqtt_client_wrapper.hpp"
#include <thread>
#include <chrono>
#include <nlohmann/json.hpp>
#include <algorithm>

std::shared_ptr<MQTTClient> mqtt_client;
std::atomic<bool> mqtt_restart_required{false};
extern std::atomic<bool> g_running;
std::atomic<bool> g_streaming{false};
std::string g_dispatch_id;
std::mutex g_dispatch_id_mutex;

static void send_heartbeat()
{
    if (!mqtt_client || !mqtt_client->isConnected())
        return;

    nlohmann::json hb_data;
    hb_data["time"] = Logger::get_current_time_utc8();
    hb_data["status"] = g_streaming ? 0 : 2;

    {
        std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
        hb_data["dispatchId"] = g_streaming ? g_dispatch_id : "";
    }

    nlohmann::json msg;
    msg["data"] = hb_data;
    msg["isEnc"] = 0;
    msg["type"] = 0;

    mqtt_client->publish(g_app_config.mqtt.topics.heartbeat_up, msg.dump());
}

// MQTT 回调
static void on_mqtt_connected()
{
    LOG_INFO("[MQTT] Connected to broker: " + g_app_config.mqtt.server_ip);

    // --- ★关键！MQTT 重连后立即执行全局复位 ---
    g_streaming = false; // 清除推流状态
    {
        std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
        g_dispatch_id.clear(); // 清除旧任务 ID
    }

    // 停掉所有流，清理 RTSP 状态
    for (const auto &cam : g_app_config.cameras)
    {
        if (RTSPManager::is_streaming(cam.name))
        {
            RTSPManager::unmount_camera(cam);
            LOG_WARN("[MQTT] Reconnected -> force unmount camera: " + cam.name);
        }
    }
    // -------------------------------------------------------

    send_heartbeat();

    const auto &topics = g_app_config.mqtt.topics;
    mqtt_client->subscribe(topics.video_down);
    mqtt_client->subscribe(topics.substream_down);
    mqtt_client->subscribe(topics.reset_down);
}

static void on_mqtt_disconnected()
{
    LOG_WARN("[MQTT] Disconnected from broker: " + g_app_config.mqtt.server_ip);
}

static void on_mqtt_message_received(const std::string &topic, const std::string &message)
{
    LOG_INFO("[MQTT] Received message on topic [" + topic + "], len = " + std::to_string(message.size()));
    LOG_INFO("[MQTT] Message content: " + message); // 打印实际内容

    try
    {
        if (topic == g_app_config.mqtt.topics.video_down)
        {
            // 处理 video_down
            auto j = nlohmann::json::parse(message);

            if (!j.contains("data") || !j["data"].contains("status") || !j["data"].contains("seqNo"))
            {
                LOG_WARN("[MQTT] video_down JSON missing required fields");
                return;
            }

            // 写 dispatchId 并设置 streaming 状态
            {
                std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
                if (j["data"].contains("dispatchId"))
                    g_dispatch_id = j["data"]["dispatchId"].get<std::string>();
            }

            int status = j["data"]["status"].get<int>();

            std::string seqNo = j["data"]["seqNo"].get<std::string>();

            bool success = true; // 标记是否操作成功

            if (status == 0)
            {
                g_streaming = true;

                for (const auto &cam : g_app_config.cameras)
                {
                    if (!cam.enabled)
                        continue;

                    // 不再提前卸载
                    RTSPManager::mount_camera(cam);
                    LOG_INFO("[MQTT] Started streaming: " + cam.name);
                }
            }
            else if (status == 1)
            {
                g_streaming = false;
                std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
                g_dispatch_id.clear(); // 停止拉流就清空 dispatchId
                // 停止推流：卸载本地配置中 enabled 的摄像头
                for (const auto &cam : g_app_config.cameras)
                {
                    if (!cam.enabled)
                        continue;

                    if (RTSPManager::is_streaming(cam.name))
                    {
                        RTSPManager::unmount_camera(cam);
                        LOG_INFO("[MQTT] Stopped streaming: " + cam.name);
                    }
                }
            }
            else
            {
                LOG_WARN("[MQTT] video_down: unknown status value " + std::to_string(status));
                success = false;
            }

            // 获取当前时间 yyyyMMddHHmmssSSS (UTC+8)
            std::string time_str = Logger::get_current_time_utc8();

            nlohmann::json reply_data;
            reply_data["time"] = time_str;
            reply_data["result"] = success ? 0 : 1;
            reply_data["seqNo"] = seqNo;

            // 封装外层
            nlohmann::json reply;
            reply["data"] = reply_data;
            reply["isEnc"] = 0;
            reply["type"] = 0;

            // 发送应答
            if (mqtt_client)
            {
                mqtt_client->publish(g_app_config.mqtt.topics.video_down_ack, reply.dump());
                LOG_INFO("[MQTT] Replied to video_down: " + reply.dump());
            }
        }
        else if (topic == g_app_config.mqtt.topics.substream_down)
        {
            // 处理 substream_down
            LOG_INFO("[MQTT] substream_down message received (not implemented yet).");
        }
        else if (topic == g_app_config.mqtt.topics.reset_down)
        {
            auto j = nlohmann::json::parse(message);

            // reset/down 的 payload 在 data 里
            auto data = j.contains("data") ? j["data"] : nlohmann::json::object();

            std::string seqNo = data.value("seqNo", "");
            std::string errCode = data.value("errorCode", "");
            std::string des = data.value("des", "");

            LOG_WARN("[MQTT] Reset command received, errorCode=" + errCode + ", des=" + des);

            // --- ★★ 新增：重置全局 streaming 状态与 dispatchId ---
            g_streaming = false;
            {
                std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
                g_dispatch_id.clear();
            }
            // ------------------------------------------------------

            // 停止所有流，相当于复位
            for (const auto &cam : g_app_config.cameras)
            {
                if (RTSPManager::is_streaming(cam.name))
                {
                    RTSPManager::unmount_camera(cam);
                    LOG_INFO("[RTSP] Camera " + cam.name + " reset/unmounted");
                }
            }

            // 组装应答 data
            nlohmann::json reply_data;
            reply_data["time"] = Logger::get_current_time_utc8();
            reply_data["result"] = 0; // 0=成功
            reply_data["seqNo"] = seqNo;

            // 外层封装
            nlohmann::json reply;
            reply["data"] = reply_data;
            reply["isEnc"] = 0;
            reply["type"] = 0;

            if (mqtt_client)
            {
                mqtt_client->publish(g_app_config.mqtt.topics.reset_down_ack, reply.dump());
                LOG_INFO("[MQTT] Replied to reset_down: " + reply.dump());
            }
        }
        else
        {
            LOG_WARN("[MQTT] Unknown topic: " + topic);
        }
    }
    catch (const std::exception &e)
    {
        LOG_ERROR(std::string("[MQTT] Failed to process incoming JSON: ") + e.what());
    }
}

void mqtt_client_thread_func()
{
    const auto &cfg = g_app_config.mqtt;
    auto heartbeat_interval = std::chrono::milliseconds(static_cast<int>(cfg.keep_alive * 0.9));

    while (g_running)
    {
        mqtt_client = std::make_unique<MQTTClient>(cfg);
        mqtt_client->setConnectCallback(on_mqtt_connected);
        mqtt_client->setDisconnectCallback(on_mqtt_disconnected);
        mqtt_client->setMessageCallback(on_mqtt_message_received);

        // 等待连接
        while (g_running && !mqtt_client->isConnected())
        {
            mqtt_client->connect();
            for (int i = 0; i < 10 && g_running && !mqtt_client->isConnected(); i++)
                std::this_thread::sleep_for(std::chrono::milliseconds(50));

            if (!g_running && !mqtt_client->isConnected())
                mqtt_client->force_disconnect();
        }

        // 主循环：心跳
        while (g_running && mqtt_client->isConnected())
        {
            send_heartbeat();

            auto sleep_time = heartbeat_interval;
            while (sleep_time.count() > 0 && g_running && mqtt_client->isConnected())
            {
                auto chunk = std::min(sleep_time, std::chrono::milliseconds(50));
                std::this_thread::sleep_for(chunk);
                sleep_time -= chunk;
            }

            if (!g_running && mqtt_client->isConnected())
                mqtt_client->force_disconnect();
        }

        // 清理
        if (mqtt_client)
        {
            if (g_running)
                mqtt_client->disconnect();
            else
                mqtt_client->force_disconnect();
            mqtt_client.reset();
        }

        mqtt_restart_required = false;

        if (!g_running)
            break;

        // 短暂等待再重连
        for (int i = 0; i < 5 && g_running; i++)
            std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }
    LOG_INFO("[MQTT] Client thread exiting.");
}