This commit is contained in:
cxh 2025-11-21 15:32:56 +08:00
parent cb0d03e2a1
commit 0fd1a90afb
2 changed files with 49 additions and 91 deletions

View File

@ -7,7 +7,6 @@
#include <unordered_map> #include <unordered_map>
#include <string> #include <string>
#include <mutex> #include <mutex>
#include <condition_variable>
#include <vector> #include <vector>
class RTSPManager class RTSPManager
@ -19,6 +18,7 @@ public:
static void mount_camera(const Camera &cam); static void mount_camera(const Camera &cam);
static void unmount_camera(const Camera &cam); static void unmount_camera(const Camera &cam);
static bool is_streaming(const std::string &cam_name); static bool is_streaming(const std::string &cam_name);
static bool is_any_streaming(); static bool is_any_streaming();
@ -26,29 +26,25 @@ private:
static GMainLoop *loop; static GMainLoop *loop;
static GMainContext *main_context; static GMainContext *main_context;
static GstRTSPServer *server; static GstRTSPServer *server;
static std::unordered_map<std::string, bool> streaming_status; static std::unordered_map<std::string, bool> streaming_status;
// 工厂创建函数 // 工厂创建
static GstRTSPMediaFactory *create_media_factory(const Camera &cam); static GstRTSPMediaFactory *create_media_factory(const Camera &cam);
// 挂载/卸载函数 // 挂载/卸载
static gboolean mount_camera_in_main(gpointer data); static gboolean mount_camera_in_main(gpointer data);
static gboolean unmount_camera_in_main(gpointer data); static gboolean unmount_camera_in_main(gpointer data);
// 静态 mutex 和工厂表 // factory 管理
static std::unordered_map<std::string, GstRTSPMediaFactory *> mounted_factories; static std::unordered_map<std::string, GstRTSPMediaFactory *> mounted_factories;
static std::mutex mounted_factories_mutex; static std::mutex mounted_factories_mutex;
// 媒体对象跟踪 // pipeline/media 管理
static std::unordered_map<std::string, std::vector<GstRTSPMedia *>> media_map; static std::unordered_map<std::string, std::vector<GstRTSPMedia *>> media_map;
static std::mutex media_map_mutex; static std::mutex media_map_mutex;
// 信号处理函数 // gstreamer 信号
static void on_media_created(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data); static void on_media_created(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data);
static void on_media_unprepared(GstRTSPMedia *media, gpointer user_data); static void on_media_unprepared(GstRTSPMedia *media, gpointer user_data);
// === 新增:同步停止用 ===
static std::mutex stop_mutex;
static std::condition_variable stop_cv;
static std::unordered_map<std::string, int> stopping_count;
}; };

View File

