cxh/kunlang_video
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
cxh 2025-09-09 16:45:54 +08:00
parent 0acb25669b
commit f571282c85
1 changed files with 15 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/mqtt_client_wrapper.cpp
View File

@ -6,6 +6,7 @@ std::atomic<bool> mqtt_restart_required{false};
extern std::atomic<bool> g_running; extern std::atomic<bool> g_running;
std::atomic<bool> g_streaming{false}; // 当前是否有摄像头在拉流
std::string g_dispatch_id; std::string g_dispatch_id;
std::mutex g_dispatch_id_mutex; std::mutex g_dispatch_id_mutex;
@ -14,22 +15,13 @@ static void send_heartbeat()
if (!mqtt_client || !mqtt_client->isConnected()) if (!mqtt_client || !mqtt_client->isConnected())
return; 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; nlohmann::json hb_data;
hb_data["time"] = Logger::get_current_time_utc8(); // UTC+8 hb_data["time"] = Logger::get_current_time_utc8(); // UTC+8
if (!g_dispatch_id.empty())
hb_data["dispatchId"] = g_dispatch_id; // 状态: 2=没有拉流, 0=正在拉流
hb_data["status"] = status; hb_data["status"] = g_streaming ? 0 : 2;
hb_data["dispatchId"] = g_streaming ? g_dispatch_id : "";
nlohmann::json msg; nlohmann::json msg;
msg["data"] = hb_data; msg["data"] = hb_data;
@ -73,19 +65,22 @@ static void on_mqtt_message_received(const std::string &topic, const std::string
return; return;
} }
// 写入 dispatchId 时加锁 // 写 dispatchId 并设置 streaming 状态
if (j["data"].contains("dispatchId"))
{ {
std::lock_guard<std::mutex> lock(g_dispatch_id_mutex); std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
if (j["data"].contains("dispatchId"))
g_dispatch_id = j["data"]["dispatchId"].get<std::string>(); g_dispatch_id = j["data"]["dispatchId"].get<std::string>();
} }
int status = j["data"]["status"].get<int>(); int status = j["data"]["status"].get<int>();
std::string seqNo = j["data"]["seqNo"].get<std::string>(); std::string seqNo = j["data"]["seqNo"].get<std::string>();
bool success = true; // 标记是否操作成功 bool success = true; // 标记是否操作成功
if (status == 0) if (status == 0)
{ {
g_streaming = true;
// 启动推流：挂载本地配置中 enabled 的摄像头 // 启动推流：挂载本地配置中 enabled 的摄像头
for (const auto &cam : g_app_config.cameras) for (const auto &cam : g_app_config.cameras)
{ {
@ -101,6 +96,9 @@ static void on_mqtt_message_received(const std::string &topic, const std::string
} }
else if (status == 1) else if (status == 1)
{ {
g_streaming = false;
std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
g_dispatch_id.clear(); // 停止拉流就清空 dispatchId
// 停止推流：卸载本地配置中 enabled 的摄像头 // 停止推流：卸载本地配置中 enabled 的摄像头
for (const auto &cam : g_app_config.cameras) for (const auto &cam : g_app_config.cameras)
{ {