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

1

Browse Source
This commit is contained in:
cxh 2025-11-24 14:49:37 +08:00
parent 394520befb
commit 81877ced1b
3 changed files with 77 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
src/main.cpp
View File

@ -6,27 +6,23 @@
#include <thread>
#include <atomic>
#include <csignal>
#include <unistd.h> // write, STDOUT_FILENO
#include <unistd.h>
#include <chrono>
// 可通过这些开关快速启用/禁用线程进行调试
constexpr bool ENABLE_RTSP_THREAD = true; // 设置为 false 禁用 RTSP 线程
constexpr bool ENABLE_MQTT_THREAD = true; // 设置为 false 禁用 MQTT 线程
constexpr bool ENABLE_RTSP_THREAD = true;
constexpr bool ENABLE_MQTT_THREAD = true;
std::atomic<bool> g_running(true);
static void minimal_signal_handler(int signum)
{
// 只做非常有限且 async-signal-safe 的操作
g_running.store(false, std::memory_order_relaxed);
const char msg[] = "[MAIN] Signal received, initiating shutdown...\n";
write(STDERR_FILENO, msg, sizeof(msg) - 1);
// 不要调用 LOG_*、malloc、std::string、mutex、exit 等
}
int main()
{
// 安装信号处理（SIGINT 和 SIGTERM）
struct sigaction sigAct{};
sigAct.sa_handler = minimal_signal_handler;
sigemptyset(&sigAct.sa_mask);
@ -35,12 +31,10 @@ int main()
sigaction(SIGTERM, &sigAct, nullptr);
signal(SIGPIPE, SIG_IGN);
// 初始化日志文件
Logger::set_log_to_file(get_executable_dir_file_path("app.log"));
try
{
// 从可执行文件所在目录读取配置文件
g_app_config = AppConfig::load_from_file(get_executable_dir_file_path("config.json"));
}
catch (const std::exception &e)
@ -49,58 +43,64 @@ int main()
return -1;
}
// 线程退出标志，用于超时等待
std::atomic<bool> rtsp_thread_exited(false);
std::atomic<bool> mqtt_thread_exited(false);
// 先在主线程初始化 GStreamer
// 初始化 GStreamer
RTSPManager::init();
// ⭐ 程序启动 → 强制 mount 所有摄像头（RTSP 永远在线）
for (const auto &cam : g_app_config.cameras)
{
if (cam.enabled)
{
RTSPManager::mount_camera(cam);
LOG_INFO("[INIT] Mounted camera: " + cam.name);
}
}
std::thread rtsp_thread;
std::thread mqtt_thread;
// 启动 RTSP 线程（如果启用）
if (ENABLE_RTSP_THREAD)
{
rtsp_thread = std::thread([&]()
{
RTSPManager::start(g_app_config.cameras);
rtsp_thread_exited.store(true, std::memory_order_relaxed); });
LOG_INFO("[MAIN] RTSP thread started");
}
else
{
LOG_INFO("[MAIN] RTSP thread disabled by build-time toggle");
LOG_INFO("[MAIN] RTSP thread disabled by toggle");
}
// 启动 MQTT 线程（如果启用）
if (ENABLE_MQTT_THREAD)
{
mqtt_thread = std::thread([&]()
{
mqtt_client_thread_func();
mqtt_thread_exited.store(true, std::memory_order_relaxed); });
LOG_INFO("[MAIN] MQTT thread started");
}
else
{
LOG_INFO("[MAIN] MQTT thread disabled by build-time toggle");
LOG_INFO("[MAIN] MQTT thread disabled by toggle");
}
// 等待退出信号
while (g_running.load(std::memory_order_relaxed))
std::this_thread::sleep_for(std::chrono::milliseconds(200));
LOG_INFO("[MAIN] Shutdown requested, stopping services...");
// 等线程优雅退出：总等待时间 (可调整)
const auto max_wait = std::chrono::seconds(5);
const auto poll_interval = std::chrono::milliseconds(100);
auto deadline = std::chrono::steady_clock::now() + max_wait;
if (ENABLE_RTSP_THREAD)
{
RTSPManager::stop(); // 只负责让 g_main_loop_run 跳出
RTSPManager::stop();
if (rtsp_thread.joinable())
rtsp_thread.join();
@ -108,8 +108,7 @@ int main()
LOG_INFO("[MAIN] RTSP thread finished and joined.");
}
// 重置 MQTT 线程等待的截止时间
deadline = std::chrono::steady_clock::now() + max_wait;
auto deadline = std::chrono::steady_clock::now() + max_wait;
if (ENABLE_MQTT_THREAD)
{
@ -133,11 +132,10 @@ int main()
}
}
// 如果有线程仍未退出，则强制终止
bool any_failed = false;
if (ENABLE_RTSP_THREAD && rtsp_thread.joinable() && !rtsp_thread_exited.load(std::memory_order_relaxed))
if (ENABLE_RTSP_THREAD && rtsp_thread.joinable() && !rtsp_thread_exited.load())
any_failed = true;
if (ENABLE_MQTT_THREAD && mqtt_thread.joinable() && !mqtt_thread_exited.load(std::memory_order_relaxed))
if (ENABLE_MQTT_THREAD && mqtt_thread.joinable() && !mqtt_thread_exited.load())
any_failed = true;
if (any_failed)

