diff --git a/src/mqtt_client_wrapper.cpp b/src/mqtt_client_wrapper.cpp index cdf084a..16b2be7 100644 --- a/src/mqtt_client_wrapper.cpp +++ b/src/mqtt_client_wrapper.cpp @@ -1,22 +1,22 @@ // mqtt_client_wrapper.cpp #include "mqtt_client_wrapper.hpp" -#include + +#include #include #include -#include +#include std::shared_ptr mqtt_client; extern std::atomic g_running; static void send_heartbeat() { - if (!mqtt_client || !mqtt_client->isConnected()) - return; + if (!mqtt_client || !mqtt_client->isConnected()) return; nlohmann::json hb_data; hb_data["time"] = Logger::get_current_time_utc8(); - hb_data["status"] = 0; // 固定上报正常 - hb_data["dispatchId"] = ""; // 已无任务状态 + hb_data["status"] = 0; // 固定上报正常 + hb_data["dispatchId"] = ""; // 已无任务状态 nlohmann::json msg; msg["data"] = hb_data; @@ -33,20 +33,17 @@ static void on_mqtt_connected() send_heartbeat(); - const auto &topics = g_app_config.mqtt.topics; + 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); } // MQTT 连接断开 -static void on_mqtt_disconnected() -{ - LOG_WARN("[MQTT] Disconnected from broker: " + g_app_config.mqtt.server_ip); -} +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) +static void on_mqtt_message_received(const std::string& topic, const std::string& message) { LOG_INFO("[MQTT] Received message on topic [" + topic + "]"); LOG_INFO("[MQTT] Content: " + message); @@ -106,7 +103,7 @@ static void on_mqtt_message_received(const std::string &topic, const std::string LOG_WARN("[MQTT] Unknown topic: " + topic); } - catch (const std::exception &e) + catch (const std::exception& e) { LOG_ERROR(std::string("[MQTT] JSON processing failed: ") + e.what()); } @@ -114,12 +111,12 @@ static void on_mqtt_message_received(const std::string &topic, const std::string void mqtt_client_thread_func() { - const auto &cfg = g_app_config.mqtt; + const auto& cfg = g_app_config.mqtt; auto heartbeat_interval = std::chrono::milliseconds((int)(cfg.keep_alive * 0.9)); while (g_running) { - mqtt_client = std::make_unique(cfg); + mqtt_client = std::make_shared(cfg); mqtt_client->setConnectCallback(on_mqtt_connected); mqtt_client->setDisconnectCallback(on_mqtt_disconnected); mqtt_client->setMessageCallback(on_mqtt_message_received); @@ -154,8 +151,7 @@ void mqtt_client_thread_func() } // 重连间隔 - for (int i = 0; i < 5 && g_running; i++) - std::this_thread::sleep_for(std::chrono::milliseconds(200)); + for (int i = 0; i < 5 && g_running; i++) std::this_thread::sleep_for(std::chrono::milliseconds(200)); } LOG_INFO("[MQTT] Client thread exiting."); diff --git a/src/rtsp_manager.cpp b/src/rtsp_manager.cpp index dbc003a..8f24a00 100644 --- a/src/rtsp_manager.cpp +++ b/src/rtsp_manager.cpp @@ -1,28 +1,30 @@ // rtsp_manager.cpp #include "rtsp_manager.hpp" + +#include +#include +#include + +#include +#include +#include +#include + #include "logger.hpp" -#include -#include -#include -#include -#include -#include -#include - // 静态变量定义 -GMainLoop *RTSPManager::loop = nullptr; -GMainContext *RTSPManager::main_context = nullptr; -GstRTSPServer *RTSPManager::server = nullptr; +GMainLoop* RTSPManager::loop = nullptr; +GMainContext* RTSPManager::main_context = nullptr; +GstRTSPServer* RTSPManager::server = nullptr; std::unordered_map RTSPManager::streaming_status; -std::unordered_map RTSPManager::mounted_factories; +std::unordered_map RTSPManager::mounted_factories; std::mutex RTSPManager::mounted_factories_mutex; -std::unordered_map> RTSPManager::media_map; +std::unordered_map> RTSPManager::media_map; std::mutex RTSPManager::media_map_mutex; -bool set_v4l2_format(const std::string &dev, int width, int height) +bool set_v4l2_format(const std::string& dev, int width, int height) { int fd = open(dev.c_str(), O_RDWR); if (fd < 0) @@ -39,8 +41,7 @@ bool set_v4l2_format(const std::string &dev, int width, int height) if (ioctl(fd, VIDIOC_G_FMT, &fmt) == 0) { bool match = true; - if (fmt.fmt.pix_mp.width != (unsigned int)width || - fmt.fmt.pix_mp.height != (unsigned int)height || + if (fmt.fmt.pix_mp.width != (unsigned int)width || fmt.fmt.pix_mp.height != (unsigned int)height || fmt.fmt.pix_mp.pixelformat != V4L2_PIX_FMT_NV12) { match = false; @@ -49,7 +50,8 @@ bool set_v4l2_format(const std::string &dev, int width, int height) if (match) { close(fd); - LOG_INFO("[RTSP] V4L2 format already NV12 " + std::to_string(width) + "x" + std::to_string(height) + " for " + dev); + LOG_INFO("[RTSP] V4L2 format already NV12 " + std::to_string(width) + "x" + std::to_string(height) + + " for " + dev); return true; } } @@ -79,7 +81,7 @@ void RTSPManager::init() LOG_INFO("[RTSP] GStreamer initialized."); } -GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam) +GstRTSPMediaFactory* RTSPManager::create_media_factory(const Camera& cam) { // 启动前把 v4l2 格式设成我们想要的 set_v4l2_format(cam.device, cam.width, cam.height); @@ -90,9 +92,7 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam) std::string caps = "video/x-raw,format=NV12," "width=" + - std::to_string(w) + - ",height=" + std::to_string(h) + - ",framerate=" + std::to_string(cam.fps) + "/1"; + std::to_string(w) + ",height=" + std::to_string(h) + ",framerate=" + std::to_string(cam.fps) + "/1"; // 关键：tee 一路给 RTSP，一路给 fakesink，保持 v4l2src / mpph264enc 永远在跑 // @@ -100,25 +100,23 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam) // v4l2src -> caps -> tee name=t // tee src0: t. -> queue -> mpph264enc -> h264parse -> rtph264pay (给 RTSP) // tee src1: t. -> queue -> fakesink (假消费者，让 DMA 持续 dequeue) - std::string launch_str = - "( v4l2src device=" + cam.device + - " io-mode=2 is-live=true do-timestamp=true" - " ! " + - caps + - " ! tee name=t " - " ! queue leaky=downstream max-size-time=0 max-size-bytes=0 max-size-buffers=0" - " ! mpph264enc name=enc rc-mode=cbr bps=" + - std::to_string(cam.bitrate) + - " gop=" + std::to_string(cam.fps) + - " header-mode=1" - " ! h264parse" - " ! rtph264pay name=pay0 pt=96 config-interval=1 " - " t. ! queue leaky=downstream max-size-buffers=3 max-size-bytes=0 max-size-time=0" - " ! fakesink sync=false async=false )"; + std::string launch_str = "( v4l2src device=" + cam.device + + " io-mode=2 is-live=true do-timestamp=true" + " ! " + + caps + + " ! tee name=t " + " ! queue leaky=downstream max-size-time=0 max-size-bytes=0 max-size-buffers=0" + " ! mpph264enc name=enc rc-mode=cbr bps=" + + std::to_string(cam.bitrate) + " gop=" + std::to_string(cam.fps) + + " header-mode=1" + " ! h264parse" + " ! rtph264pay name=pay0 pt=96 config-interval=1 " + " t. ! queue leaky=downstream max-size-buffers=3 max-size-bytes=0 max-size-time=0" + " ! fakesink sync=false async=false )"; LOG_INFO("[RTSP] Launch for " + cam.name + ": " + launch_str); - GstRTSPMediaFactory *factory = gst_rtsp_media_factory_new(); + GstRTSPMediaFactory* factory = gst_rtsp_media_factory_new(); gst_rtsp_media_factory_set_launch(factory, launch_str.c_str()); // 先保持每个客户端独立 pipeline，逻辑简单可靠 @@ -127,17 +125,13 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam) // 客户端断开时不要乱 reset，交给我们自己处理 / 或干脆不动 gst_rtsp_media_factory_set_suspend_mode(factory, GST_RTSP_SUSPEND_MODE_NONE); - g_signal_connect_data(factory, - "media-configure", - G_CALLBACK(on_media_created), - g_strdup(cam.name.c_str()), - (GClosureNotify)g_free, - (GConnectFlags)0); + g_signal_connect_data(factory, "media-configure", G_CALLBACK(on_media_created), g_strdup(cam.name.c_str()), + (GClosureNotify)g_free, (GConnectFlags)0); return factory; } -void RTSPManager::start(const std::vector &cams) +void RTSPManager::start(const std::vector& cams) { server = gst_rtsp_server_new(); gst_rtsp_server_set_service(server, "8554"); @@ -146,15 +140,14 @@ void RTSPManager::start(const std::vector &cams) main_context = g_main_loop_get_context(loop); // 先获取 mountpoints - GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); + GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server); // 在这里统一挂载所有 enabled 摄像头 - for (const auto &cam : cams) + for (const auto& cam : cams) { - if (!cam.enabled) - continue; + if (!cam.enabled) continue; - GstRTSPMediaFactory *factory = create_media_factory(cam); + GstRTSPMediaFactory* factory = create_media_factory(cam); std::string mount_point = "/" + cam.name; gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory); @@ -165,8 +158,7 @@ void RTSPManager::start(const std::vector &cams) streaming_status[cam.name] = true; } - LOG_INFO("[RTSP] Camera '" + cam.name + - "' mounted at rtsp://0.0.0.0:8554" + mount_point); + LOG_INFO("[RTSP] Camera '" + cam.name + "' mounted at rtsp://0.0.0.0:8554" + mount_point); } g_object_unref(mounts); @@ -190,69 +182,55 @@ void RTSPManager::start(const std::vector &cams) LOG_INFO("[RTSP] Server stopped."); } -void RTSPManager::on_media_created(GstRTSPMediaFactory *, GstRTSPMedia *media, gpointer user_data) +void RTSPManager::on_media_created(GstRTSPMediaFactory*, GstRTSPMedia* media, gpointer user_data) { - const char *cam_name = static_cast(user_data); - + const char* cam_name = static_cast(user_data); LOG_INFO(std::string("[RTSP] media-configure for camera: ") + cam_name); - // 保存 media，用于之后 unmount 时统一管理 - g_object_ref(media); + // ✅ 只记录指针，不 g_object_ref（生命周期归 gst-rtsp-server） { std::lock_guard lock(media_map_mutex); media_map[cam_name].push_back(media); } - g_signal_connect_data( - media, - "unprepared", - G_CALLBACK(on_media_unprepared), - g_strdup(cam_name), - (GClosureNotify)g_free, - (GConnectFlags)0); - - // 不再手动改 pipeline 状态，交给 gst-rtsp-server 自己管理 - // 如果后面发现确实有卡在 PAUSED 的情况，再在这里加逻辑 + g_signal_connect_data(media, "unprepared", G_CALLBACK(on_media_unprepared), g_strdup(cam_name), + (GClosureNotify)g_free, (GConnectFlags)0); } -void RTSPManager::on_media_unprepared(GstRTSPMedia *media, gpointer user_data) +void RTSPManager::on_media_unprepared(GstRTSPMedia* media, gpointer user_data) { - const char *cam_name = static_cast(user_data); - + const char* cam_name = static_cast(user_data); LOG_INFO(std::string("[RTSP] media-unprepared: ") + cam_name); + std::lock_guard lock(media_map_mutex); + auto it = media_map.find(cam_name); + if (it != media_map.end()) { - std::lock_guard lock(media_map_mutex); - auto it = media_map.find(cam_name); - if (it != media_map.end()) - { - auto &vec = it->second; - vec.erase(std::remove(vec.begin(), vec.end(), media), vec.end()); - if (vec.empty()) - media_map.erase(it); - } - else - { - LOG_WARN(std::string("[RTSP] media-unprepared but no entry in media_map for camera: ") + cam_name); - } + auto& vec = it->second; + vec.erase(std::remove(vec.begin(), vec.end(), media), vec.end()); + if (vec.empty()) media_map.erase(it); + } + else + { + LOG_WARN(std::string("[RTSP] media-unprepared but no entry in media_map for camera: ") + cam_name); } - g_object_unref(media); + // ✅ 不要 g_object_unref(media) —— gst-rtsp-server 会处理 } gboolean RTSPManager::mount_camera_in_main(gpointer data) { - Camera *cam = static_cast(data); + Camera* cam = static_cast(data); if (!cam || !server) { delete cam; return G_SOURCE_REMOVE; } - GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); + GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server); std::string mount_point = "/" + cam->name; - GstRTSPMediaFactory *factory = create_media_factory(*cam); + GstRTSPMediaFactory* factory = create_media_factory(*cam); gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory); g_object_unref(mounts); @@ -271,7 +249,7 @@ gboolean RTSPManager::mount_camera_in_main(gpointer data) gboolean RTSPManager::unmount_camera_in_main(gpointer data) { - Camera *cam = static_cast(data); + Camera* cam = static_cast(data); if (!cam || !server) { delete cam; @@ -281,31 +259,15 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data) std::string cam_name = cam->name; std::string mount_point = "/" + cam_name; - // 停掉 media + // ✅ 不要手动 gst_rtsp_media_unprepare / set_state(NULL) + // 只移除我们自己的记录。media 的 teardown 交给 gst-rtsp-server。 { std::lock_guard lock(media_map_mutex); - auto it = media_map.find(cam_name); - if (it != media_map.end()) - { - for (GstRTSPMedia *media : it->second) - { - GstElement *pipeline = gst_rtsp_media_get_element(media); - if (pipeline) - { - gst_element_set_state(pipeline, GST_STATE_NULL); - gst_object_unref(pipeline); - } - - gst_rtsp_media_unprepare(media); - } - - it->second.clear(); - media_map.erase(it); - } + media_map.erase(cam_name); } // 移除 factory - GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); + GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server); if (mounts) { gst_rtsp_mount_points_remove_factory(mounts, mount_point.c_str()); @@ -317,8 +279,11 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data) auto it = mounted_factories.find(cam_name); if (it != mounted_factories.end()) { - if (it->second) - g_object_unref(it->second); + // ⚠️ 这里是否 unref factory 取决于 mount_points 是否持有引用。 + // 一般 remove_factory 后会释放其引用；你再 unref 可能导致过度 unref。 + // 为了稳，建议：先别手动 unref，避免 double-unref。 + // if (it->second) g_object_unref(it->second); + mounted_factories.erase(it); } streaming_status[cam_name] = false; @@ -329,21 +294,21 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data) return G_SOURCE_REMOVE; } -void RTSPManager::mount_camera(const Camera &cam) +void RTSPManager::mount_camera(const Camera& cam) { - Camera *camCopy = new Camera(cam); - g_main_context_invoke(main_context, [](gpointer data) -> gboolean - { return RTSPManager::mount_camera_in_main(data); }, camCopy); + Camera* camCopy = new Camera(cam); + g_main_context_invoke( + main_context, [](gpointer data) -> gboolean { return RTSPManager::mount_camera_in_main(data); }, camCopy); } -void RTSPManager::unmount_camera(const Camera &cam) +void RTSPManager::unmount_camera(const Camera& cam) { - Camera *camCopy = new Camera(cam); - g_main_context_invoke(main_context, [](gpointer data) -> gboolean - { return RTSPManager::unmount_camera_in_main(data); }, camCopy); + Camera* camCopy = new Camera(cam); + g_main_context_invoke( + main_context, [](gpointer data) -> gboolean { return RTSPManager::unmount_camera_in_main(data); }, camCopy); } -bool RTSPManager::is_streaming(const std::string &cam_name) +bool RTSPManager::is_streaming(const std::string& cam_name) { std::lock_guard lock(mounted_factories_mutex); auto it = streaming_status.find(cam_name); @@ -353,9 +318,8 @@ bool RTSPManager::is_streaming(const std::string &cam_name) bool RTSPManager::is_any_streaming() { std::lock_guard lock(mounted_factories_mutex); - for (auto &kv : streaming_status) - if (kv.second) - return true; + for (auto& kv : streaming_status) + if (kv.second) return true; return false; } @@ -367,7 +331,7 @@ void RTSPManager::stop() main_context, [](gpointer data) -> gboolean { - g_main_loop_quit(static_cast(data)); + g_main_loop_quit(static_cast(data)); return G_SOURCE_REMOVE; }, loop);