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

temp

Browse Source
This commit is contained in:
cxh 2025-10-14 17:40:48 +08:00
parent ec4790e65d
commit 5137f45e30
1 changed files with 43 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
src/mqtt_client_wrapper.cpp
View File

@ -57,79 +57,63 @@ static void on_mqtt_message_received(const std::string &topic, const std::string
{ {
auto j = nlohmann::json::parse(message); auto j = nlohmann::json::parse(message);
if (topic == g_app_config.mqtt.topics.video_down) if (topic == g_app_config.mqtt.topics.video_down && j["type"] == "request")
{ {
if (topic.find("/kun/vehicle/video/push/") != std::string::npos && 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)
{ {
std::string seqNo = j.value("seqNo", ""); if (ch < 0 || ch >= static_cast<int>(g_app_config.cameras.size()))
auto data = j["data"]; continue;
int switch_val = data.value("switch", 0); Camera &cam = g_app_config.cameras[ch];
int streamType = data.value("streamType", 0); // 0主 1子 bool op_result = false;
StreamType type = (streamType == 0) ? StreamType::MAIN : StreamType::SUB;
auto channels = data.value("channels", std::vector<int>{});
nlohmann::json resp_data = nlohmann::json::array(); if (switch_val == 0)
for (int ch : channels)
{ {
// 用数组索引找到摄像头 if (!RTMPManager::is_streaming(cam.name))
if (ch < 0 || ch >= static_cast<int>(g_app_config.cameras.size()))
{ {
LOG_WARN("[MQTT] Invalid channel: " + std::to_string(ch)); RTMPManager::start_camera(cam, type);
continue; op_result = true;
} }
}
Camera &cam = g_app_config.cameras[ch]; else
{
bool op_result = false; if (RTMPManager::is_streaming(cam.name))
if (switch_val == 0)
{ {
// 启动流 RTMPManager::stop_camera(cam.name, type);
if (!RTMPManager::is_streaming(cam.name, type)) op_result = true;
{
RTMPManager::start_camera(cam, type);
op_result = true;
}
} }
else
{
// 停止流
if (RTMPManager::is_streaming(cam.name, type))
{
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 ch_resp;
nlohmann::json reply; ch_resp["loc"] = ch;
reply["type"] = "response"; ch_resp["url"] = op_result ? RTMPManager::get_stream_url(cam.name, type) : "";
reply["seqNo"] = seqNo; ch_resp["result"] = op_result ? 0 : 1;
reply["data"] = resp_data; ch_resp["reason"] = op_result ? "" : "already in requested state";
mqtt_client->publish(topic, reply.dump()); resp_data.push_back(ch_resp);
LOG_INFO("[MQTT] Replied to video push request: " + reply.dump());
}
else
{
LOG_WARN("[MQTT] Unknown topic or non-request type: " + topic);
} }
nlohmann::json reply;
reply["type"] = "response";
reply["seqNo"] = seqNo;
reply["data"] = resp_data;
mqtt_client->publish(topic, reply.dump());
} }
catch (const std::exception &e) }
{ catch (const std::exception &e)
LOG_ERROR(std::string("[MQTT] Failed to process incoming JSON: ") + e.what()); {
} LOG_ERROR(std::string("[MQTT] Failed to process incoming JSON: ") + e.what());
} }
} }