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-12-17 14:13:52 +08:00
parent 259efde6ca
commit 9017939660
2 changed files with 99 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/mqtt_client_wrapper.cpp
View File

@ -1,22 +1,22 @@
// mqtt_client_wrapper.cpp // mqtt_client_wrapper.cpp
#include "mqtt_client_wrapper.hpp" #include "mqtt_client_wrapper.hpp"
#include <thread>
#include <algorithm>
#include <chrono> #include <chrono>
#include <nlohmann/json.hpp> #include <nlohmann/json.hpp>
#include <algorithm> #include <thread>
std::shared_ptr<MQTTClient> mqtt_client; std::shared_ptr<MQTTClient> mqtt_client;
extern std::atomic<bool> g_running; extern std::atomic<bool> g_running;
static void send_heartbeat() static void send_heartbeat()
{ {
if (!mqtt_client || !mqtt_client->isConnected()) if (!mqtt_client || !mqtt_client->isConnected()) return;
return;
nlohmann::json hb_data; nlohmann::json hb_data;
hb_data["time"] = Logger::get_current_time_utc8(); hb_data["time"] = Logger::get_current_time_utc8();
hb_data["status"] = 0; // 固定上报正常 hb_data["status"] = 0; // 固定上报正常
hb_data["dispatchId"] = ""; // 已无任务状态 hb_data["dispatchId"] = ""; // 已无任务状态
nlohmann::json msg; nlohmann::json msg;
msg["data"] = hb_data; msg["data"] = hb_data;
@ -33,20 +33,17 @@ static void on_mqtt_connected()
send_heartbeat(); 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.video_down);
mqtt_client->subscribe(topics.substream_down); mqtt_client->subscribe(topics.substream_down);
mqtt_client->subscribe(topics.reset_down); mqtt_client->subscribe(topics.reset_down);
} }
// MQTT 连接断开 // MQTT 连接断开
static void on_mqtt_disconnected() static void on_mqtt_disconnected() { LOG_WARN("[MQTT] Disconnected from broker: " + g_app_config.mqtt.server_ip); }
{
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] Received message on topic [" + topic + "]");
LOG_INFO("[MQTT] Content: " + message); 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); 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()); 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() 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)); auto heartbeat_interval = std::chrono::milliseconds((int)(cfg.keep_alive * 0.9));
while (g_running) while (g_running)
{ {
mqtt_client = std::make_unique<MQTTClient>(cfg); mqtt_client = std::make_shared<MQTTClient>(cfg);
mqtt_client->setConnectCallback(on_mqtt_connected); mqtt_client->setConnectCallback(on_mqtt_connected);
mqtt_client->setDisconnectCallback(on_mqtt_disconnected); mqtt_client->setDisconnectCallback(on_mqtt_disconnected);
mqtt_client->setMessageCallback(on_mqtt_message_received); 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++) for (int i = 0; i < 5 && g_running; i++) std::this_thread::sleep_for(std::chrono::milliseconds(200));
std::this_thread::sleep_for(std::chrono::milliseconds(200));
} }
LOG_INFO("[MQTT] Client thread exiting."); LOG_INFO("[MQTT] Client thread exiting.");

208
src/rtsp_manager.cpp
View File