@ -1,6 +1,7 @@
// rtsp_manager.cpp // rtsp_manager.cpp
#include "rtsp_manager.hpp" #include "rtsp_manager.hpp"
#include "logger.hpp" #include "logger.hpp"
#include <iostream> #include <iostream>
#include <algorithm> #include <algorithm>
#include <thread> #include <thread>
@ -10,14 +11,13 @@
GMainLoop *RTSPManager::loop = nullptr; GMainLoop *RTSPManager::loop = nullptr;
GMainContext *RTSPManager::main_context = nullptr; GMainContext *RTSPManager::main_context = nullptr;
GstRTSPServer *RTSPManager::server = nullptr; GstRTSPServer *RTSPManager::server = nullptr;
std::unordered_map<std::string, bool> RTSPManager::streaming_status; std::unordered_map<std::string, bool> RTSPManager::streaming_status;
std::unordered_map<std::string, GstRTSPMediaFactory *> RTSPManager::mounted_factories; std::unordered_map<std::string, GstRTSPMediaFactory *> RTSPManager::mounted_factories;
std::mutex RTSPManager::mounted_factories_mutex; std::mutex RTSPManager::mounted_factories_mutex;
std::unordered_map<std::string, std::vector<GstRTSPMedia *>> RTSPManager::media_map; std::unordered_map<std::string, std::vector<GstRTSPMedia *>> RTSPManager::media_map;
std::mutex RTSPManager::media_map_mutex; std::mutex RTSPManager::media_map_mutex;
std::mutex RTSPManager::stop_mutex;
std::condition_variable RTSPManager::stop_cv;
std::unordered_map<std::string, int> RTSPManager::stopping_count;
void RTSPManager::init() void RTSPManager::init()
{ {
@ -25,10 +25,8 @@ void RTSPManager::init()
LOG_INFO("[RTSP] GStreamer initialized."); LOG_INFO("[RTSP] GStreamer initialized.");
} }
// 创建 media factory
GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam) GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam)
{ {
// 输出分辨率直接用相机原始分辨率
int out_width = cam.width; int out_width = cam.width;
int out_height = cam.height; int out_height = cam.height;
@ -48,18 +46,18 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam)
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()); gst_rtsp_media_factory_set_launch(factory, launch_str.c_str());
gst_rtsp_media_factory_set_shared(factory, TRUE); gst_rtsp_media_factory_set_shared(factory, TRUE);
gst_rtsp_media_factory_set_suspend_mode(factory, GST_RTSP_SUSPEND_MODE_RESET); gst_rtsp_media_factory_set_suspend_mode(factory, GST_RTSP_SUSPEND_MODE_RESET);
// 使用 media-configure 信号代替 media-created // 监听 pipeline 创建
g_signal_connect_data(factory, "media-configure", G_CALLBACK(on_media_created), g_signal_connect_data(factory, "media-configure", G_CALLBACK(on_media_created),
g_strdup(cam.name.c_str()), (GClosureNotify)g_free, static_cast<GConnectFlags>(0)); g_strdup(cam.name.c_str()),
(GClosureNotify)g_free,
(GConnectFlags)0);
return factory; return factory;
} }
// 启动 RTSP server void RTSPManager::start(const std::vector<Camera> &)
void RTSPManager::start(const std::vector<Camera> &cameras)
{ {
server = gst_rtsp_server_new(); server = gst_rtsp_server_new();
gst_rtsp_server_set_service(server, "8554"); gst_rtsp_server_set_service(server, "8554");
@ -86,51 +84,46 @@ void RTSPManager::start(const std::vector<Camera> &cameras)
LOG_INFO("[RTSP] Server stopped."); LOG_INFO("[RTSP] Server stopped.");
} }
// media-configure 信号处理 void RTSPManager::on_media_created(GstRTSPMediaFactory *, GstRTSPMedia *media, gpointer user_data)
void RTSPManager::on_media_created(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data)
{ {
const char *cam_name = static_cast<const char *>(user_data); const char *cam_name = static_cast<const char *>(user_data);
g_object_ref(media); // 增加引用计数,防止被提前销毁 LOG_INFO(std::string("[RTSP] media-configure for camera: ") + cam_name);
g_object_ref(media);
{ {
std::lock_guard<std::mutex> lock(media_map_mutex); std::lock_guard<std::mutex> lock(media_map_mutex);
media_map[cam_name].push_back(media); media_map[cam_name].push_back(media);
} }
// 连接 unprepared 信号,当 pipeline 被销毁时移除
g_signal_connect_data(media, "unprepared", G_CALLBACK(on_media_unprepared), g_signal_connect_data(media, "unprepared", G_CALLBACK(on_media_unprepared),
g_strdup(cam_name), (GClosureNotify)g_free, static_cast<GConnectFlags>(0)); g_strdup(cam_name),
(GClosureNotify)g_free,
(GConnectFlags)0);
} }
// unprepared 信号处理
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<const char *>(user_data); const char *cam_name = static_cast<const char *>(user_data);
LOG_INFO(std::string("[RTSP] media-unprepared: ") + cam_name);
{ {
std::lock_guard<std::mutex> lock(media_map_mutex); std::lock_guard<std::mutex> lock(media_map_mutex);
auto it = media_map.find(cam_name); auto it = media_map.find(cam_name);
if (it != media_map.end()) if (it != media_map.end())
{ {
auto &media_list = it->second; auto &vec = it->second;
media_list.erase(std::remove(media_list.begin(), media_list.end(), media), media_list.end()); vec.erase(std::remove(vec.begin(), vec.end(), media), vec.end());
if (media_list.empty()) if (vec.empty())
{
media_map.erase(it); media_map.erase(it);
} }
} }
g_object_unref(media);
} }
// === ★ 新增:减计数并在最后一个时唤醒等待线程 ===
{
std::lock_guard<std::mutex> lk(stop_mutex);
if (--stopping_count[cam_name] == 0)
stop_cv.notify_all();
}
g_object_unref(media); // 释放引用
}
// 挂载摄像头
gboolean RTSPManager::mount_camera_in_main(gpointer data) gboolean RTSPManager::mount_camera_in_main(gpointer data)
{ {
Camera *cam = static_cast<Camera *>(data); Camera *cam = static_cast<Camera *>(data);
@ -141,20 +134,9 @@ gboolean RTSPManager::mount_camera_in_main(gpointer data)
} }
GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server);
if (!mounts)
{
delete cam;
return G_SOURCE_REMOVE;
}
std::string mount_point = "/" + cam->name; std::string mount_point = "/" + cam->name;
GstRTSPMediaFactory *factory = create_media_factory(*cam); 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); gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory);
g_object_unref(mounts); g_object_unref(mounts);
@ -166,11 +148,11 @@ gboolean RTSPManager::mount_camera_in_main(gpointer data)
} }
LOG_INFO("[RTSP] Camera '" + cam->name + "' mounted at rtsp://localhost:8554" + mount_point); LOG_INFO("[RTSP] Camera '" + cam->name + "' mounted at rtsp://localhost:8554" + mount_point);
delete cam; delete cam;
return G_SOURCE_REMOVE; return G_SOURCE_REMOVE;
} }
// 卸载摄像头
gboolean RTSPManager::unmount_camera_in_main(gpointer data) gboolean RTSPManager::unmount_camera_in_main(gpointer data)
{ {
Camera *cam = static_cast<Camera *>(data); Camera *cam = static_cast<Camera *>(data);
@ -179,22 +161,16 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data)
delete cam; delete cam;
return G_SOURCE_REMOVE; return G_SOURCE_REMOVE;
} }
std::string cam_name = cam->name; std::string cam_name = cam->name;
std::string mount_point = "/" + cam_name; std::string mount_point = "/" + cam_name;
// 停止所有媒体 // 停掉 media
{ {
std::lock_guard<std::mutex> lock(media_map_mutex); std::lock_guard<std::mutex> lock(media_map_mutex);
auto it = media_map.find(cam_name); auto it = media_map.find(cam_name);
if (it != media_map.end()) if (it != media_map.end())
{ {
// === ★ 新增:记录需要等待的 media 数量 ===
{
std::lock_guard<std::mutex> lk(stop_mutex);
stopping_count[cam_name] = it->second.size();
}
for (GstRTSPMedia *media : it->second) for (GstRTSPMedia *media : it->second)
{ {
GstElement *pipeline = gst_rtsp_media_get_element(media); GstElement *pipeline = gst_rtsp_media_get_element(media);
@ -204,13 +180,15 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data)
gst_object_unref(pipeline); gst_object_unref(pipeline);
} }
// 这一步会触发 on_media_unprepared 信号
gst_rtsp_media_unprepare(media); gst_rtsp_media_unprepare(media);
} }
it->second.clear();
media_map.erase(it);
} }
} }
// 卸载 factory // 移除 factory
GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server);
if (mounts) if (mounts)
{ {
@ -223,7 +201,7 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data)
auto it = mounted_factories.find(cam_name); auto it = mounted_factories.find(cam_name);
if (it != mounted_factories.end()) if (it != mounted_factories.end())
{ {
if (it->second && G_IS_OBJECT(it->second)) if (it->second)
g_object_unref(it->second); g_object_unref(it->second);
mounted_factories.erase(it); mounted_factories.erase(it);
} }
@ -235,11 +213,9 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data)
return G_SOURCE_REMOVE; return G_SOURCE_REMOVE;
} }
// 公共挂载/卸载接口
void RTSPManager::mount_camera(const Camera &cam) void RTSPManager::mount_camera(const Camera &cam)
{ {
Camera *camCopy = new Camera(cam); Camera *camCopy = new Camera(cam);
g_main_context_invoke(main_context, [](gpointer data) -> gboolean g_main_context_invoke(main_context, [](gpointer data) -> gboolean
{ return RTSPManager::mount_camera_in_main(data); }, camCopy); { return RTSPManager::mount_camera_in_main(data); }, camCopy);
} }
@ -247,18 +223,10 @@ void RTSPManager::mount_camera(const Camera &cam)
void RTSPManager::unmount_camera(const Camera &cam) void RTSPManager::unmount_camera(const Camera &cam)
{ {
Camera *camCopy = new Camera(cam); Camera *camCopy = new Camera(cam);
g_main_context_invoke(main_context, [](gpointer data) -> gboolean g_main_context_invoke(main_context, [](gpointer data) -> gboolean
{ return RTSPManager::unmount_camera_in_main(data); }, camCopy); { return RTSPManager::unmount_camera_in_main(data); }, camCopy);
std::unique_lock<std::mutex> lk(stop_mutex);
stop_cv.wait(lk, [&]
{
auto it = stopping_count.find(cam.name);
return it == stopping_count.end() || it->second == 0; });
} }
// 是否正在流
bool RTSPManager::is_streaming(const std::string &cam_name) bool RTSPManager::is_streaming(const std::string &cam_name)
{ {
std::lock_guard<std::mutex> lock(mounted_factories_mutex); std::lock_guard<std::mutex> lock(mounted_factories_mutex);
@ -266,32 +234,26 @@ bool RTSPManager::is_streaming(const std::string &cam_name)
return it != streaming_status.end() ? it->second : false; return it != streaming_status.end() ? it->second : false;
} }
// 停止 server bool RTSPManager::is_any_streaming()
{
std::lock_guard<std::mutex> lock(mounted_factories_mutex);
for (auto &kv : streaming_status)
if (kv.second)
return true;
return false;
}
void RTSPManager::stop() void RTSPManager::stop()
{ {
// Ctrl-C 退出时,不再尝试逐路卸载摄像头,
// 只负责让 g_main_loop_run() 退出,线程自然收尾。
if (loop) if (loop)
{ {
g_main_context_invoke( g_main_context_invoke(
main_context, main_context,
[](gpointer data) -> gboolean [](gpointer data) -> gboolean
{ {
GMainLoop *lp = static_cast<GMainLoop *>(data); g_main_loop_quit(static_cast<GMainLoop *>(data));
if (lp)
g_main_loop_quit(lp);
return G_SOURCE_REMOVE; return G_SOURCE_REMOVE;
}, },
loop); loop);
} }
} }
// 新增接口:检查是否还有摄像头在流
bool RTSPManager::is_any_streaming()
{
std::lock_guard<std::mutex> lock(mounted_factories_mutex);
for (const auto &kv : streaming_status)
if (kv.second)
return true;
return false;
}