282 lines
9.0 KiB
C++
282 lines
9.0 KiB
C++
// mqtt_client_wrapper.cpp
|
|
#include "mqtt_client_wrapper.hpp"
|
|
#include <thread>
|
|
#include <chrono>
|
|
#include <nlohmann/json.hpp>
|
|
|
|
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);
|
|
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;
|
|
// 启动推流：挂载本地配置中 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)
|
|
{
|
|
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);
|
|
|
|
// 停止所有流，相当于复位
|
|
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())
|
|
{
|
|
try
|
|
{
|
|
mqtt_client->connect();
|
|
}
|
|
catch (...)
|
|
{
|
|
}
|
|
|
|
// 更频繁地检查退出标志
|
|
for (int i = 0; i < 10 && g_running && !mqtt_client->isConnected(); i++)
|
|
{
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(200));
|
|
}
|
|
}
|
|
|
|
while (g_running && mqtt_client->isConnected())
|
|
{
|
|
try
|
|
{
|
|
send_heartbeat();
|
|
}
|
|
catch (const std::exception &e)
|
|
{
|
|
LOG_ERROR("[MQTT] Heartbeat error: " + std::string(e.what()));
|
|
}
|
|
// 将长睡眠拆分为多个短睡眠，以便更频繁检查退出标志
|
|
auto sleep_time = heartbeat_interval;
|
|
while (sleep_time.count() > 0 && g_running && mqtt_client->isConnected())
|
|
{
|
|
auto chunk = std::min(sleep_time, std::chrono::milliseconds(200));
|
|
std::this_thread::sleep_for(chunk);
|
|
sleep_time -= chunk;
|
|
}
|
|
}
|
|
|
|
// 清理资源
|
|
if (mqtt_client)
|
|
{
|
|
// 只有在运行标志仍然为true时才尝试正常断开
|
|
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.");
|
|
}
|