#include "rtsp_manager.hpp" #include "logger.hpp" #include // 静态变量定义 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::mutex RTSPManager::mounted_factories_mutex; void RTSPManager::init() { gst_init(nullptr, nullptr); LOG_INFO("[RTSP] GStreamer initialized."); } GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam) { std::string launch_str = "( v4l2src device=" + cam.device + " ! video/x-raw,format=NV12,width=" + std::to_string(cam.width) + ",height=" + std::to_string(cam.height) + ",framerate=" + std::to_string(cam.fps) + "/1" " ! videoconvert ! queue ! mpph264enc ! rtph264pay name=pay0 pt=96 )"; GstRTSPMediaFactory *factory = gst_rtsp_media_factory_new(); gst_rtsp_media_factory_set_launch(factory, launch_str.c_str()); gst_rtsp_media_factory_set_shared(factory, TRUE); // 为 factory 添加 bus 监听 EOS GstElement *element = gst_rtsp_media_factory_get_element(factory); if (element) { GstBus *bus = gst_element_get_bus(element); gst_bus_add_watch(bus, [](GstBus *bus, GstMessage *msg, gpointer data) -> gboolean { if(GST_MESSAGE_TYPE(msg) == GST_MESSAGE_EOS) { std::string cam_name = static_cast(data); std::lock_guard lock(mounted_factories_mutex); auto it = mounted_factories.find(cam_name); if(it != mounted_factories.end() && it->second.pending_unmount) { if(it->second.factory) g_object_unref(it->second.factory); mounted_factories.erase(it); streaming_status.erase(cam_name); LOG_INFO("[RTSP] Camera '" + cam_name + "' factory cleaned up after EOS."); } delete[] static_cast(data); } return TRUE; }, new char[cam.name.size() + 1]{0}); gst_object_unref(bus); } return factory; } void RTSPManager::start(const std::vector &cameras) { server = gst_rtsp_server_new(); gst_rtsp_server_set_service(server, "8554"); loop = g_main_loop_new(nullptr, FALSE); main_context = g_main_loop_get_context(loop); gst_rtsp_server_attach(server, nullptr); LOG_INFO("[RTSP] Server running on rtsp://localhost:8554"); g_main_loop_run(loop); if (server) { g_object_unref(server); server = nullptr; } if (loop) { g_main_loop_unref(loop); loop = nullptr; } LOG_INFO("[RTSP] Server stopped."); } // --- 静态成员实现 --- gboolean RTSPManager::mount_camera_in_main(gpointer data) { Camera *cam = static_cast(data); if (!cam || !server) { delete cam; return G_SOURCE_REMOVE; } GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); if (!mounts) { delete cam; return G_SOURCE_REMOVE; } std::string mount_point = "/" + cam->name; GstRTSPMediaFactory *factory = create_media_factory(*cam); if (!factory) { g_object_unref(mounts); delete cam; return G_SOURCE_REMOVE; } gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory); g_object_unref(mounts); { std::lock_guard lock(mounted_factories_mutex); mounted_factories[cam->name] = factory; streaming_status[cam->name] = true; } LOG_INFO("[RTSP] Camera '" + cam->name + "' mounted at rtsp://localhost:8554" + mount_point); delete cam; return G_SOURCE_REMOVE; } gboolean RTSPManager::unmount_camera_in_main(gpointer data) { Camera *cam = static_cast(data); if (!cam || !server) { delete cam; return G_SOURCE_REMOVE; } GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); if (!mounts) { delete cam; return G_SOURCE_REMOVE; } std::string mount_point = "/" + cam->name; { std::lock_guard lock(mounted_factories_mutex); auto it = mounted_factories.find(cam->name); if (it != mounted_factories.end()) { // 发送 EOS,让 pipeline 停止 GstRTSPMediaFactory *factory = it->second.factory; if (factory) { GstElement *elem = gst_rtsp_media_factory_get_element(factory); if (elem) gst_element_send_event(elem, gst_event_new_eos()); } // 标记 pending_unmount it->second.pending_unmount = true; } } // 从 mount points 立即移除,让客户端断流,但 factory 还保留 gst_rtsp_mount_points_remove_factory(mounts, mount_point.c_str()); g_object_unref(mounts); LOG_INFO("[RTSP] Camera '" + cam->name + "' unmount requested (EOS sent)."); delete cam; return G_SOURCE_REMOVE; } 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); } 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); } bool RTSPManager::is_streaming(const std::string &cam_name) { std::lock_guard lock(mounted_factories_mutex); auto it = streaming_status.find(cam_name); return it != streaming_status.end() ? it->second : false; } void RTSPManager::stop() { if (loop) { g_main_context_invoke(main_context, [](gpointer data) -> gboolean { g_main_loop_quit(static_cast(data)); return G_SOURCE_REMOVE; }, loop); } }