yituo_video/src/mqtt_client_wrapper.cpp

// mqtt_client_wrapper.cpp
#include "mqtt_client_wrapper.hpp"

std::shared_ptr<MQTTClient> mqtt_client;
std::atomic<bool> mqtt_restart_required{false};

extern std::atomic<bool> g_running;

std::string g_dispatch_id;
std::mutex g_dispatch_id_mutex;

auto last_heartbeat = std::chrono::steady_clock::now();
auto heartbeat_interval = std::chrono::milliseconds(static_cast<int>(g_app_config.mqtt.keep_alive * 0.9));

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

    // 判断当前状态
    int status = 2; // 默认等待中
    for (const auto &cam : g_app_config.cameras)
    {
        if (cam.enabled && RTSPManager::is_streaming(cam.name))
        {
            status = 0; // 有推流摄像头
            break;
        }
    }

    nlohmann::json hb_data;
    hb_data["time"] = Logger::get_current_time_utc8();

    {
        std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
        if (!g_dispatch_id.empty())
            hb_data["dispatchId"] = g_dispatch_id;
    }

    hb_data["status"] = status;

    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());
    LOG_INFO("[MQTT] Sent heartbeat: " + msg.dump());
}

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

    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()));

    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 时加锁
            if (j["data"].contains("dispatchId"))
            {
                std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
                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)
            {
                // 启动推流：挂载本地配置中 enabled 的摄像头
                for (const auto &cam : g_app_config.cameras)
                {
                    if (!cam.enabled)
                        continue;

                    if (!RTSPManager::is_streaming(cam.name))
                    {
                        RTSPManager::mount_camera(cam);
                        LOG_INFO("[MQTT] Started streaming: " + cam.name);
                    }
                }
            }
            else if (status == 1)
            {
                // 停止推流：卸载本地配置中 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.substream_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);

            // 停止所有流，相当于复位
            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;

    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);

        mqtt_client->connect();

        while (!mqtt_restart_required && g_running)
        {
            auto now = std::chrono::steady_clock::now();
            if (now - last_heartbeat >= heartbeat_interval)
            {
                try
                {
                    send_heartbeat();
                }
                catch (const std::exception &e)
                {
                    LOG_ERROR(std::string("[MQTT] Heartbeat error: ") + e.what());
                }
                last_heartbeat = now;
            }

            std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 轻量轮询
        }

        // 需要重启或退出
        mqtt_client->disconnect();
        mqtt_client.reset();
        mqtt_restart_required = false;
    }
    LOG_INFO("[MQTT] Client thread exiting.");
}