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

    const auto &topics = g_app_config.mqtt.topics;
    mqtt_client->subscribe(topics.video_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
    {
        auto j = nlohmann::json::parse(message);

        if (topic == g_app_config.mqtt.topics.video_down && j["type"] == "request")
        {
            std::string seqNo = j.value("seqNo", "");
            auto data = j["data"];

            int switch_val = data.value("switch", 0);
            int streamType = data.value("streamType", 0); // 0主 1子
            StreamType type = (streamType == 0) ? StreamType::MAIN : StreamType::SUB;
            auto channels = data.value("channels", std::vector<int>{});

            nlohmann::json resp_data = nlohmann::json::array();

            for (int ch : channels)
            {
                if (ch < 0 || ch >= static_cast<int>(g_app_config.cameras.size()))
                    continue;

                Camera &cam = g_app_config.cameras[ch];
                bool op_result = false;

                if (switch_val == 0)
                {
                    if (!RTMPManager::is_streaming(cam.name))
                    {
                        RTMPManager::start_camera(cam, type);
                        op_result = true;
                    }
                }
                else
                {
                    if (RTMPManager::is_streaming(cam.name))
                    {
                        RTMPManager::stop_camera(cam.name, type);
                        op_result = true;
                    }
                }

                nlohmann::json ch_resp;
                ch_resp["loc"] = ch;
                ch_resp["url"] = op_result ? RTMPManager::get_stream_url(cam.name, type) : "";
                ch_resp["result"] = op_result ? 0 : 1;
                ch_resp["reason"] = op_result ? "" : "already in requested state";

                resp_data.push_back(ch_resp);
            }

            nlohmann::json reply;
            reply["type"] = "response";
            reply["seqNo"] = seqNo;
            reply["data"] = resp_data;

            mqtt_client->publish(topic, reply.dump());
        }
    }
    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.");
}