234
src/mqtt_client_wrapper.cpp
View File

@ -6,11 +6,7 @@
#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()
{
@ -19,12 +15,8 @@ static void send_heartbeat()
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 : "";
}
hb_data["status"] = 0; // 固定上报正常
hb_data["dispatchId"] = ""; // 已无任务状态
nlohmann::json msg;
msg["data"] = hb_data;
@ -34,29 +26,11 @@ static void send_heartbeat()
mqtt_client->publish(g_app_config.mqtt.topics.heartbeat_up, msg.dump());
}
// MQTT 回调
// MQTT 连接成功
static void on_mqtt_connected()
{
LOG_INFO("[MQTT] Connected to broker: " + g_app_config.mqtt.server_ip);
// --- ★关键！MQTT 重连后立即执行全局复位 ---
g_streaming = false; // 清除推流状态
{
std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
g_dispatch_id.clear(); // 清除旧任务 ID
}
// 停掉所有流，清理 RTSP 状态
for (const auto &cam : g_app_config.cameras)
{
if (RTSPManager::is_streaming(cam.name))
{
RTSPManager::unmount_camera(cam);
LOG_WARN("[MQTT] Reconnected -> force unmount camera: " + cam.name);
}
}
// -------------------------------------------------------
send_heartbeat();
const auto &topics = g_app_config.mqtt.topics;
@ -65,170 +39,83 @@ static void on_mqtt_connected()
mqtt_client->subscribe(topics.reset_down);
}
// MQTT 连接断开
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); // 打印实际内容
LOG_INFO("[MQTT] Received message on topic [" + topic + "]");
LOG_INFO("[MQTT] Content: " + message);
try
{
// ------- video_down 应答 -------
if (topic == g_app_config.mqtt.topics.video_down)
{
// 处理 video_down
auto j = nlohmann::json::parse(message);
auto seqNo = j["data"].value("seqNo", "");
if (!j.contains("data") || !j["data"].contains("status") || !j["data"].contains("seqNo"))
{
LOG_WARN("[MQTT] video_down JSON missing required fields");
return;
}
LOG_INFO("[MQTT] video_down received, RTSP always running (no action).");
// 写 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;
for (const auto &cam : g_app_config.cameras)
{
if (!cam.enabled)
continue;
// 不再提前卸载
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);
// --- ★★ 新增：重置全局 streaming 状态与 dispatchId ---
g_streaming = false;
{
std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
g_dispatch_id.clear();
}
// ------------------------------------------------------
// 停止所有流，相当于复位
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["result"] = 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());
}
mqtt_client->publish(g_app_config.mqtt.topics.video_down_ack, reply.dump());
return;
}
else
// ------- substream_down 应答 -------
if (topic == g_app_config.mqtt.topics.substream_down)
{
LOG_WARN("[MQTT] Unknown topic: " + topic);
LOG_INFO("[MQTT] substream_down received (ignored).");
return;
}
// ------- reset/down 应答 -------
if (topic == g_app_config.mqtt.topics.reset_down)
{
auto j = nlohmann::json::parse(message);
auto seqNo = j["data"].value("seqNo", "");
LOG_WARN("[MQTT] reset/down received (ignored).");
nlohmann::json reply_data;
reply_data["time"] = Logger::get_current_time_utc8();
reply_data["result"] = 0;
reply_data["seqNo"] = seqNo;
nlohmann::json reply;
reply["data"] = reply_data;
reply["isEnc"] = 0;
reply["type"] = 0;
mqtt_client->publish(g_app_config.mqtt.topics.reset_down_ack, reply.dump());
return;
}
LOG_WARN("[MQTT] Unknown topic: " + topic);
}
catch (const std::exception &e)
{
LOG_ERROR(std::string("[MQTT] Failed to process incoming JSON: ") + e.what());
LOG_ERROR(std::string("[MQTT] JSON processing failed: ") + 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));
auto heartbeat_interval = std::chrono::milliseconds((int)(cfg.keep_alive * 0.9));
while (g_running)
{
@ -241,48 +128,35 @@ void mqtt_client_thread_func()
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 remaining = heartbeat_interval;
while (remaining.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;
auto step = std::min(remaining, std::chrono::milliseconds(50));
std::this_thread::sleep_for(step);
remaining -= step;
}
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->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.");
}

1
src/rtsp_manager.cpp
View File

@ -48,7 +48,6 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam)
gst_rtsp_media_factory_set_shared(factory, TRUE);
gst_rtsp_media_factory_set_suspend_mode(factory, GST_RTSP_SUSPEND_MODE_RESET);
// ⭐ 修复点：最后一个参数必须用 GConnectFlags 类型
g_signal_connect_data(factory,
"media-configure",
G_CALLBACK(on_media_created),