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

// 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
    {
    }
    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)
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }

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