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

优化管线

Browse Source
This commit is contained in:
cxh 2025-11-12 17:47:56 +08:00
parent 0281aeafdd
commit 9dde95485e
1 changed files with 35 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
src/rtmp_manager.cpp
View File

@ -1,4 +1,4 @@
// rtsp_manager.cpp // rtmp_manager.cpp
#include "rtmp_manager.hpp" #include "rtmp_manager.hpp"
#include <arpa/inet.h> #include <arpa/inet.h>
@ -12,14 +12,14 @@
#include <thread> #include <thread>
// ========== 工具函数 ========== // ========== 工具函数 ==========
static bool device_exists(const std::string &path) static bool device_exists(const std::string& path)
{ {
struct stat st; struct stat st;
return (stat(path.c_str(), &st) == 0); return (stat(path.c_str(), &st) == 0);
} }
// 动态获取指定网卡 IPv4 地址 // 动态获取指定网卡 IPv4 地址
static std::string get_ip_address(const std::string &ifname) static std::string get_ip_address(const std::string& ifname)
{ {
struct ifaddrs *ifaddr, *ifa; struct ifaddrs *ifaddr, *ifa;
char ip[INET_ADDRSTRLEN] = {0}; char ip[INET_ADDRSTRLEN] = {0};
@ -36,7 +36,7 @@ static std::string get_ip_address(const std::string &ifname)
if (ifa->ifa_addr->sa_family == AF_INET && ifname == ifa->ifa_name) if (ifa->ifa_addr->sa_family == AF_INET && ifname == ifa->ifa_name)
{ {
void *addr = &((struct sockaddr_in *)ifa->ifa_addr)->sin_addr; void* addr = &((struct sockaddr_in*)ifa->ifa_addr)->sin_addr;
if (inet_ntop(AF_INET, addr, ip, sizeof(ip))) if (inet_ntop(AF_INET, addr, ip, sizeof(ip)))
{ {
freeifaddrs(ifaddr); freeifaddrs(ifaddr);
@ -60,10 +60,10 @@ void RTMPManager::init()
LOG_INFO("[RTMP] GStreamer initialized."); LOG_INFO("[RTMP] GStreamer initialized.");
} }
std::string RTMPManager::make_key(const std::string &name) { return name + "_main"; } std::string RTMPManager::make_key(const std::string& name) { return name + "_main"; }
// ========== 创建推流管线 ========== // ========== 创建推流管线 ==========
GstElement *RTMPManager::create_pipeline(const Camera &cam) GstElement* RTMPManager::create_pipeline(const Camera& cam)
{ {
const int width = cam.width; const int width = cam.width;
const int height = cam.height; const int height = cam.height;
@ -74,28 +74,26 @@ GstElement *RTMPManager::create_pipeline(const Camera &cam)
const std::string app = "camera"; const std::string app = "camera";
const std::string location = "rtmp://127.0.0.1:1935/" + app + "/" + stream_name + "?vhost=live"; const std::string location = "rtmp://127.0.0.1:1935/" + app + "/" + stream_name + "?vhost=live";
// 给关键元素起名字：src, vc, enc, par, mux, sink, tee
std::string pipeline_str = "v4l2src name=src device=" + cam.device + std::string pipeline_str = "v4l2src name=src device=" + cam.device +
" ! video/x-raw,width=" + std::to_string(width) + ",height=" + std::to_string(height) + " ! "
"video/x-raw,format=NV12,width=" +
std::to_string(width) + ",height=" + std::to_string(height) +
",framerate=" + std::to_string(fps) + ",framerate=" + std::to_string(fps) +
"/1 " "/1 ! "
"! videoconvert name=vc ! video/x-raw,format=NV12 " "tee name=t "
"! tee name=t " "t. ! queue max-size-buffers=5 leaky=downstream ! "
"t. ! queue " "mpph264enc bps=" +
"! mpph264enc name=enc bps=" +
std::to_string(bitrate) + " gop=" + std::to_string(fps) + std::to_string(bitrate) + " gop=" + std::to_string(fps) +
" " " rc-mode=cbr ! "
"! h264parse name=par " "h264parse ! flvmux streamable=true name=mux "
"! flvmux name=mux streamable=true "
"audiotestsrc wave=silence ! audioconvert ! audioresample ! voaacenc ! aacparse ! mux. " "audiotestsrc wave=silence ! audioconvert ! audioresample ! voaacenc ! aacparse ! mux. "
"mux. ! rtmpsink name=sink location=\"" + "mux. ! rtmpsink location=\"" +
location + location +
"\" sync=false " "\" sync=false async=false "
// 预留第二路 tee 分支（以后加叠加/探测都行），先丢掉
"t. ! queue ! fakesink sync=false"; "t. ! queue ! fakesink sync=false";
GError *error = nullptr; GError* error = nullptr;
GstElement *pipeline = gst_parse_launch(pipeline_str.c_str(), &error); GstElement* pipeline = gst_parse_launch(pipeline_str.c_str(), &error);
if (error) if (error)
{ {
LOG_ERROR(std::string("[RTMP] Pipeline creation failed: ") + error->message); LOG_ERROR(std::string("[RTMP] Pipeline creation failed: ") + error->message);
@ -106,7 +104,7 @@ GstElement *RTMPManager::create_pipeline(const Camera &cam)
} }
// ========== 主推流循环 ========== // ========== 主推流循环 ==========
void RTMPManager::stream_loop(Camera cam, StreamContext *ctx) void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
{ {
const std::string key = make_key(cam.name); const std::string key = make_key(cam.name);
while (ctx->running) while (ctx->running)
@ -123,7 +121,7 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
} }
// 创建管线 // 创建管线
GstElement *pipeline = create_pipeline(cam); GstElement* pipeline = create_pipeline(cam);
if (!pipeline) if (!pipeline)
{ {
ctx->status.running = false; ctx->status.running = false;
@ -132,22 +130,22 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
continue; continue;
} }
gst_element_set_name(pipeline, key.c_str()); gst_element_set_name(pipeline, key.c_str());
GstBus *bus = gst_element_get_bus(pipeline); GstBus* bus = gst_element_get_bus(pipeline);
// 在 videoconvert 的 src pad 上挂 probe 判断是否有帧流过 // 在 videoconvert 的 src pad 上挂 probe 判断是否有帧流过
bool got_frame = false; bool got_frame = false;
{ {
GstElement *vc = gst_bin_get_by_name(GST_BIN(pipeline), "vc"); GstElement* vc = gst_bin_get_by_name(GST_BIN(pipeline), "vc");
if (vc) if (vc)
{ {
GstPad *pad = gst_element_get_static_pad(vc, "src"); GstPad* pad = gst_element_get_static_pad(vc, "src");
if (pad) if (pad)
{ {
gst_pad_add_probe( gst_pad_add_probe(
pad, GST_PAD_PROBE_TYPE_BUFFER, pad, GST_PAD_PROBE_TYPE_BUFFER,
[](GstPad *, GstPadProbeInfo *, gpointer user_data) -> GstPadProbeReturn [](GstPad*, GstPadProbeInfo*, gpointer user_data) -> GstPadProbeReturn
{ {
*static_cast<bool *>(user_data) = true; *static_cast<bool*>(user_data) = true;
return GST_PAD_PROBE_OK; return GST_PAD_PROBE_OK;
}, },
&got_frame, nullptr); &got_frame, nullptr);
@ -228,7 +226,7 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
} }
// 监听错误/EOS // 监听错误/EOS
GstMessage *msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND, GstMessage* msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND,
(GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS)); (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
if (!msg) continue; if (!msg) continue;
@ -237,7 +235,7 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
{ {
case GST_MESSAGE_ERROR: case GST_MESSAGE_ERROR:
{ {
GError *err = nullptr; GError* err = nullptr;
gst_message_parse_error(msg, &err, nullptr); gst_message_parse_error(msg, &err, nullptr);
ctx->status.running = false; ctx->status.running = false;
ctx->status.last_error = err ? err->message : "GStreamer error"; ctx->status.last_error = err ? err->message : "GStreamer error";
@ -283,7 +281,7 @@ void RTMPManager::start_all()
std::lock_guard<std::mutex> lock(streams_mutex); std::lock_guard<std::mutex> lock(streams_mutex);
int delay_ms = 0; int delay_ms = 0;
for (auto &cam : g_app_config.cameras) for (auto& cam : g_app_config.cameras)
{ {
auto key = make_key(cam.name); auto key = make_key(cam.name);
if (streams.find(key) != streams.end()) if (streams.find(key) != streams.end())
@ -310,20 +308,20 @@ void RTMPManager::start_all()
void RTMPManager::stop_all() void RTMPManager::stop_all()
{ {
std::lock_guard<std::mutex> lock(streams_mutex); std::lock_guard<std::mutex> lock(streams_mutex);
for (auto &kv : streams) kv.second->running.store(false); for (auto& kv : streams) kv.second->running.store(false);
for (auto &kv : streams) for (auto& kv : streams)
if (kv.second->thread.joinable()) kv.second->thread.join(); if (kv.second->thread.joinable()) kv.second->thread.join();
streams.clear(); streams.clear();
} }
bool RTMPManager::is_streaming(const std::string &cam_name) bool RTMPManager::is_streaming(const std::string& cam_name)
{ {
std::lock_guard<std::mutex> lock(streams_mutex); std::lock_guard<std::mutex> lock(streams_mutex);
auto it = streams.find(make_key(cam_name)); auto it = streams.find(make_key(cam_name));
return (it != streams.end() && it->second->status.running); return (it != streams.end() && it->second->status.running);
} }
std::string RTMPManager::get_stream_url(const std::string &cam_name) std::string RTMPManager::get_stream_url(const std::string& cam_name)
{ {
std::string ip = get_ip_address("enP2p33s0"); std::string ip = get_ip_address("enP2p33s0");
if (ip.empty()) ip = "127.0.0.1"; if (ip.empty()) ip = "127.0.0.1";
@ -337,7 +335,7 @@ std::vector<RTMPManager::ChannelInfo> RTMPManager::get_all_channels_status()
for (size_t i = 0; i < g_app_config.cameras.size(); ++i) for (size_t i = 0; i < g_app_config.cameras.size(); ++i)
{ {
const auto &cam = g_app_config.cameras[i]; const auto& cam = g_app_config.cameras[i];
auto key = make_key(cam.name); auto key = make_key(cam.name);
ChannelInfo ch; ChannelInfo ch;
@ -349,7 +347,7 @@ std::vector<RTMPManager::ChannelInfo> RTMPManager::get_all_channels_status()
auto it = streams.find(key); auto it = streams.find(key);
if (it != streams.end()) if (it != streams.end())
{ {
auto &status = it->second->status; auto& status = it->second->status;
ch.running = status.running; ch.running = status.running;
if (status.running) if (status.running)
{ {