@ -1,28 +1,30 @@
// rtsp_manager.cpp // rtsp_manager.cpp
#include "rtsp_manager.hpp" #include "rtsp_manager.hpp"
#include <fcntl.h>
#include <linux/videodev2.h>
#include <sys/ioctl.h>
#include <algorithm>
#include <chrono>
#include <iostream>
#include <thread>
#include "logger.hpp" #include "logger.hpp"
#include <iostream>
#include <algorithm>
#include <thread>
#include <chrono>
#include <fcntl.h>
#include <sys/ioctl.h>
#include <linux/videodev2.h>
// 静态变量定义 // 静态变量定义
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;
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); int fd = open(dev.c_str(), O_RDWR);
if (fd < 0) 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) if (ioctl(fd, VIDIOC_G_FMT, &fmt) == 0)
{ {
bool match = true; bool match = true;
if (fmt.fmt.pix_mp.width != (unsigned int)width || if (fmt.fmt.pix_mp.width != (unsigned int)width || fmt.fmt.pix_mp.height != (unsigned int)height ||
fmt.fmt.pix_mp.height != (unsigned int)height ||
fmt.fmt.pix_mp.pixelformat != V4L2_PIX_FMT_NV12) fmt.fmt.pix_mp.pixelformat != V4L2_PIX_FMT_NV12)
{ {
match = false; match = false;
@ -49,7 +50,8 @@ bool set_v4l2_format(const std::string &dev, int width, int height)
if (match) if (match)
{ {
close(fd); 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; return true;
} }
} }
@ -79,7 +81,7 @@ void RTSPManager::init()
LOG_INFO("[RTSP] GStreamer initialized."); LOG_INFO("[RTSP] GStreamer initialized.");
} }
GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam) GstRTSPMediaFactory* RTSPManager::create_media_factory(const Camera& cam)
{ {
// 启动前把 v4l2 格式设成我们想要的 // 启动前把 v4l2 格式设成我们想要的
set_v4l2_format(cam.device, cam.width, cam.height); set_v4l2_format(cam.device, cam.width, cam.height);
@ -90,9 +92,7 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam)
std::string caps = std::string caps =
"video/x-raw,format=NV12," "video/x-raw,format=NV12,"
"width=" + "width=" +
std::to_string(w) + std::to_string(w) + ",height=" + std::to_string(h) + ",framerate=" + std::to_string(cam.fps) + "/1";
",height=" + std::to_string(h) +
",framerate=" + std::to_string(cam.fps) + "/1";
// 关键：tee 一路给 RTSP，一路给 fakesink，保持 v4l2src / mpph264enc 永远在跑 // 关键：tee 一路给 RTSP，一路给 fakesink，保持 v4l2src / mpph264enc 永远在跑
// //
@ -100,25 +100,23 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam)
// v4l2src -> caps -> tee name=t // v4l2src -> caps -> tee name=t
// tee src0: t. -> queue -> mpph264enc -> h264parse -> rtph264pay (给 RTSP) // tee src0: t. -> queue -> mpph264enc -> h264parse -> rtph264pay (给 RTSP)
// tee src1: t. -> queue -> fakesink (假消费者，让 DMA 持续 dequeue) // tee src1: t. -> queue -> fakesink (假消费者，让 DMA 持续 dequeue)
std::string launch_str = std::string launch_str = "( v4l2src device=" + cam.device +
"( v4l2src device=" + cam.device + " io-mode=2 is-live=true do-timestamp=true"
" io-mode=2 is-live=true do-timestamp=true" " ! " +
" ! " + caps +
caps + " ! tee name=t "
" ! tee name=t " " ! queue leaky=downstream max-size-time=0 max-size-bytes=0 max-size-buffers=0"
" ! queue leaky=downstream max-size-time=0 max-size-bytes=0 max-size-buffers=0" " ! mpph264enc name=enc rc-mode=cbr bps=" +
" ! mpph264enc name=enc rc-mode=cbr bps=" + std::to_string(cam.bitrate) + " gop=" + std::to_string(cam.fps) +
std::to_string(cam.bitrate) + " header-mode=1"
" gop=" + std::to_string(cam.fps) + " ! h264parse"
" header-mode=1" " ! rtph264pay name=pay0 pt=96 config-interval=1 "
" ! h264parse" " t. ! queue leaky=downstream max-size-buffers=3 max-size-bytes=0 max-size-time=0"
" ! rtph264pay name=pay0 pt=96 config-interval=1 " " ! fakesink sync=false async=false )";
" 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); 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()); gst_rtsp_media_factory_set_launch(factory, launch_str.c_str());
// 先保持每个客户端独立 pipeline，逻辑简单可靠 // 先保持每个客户端独立 pipeline，逻辑简单可靠
@ -127,17 +125,13 @@ GstRTSPMediaFactory *RTSPManager::create_media_factory(const Camera &cam)
// 客户端断开时不要乱 reset，交给我们自己处理 / 或干脆不动 // 客户端断开时不要乱 reset，交给我们自己处理 / 或干脆不动
gst_rtsp_media_factory_set_suspend_mode(factory, GST_RTSP_SUSPEND_MODE_NONE); gst_rtsp_media_factory_set_suspend_mode(factory, GST_RTSP_SUSPEND_MODE_NONE);
g_signal_connect_data(factory, g_signal_connect_data(factory, "media-configure", G_CALLBACK(on_media_created), g_strdup(cam.name.c_str()),
"media-configure", (GClosureNotify)g_free, (GConnectFlags)0);
G_CALLBACK(on_media_created),
g_strdup(cam.name.c_str()),
(GClosureNotify)g_free,
(GConnectFlags)0);
return factory; return factory;
} }
void RTSPManager::start(const std::vector<Camera> &cams) void RTSPManager::start(const std::vector<Camera>& cams)
{ {
server = gst_rtsp_server_new(); server = gst_rtsp_server_new();
gst_rtsp_server_set_service(server, "8554"); gst_rtsp_server_set_service(server, "8554");
@ -146,15 +140,14 @@ void RTSPManager::start(const std::vector<Camera> &cams)
main_context = g_main_loop_get_context(loop); main_context = g_main_loop_get_context(loop);
// 先获取 mountpoints // 先获取 mountpoints
GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server);
// 在这里统一挂载所有 enabled 摄像头 // 在这里统一挂载所有 enabled 摄像头
for (const auto &cam : cams) for (const auto& cam : cams)
{ {
if (!cam.enabled) if (!cam.enabled) continue;
continue;
GstRTSPMediaFactory *factory = create_media_factory(cam); GstRTSPMediaFactory* factory = create_media_factory(cam);
std::string mount_point = "/" + cam.name; std::string mount_point = "/" + cam.name;
gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory); gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory);
@ -165,8 +158,7 @@ void RTSPManager::start(const std::vector<Camera> &cams)
streaming_status[cam.name] = true; streaming_status[cam.name] = true;
} }
LOG_INFO("[RTSP] Camera '" + cam.name + LOG_INFO("[RTSP] Camera '" + cam.name + "' mounted at rtsp://0.0.0.0:8554" + mount_point);
"' mounted at rtsp://0.0.0.0:8554" + mount_point);
} }
g_object_unref(mounts); g_object_unref(mounts);
@ -190,69 +182,55 @@ void RTSPManager::start(const std::vector<Camera> &cams)
LOG_INFO("[RTSP] Server stopped."); 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<const char *>(user_data); const char* cam_name = static_cast<const char*>(user_data);
LOG_INFO(std::string("[RTSP] media-configure for camera: ") + cam_name); LOG_INFO(std::string("[RTSP] media-configure for camera: ") + cam_name);
// 保存 media，用于之后 unmount 时统一管理 // ✅ 只记录指针，不 g_object_ref（生命周期归 gst-rtsp-server）
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);
} }
g_signal_connect_data( g_signal_connect_data(media, "unprepared", G_CALLBACK(on_media_unprepared), g_strdup(cam_name),
media, (GClosureNotify)g_free, (GConnectFlags)0);
"unprepared",
G_CALLBACK(on_media_unprepared),
g_strdup(cam_name),
(GClosureNotify)g_free,
(GConnectFlags)0);
// 不再手动改 pipeline 状态，交给 gst-rtsp-server 自己管理
// 如果后面发现确实有卡在 PAUSED 的情况，再在这里加逻辑
} }
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); LOG_INFO(std::string("[RTSP] media-unprepared: ") + cam_name);
std::lock_guard<std::mutex> lock(media_map_mutex);
auto it = media_map.find(cam_name);
if (it != media_map.end())
{ {
std::lock_guard<std::mutex> lock(media_map_mutex); auto& vec = it->second;
auto it = media_map.find(cam_name); vec.erase(std::remove(vec.begin(), vec.end(), media), vec.end());
if (it != media_map.end()) if (vec.empty()) media_map.erase(it);
{ }
auto &vec = it->second; else
vec.erase(std::remove(vec.begin(), vec.end(), media), vec.end()); {
if (vec.empty()) LOG_WARN(std::string("[RTSP] media-unprepared but no entry in media_map for camera: ") + cam_name);
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) gboolean RTSPManager::mount_camera_in_main(gpointer data)
{ {
Camera *cam = static_cast<Camera *>(data); Camera* cam = static_cast<Camera*>(data);
if (!cam || !server) if (!cam || !server)
{ {
delete cam; delete cam;
return G_SOURCE_REMOVE; 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; 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); gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory);
g_object_unref(mounts); g_object_unref(mounts);
@ -271,7 +249,7 @@ gboolean RTSPManager::mount_camera_in_main(gpointer data)
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);
if (!cam || !server) if (!cam || !server)
{ {
delete cam; delete cam;
@ -281,31 +259,15 @@ gboolean RTSPManager::unmount_camera_in_main(gpointer data)
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 // ✅ 不要手动 gst_rtsp_media_unprepare / set_state(NULL)
// 只移除我们自己的记录。media 的 teardown 交给 gst-rtsp-server。
{ {
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); media_map.erase(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);
}
} }
// 移除 factory // 移除 factory
GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server); GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server);
if (mounts) if (mounts)
{ {
gst_rtsp_mount_points_remove_factory(mounts, mount_point.c_str()); 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); auto it = mounted_factories.find(cam_name);
if (it != mounted_factories.end()) if (it != mounted_factories.end())
{ {
if (it->second) // ⚠️ 这里是否 unref factory 取决于 mount_points 是否持有引用。
g_object_unref(it->second); // 一般 remove_factory 后会释放其引用；你再 unref 可能导致过度 unref。
// 为了稳，建议：先别手动 unref，避免 double-unref。
// if (it->second) g_object_unref(it->second);
mounted_factories.erase(it); mounted_factories.erase(it);
} }
streaming_status[cam_name] = false; streaming_status[cam_name] = false;
@ -329,21 +294,21 @@ 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(
{ return RTSPManager::mount_camera_in_main(data); }, camCopy); 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); Camera* camCopy = new Camera(cam);
g_main_context_invoke(main_context, [](gpointer data) -> gboolean g_main_context_invoke(
{ return RTSPManager::unmount_camera_in_main(data); }, camCopy); 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<std::mutex> lock(mounted_factories_mutex); std::lock_guard<std::mutex> lock(mounted_factories_mutex);
auto it = streaming_status.find(cam_name); 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() bool RTSPManager::is_any_streaming()
{ {
std::lock_guard<std::mutex> lock(mounted_factories_mutex); std::lock_guard<std::mutex> lock(mounted_factories_mutex);
for (auto &kv : streaming_status) for (auto& kv : streaming_status)
if (kv.second) if (kv.second) return true;
return true;
return false; return false;
} }
@ -367,7 +331,7 @@ void RTSPManager::stop()
main_context, main_context,
[](gpointer data) -> gboolean [](gpointer data) -> gboolean
{ {
g_main_loop_quit(static_cast<GMainLoop *>(data)); g_main_loop_quit(static_cast<GMainLoop*>(data));
return G_SOURCE_REMOVE; return G_SOURCE_REMOVE;
}, },
loop); loop);