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 2026-01-09 09:59:46 +08:00
parent b0bb06dff7
commit b6fd603408
1 changed files with 67 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

319
src/rtmp_manager.cpp
View File

@ -3,12 +3,10 @@
#include <arpa/inet.h> #include <arpa/inet.h>
#include <ifaddrs.h> #include <ifaddrs.h>
#include <net/if.h>
#include <netinet/in.h> #include <netinet/in.h>
#include <sys/stat.h> #include <sys/stat.h>
#include <chrono> #include <chrono>
#include <cstring>
#include <thread> #include <thread>
// ========== 工具函数 ========== // ========== 工具函数 ==========
@ -18,36 +16,25 @@ static bool device_exists(const std::string& path)
return (stat(path.c_str(), &st) == 0); return (stat(path.c_str(), &st) == 0);
} }
// 动态获取指定网卡 IPv4 地址
std::string get_ip_address(const std::string& ifname) std::string get_ip_address(const std::string& ifname)
{ {
struct ifaddrs *ifaddr, *ifa; struct ifaddrs* ifaddr = nullptr;
if (getifaddrs(&ifaddr) == -1) return "";
char ip[INET_ADDRSTRLEN] = {0}; char ip[INET_ADDRSTRLEN] = {0};
for (auto* ifa = ifaddr; ifa; ifa = ifa->ifa_next)
if (getifaddrs(&ifaddr) == -1)
{ {
perror("getifaddrs"); if (!ifa->ifa_addr) continue;
return "";
}
for (ifa = ifaddr; ifa != nullptr; ifa = ifa->ifa_next)
{
if (ifa->ifa_addr == nullptr) continue;
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; auto* addr = &((struct sockaddr_in*)ifa->ifa_addr)->sin_addr;
if (inet_ntop(AF_INET, addr, ip, sizeof(ip))) inet_ntop(AF_INET, addr, ip, sizeof(ip));
{ break;
}
}
freeifaddrs(ifaddr); freeifaddrs(ifaddr);
return ip; return ip;
} }
}
}
freeifaddrs(ifaddr);
return "";
}
// ========== 静态成员 ========== // ========== 静态成员 ==========
std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams; std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
@ -62,102 +49,40 @@ void RTMPManager::init()
std::string RTMPManager::make_key(const std::string& name) { return name + "_main"; } std::string RTMPManager::make_key(const std::string& name) { return name + "_main"; }
// ========== 创建推流管线 ========== // ========== 创建 RTMP Pipeline ==========
// GstElement* RTMPManager::create_pipeline(const Camera& cam)
// {
// const int width = cam.width;
// const int height = cam.height;
// const int fps = cam.fps;
// const int bitrate = cam.bitrate;
// // MediaMTX 中的 stream key
// const std::string stream_name = cam.name;
// // RTMP 推送到 MediaMTX
// // mediamtx.yml 中 paths 会自动创建
// const std::string rtmp_url = "rtmp://127.0.0.1:1935/" + stream_name;
// /*
// * Pipeline 说明：
// * v4l2src -> mpph264enc -> h264parse -> flvmux -> rtmpsink
// *
// * - 不使用 tee（降低死锁概率）
// * - 不引入音频（MediaMTX 不强制要求）
// * - 纯视频、纯 RTMP、纯 TCP
// */
// std::string pipeline_str = "v4l2src name=src device=" + cam.device +
// " ! video/x-raw,format=NV12,width=" + std::to_string(width) +
// ",height=" + std::to_string(height) + ",framerate=" + std::to_string(fps) +
// "/1 "
// " ! mpph264enc bps=" +
// std::to_string(bitrate) + " gop=" + std::to_string(fps) +
// " rc-mode=cbr "
// " ! h264parse name=parse "
// " ! flvmux streamable=true "
// " ! rtmpsink location=\"" +
// rtmp_url +
// "\" "
// " sync=false async=false";
// GError* error = nullptr;
// GstElement* pipeline = gst_parse_launch(pipeline_str.c_str(), &error);
// if (error)
// {
// LOG_ERROR(std::string("[RTMP] Pipeline creation failed: ") + error->message);
// g_error_free(error);
// return nullptr;
// }
// return pipeline;
// }
GstElement* RTMPManager::create_pipeline(const Camera& cam) GstElement* RTMPManager::create_pipeline(const Camera& cam)
{ {
const int fps = 30; const int gop = 15;
const int bitrate = cam.bitrate;
const int gop = 15; // ⭐ 从 4 改为 15（非常重要）
const std::string stream_name = cam.name; const std::string rtmp_url = "rtmp://127.0.0.1:1935/" + cam.name;
const std::string rtmp_url = "rtmp://127.0.0.1:1935/" + stream_name;
std::string pipeline_str = "v4l2src device=" + cam.device + std::string pipeline_str = "v4l2src device=" + cam.device +
" io-mode=dmabuf " " io-mode=dmabuf "
"! video/x-raw,format=NV12," "! video/x-raw,format=NV12,width=1280,height=960,framerate=30/1 "
"width=1280,height=960,"
"framerate=30/1 "
"! videocrop top=120 bottom=120 " "! videocrop top=120 bottom=120 "
"! videoscale " "! videoscale "
"! video/x-raw," "! video/x-raw,width=" +
"width=" +
std::to_string(cam.width) + ",height=" + std::to_string(cam.height) + std::to_string(cam.width) + ",height=" + std::to_string(cam.height) +
" " " "
// ⭐ 缓冲拉开，避免瞬时饿死
"! queue max-size-buffers=8 max-size-time=0 leaky=downstream " "! queue max-size-buffers=8 max-size-time=0 leaky=downstream "
"! mpph264enc " "! mpph264enc rc-mode=cbr "
"rc-mode=cbr "
"bps=" + "bps=" +
std::to_string(bitrate) + std::to_string(cam.bitrate) +
" " " "
"gop=" + "gop=" +
std::to_string(gop) + std::to_string(gop) +
" " " "
"header-mode=each-idr " "header-mode=each-idr profile=main "
"profile=main "
"! h264parse config-interval=1 " "! h264parse name=parse config-interval=1 "
"! video/x-h264,stream-format=avc,alignment=au " "! video/x-h264,stream-format=avc,alignment=au "
"! flvmux streamable=true " "! flvmux streamable=true "
"! rtmpsink location=\"" + "! rtmpsink location=\"" +
rtmp_url + rtmp_url + "\" sync=false async=false";
"\" "
"sync=false async=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);
@ -167,7 +92,6 @@ GstElement* RTMPManager::create_pipeline(const Camera& cam)
g_error_free(error); g_error_free(error);
return nullptr; return nullptr;
} }
return pipeline; return pipeline;
} }
@ -176,208 +100,137 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
{ {
const std::string key = make_key(cam.name); const std::string key = make_key(cam.name);
constexpr int64_t START_TIMEOUT_MS = 5000; // 启动阶段无帧 constexpr int64_t START_TIMEOUT_MS = 12000; // ⭐ 启动宽限期
constexpr int64_t NO_FRAME_TIMEOUT_MS = 10000; // 运行阶段突然无帧 ⇒ pipeline 卡死重启 constexpr int64_t NO_FRAME_TIMEOUT_MS = 10000;
while (ctx->thread_running) while (ctx->thread_running)
{ {
// 1) 检查设备节点 if (!device_exists(cam.device))
struct stat st{};
if (stat(cam.device.c_str(), &st) != 0)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); LOG_WARN("[RTMP] " + key + " - device not found");
ctx->status.running = false;
ctx->status.last_error = "Device not found: " + cam.device;
LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
std::this_thread::sleep_for(std::chrono::seconds(3)); std::this_thread::sleep_for(std::chrono::seconds(3));
continue; continue;
} }
// 2) 创建 pipeline
GstElement* pipeline = create_pipeline(cam); GstElement* pipeline = create_pipeline(cam);
if (!pipeline) if (!pipeline)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false;
ctx->status.last_error = "Pipeline creation failed";
std::this_thread::sleep_for(std::chrono::seconds(3)); std::this_thread::sleep_for(std::chrono::seconds(3));
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);
// 3) 帧探测：记录 last_frame_ms
ctx->last_frame_ms.store(0, std::memory_order_relaxed); ctx->last_frame_ms.store(0, std::memory_order_relaxed);
// ⭐ 帧探测挂在 h264parse
if (GstElement* parse = gst_bin_get_by_name(GST_BIN(pipeline), "parse"))
{ {
GstElement* src = gst_bin_get_by_name(GST_BIN(pipeline), "src"); if (GstPad* pad = gst_element_get_static_pad(parse, "src"))
if (src)
{
GstPad* pad = gst_element_get_static_pad(src, "src");
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 data) -> GstPadProbeReturn [](GstPad*, GstPadProbeInfo*, gpointer data) -> GstPadProbeReturn
{ {
auto ts = static_cast<std::atomic<int64_t>*>(data); auto* ts = static_cast<std::atomic<int64_t>*>(data);
auto now = std::chrono::steady_clock::now().time_since_epoch(); auto now = std::chrono::steady_clock::now().time_since_epoch();
auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(now).count(); ts->store(std::chrono::duration_cast<std::chrono::milliseconds>(now).count(),
ts->store(ms, std::memory_order_relaxed); std::memory_order_relaxed);
return GST_PAD_PROBE_OK; return GST_PAD_PROBE_OK;
}, },
&(ctx->last_frame_ms), nullptr); &ctx->last_frame_ms, nullptr);
gst_object_unref(pad); gst_object_unref(pad);
} }
gst_object_unref(src); gst_object_unref(parse);
}
} }
// 4) 启动 pipeline
LOG_INFO("[RTMP] Starting stream: " + key); LOG_INFO("[RTMP] Starting stream: " + key);
gst_element_set_state(pipeline, GST_STATE_PLAYING); gst_element_set_state(pipeline, GST_STATE_PLAYING);
GstState state = GST_STATE_NULL, pending = GST_STATE_NULL; GstState state;
if (gst_element_get_state(pipeline, &state, &pending, 5 * GST_SECOND) != GST_STATE_CHANGE_SUCCESS || if (gst_element_get_state(pipeline, &state, nullptr, 5 * GST_SECOND) != GST_STATE_CHANGE_SUCCESS ||
state != GST_STATE_PLAYING) state != GST_STATE_PLAYING)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); LOG_WARN("[RTMP] " + key + " failed to enter PLAYING");
ctx->status.running = false;
ctx->status.last_error = "Failed to enter PLAYING";
LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
gst_element_set_state(pipeline, GST_STATE_NULL); gst_element_set_state(pipeline, GST_STATE_NULL);
if (bus) gst_object_unref(bus); gst_object_unref(bus);
gst_object_unref(pipeline); gst_object_unref(pipeline);
std::this_thread::sleep_for(std::chrono::seconds(3));
std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS));
continue; continue;
} }
// pipeline 成功启动
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = true;
ctx->status.last_error.clear();
}
LOG_INFO("[RTMP] " + key + " confirmed PLAYING"); LOG_INFO("[RTMP] " + key + " confirmed PLAYING");
auto launch_tp = std::chrono::steady_clock::now(); auto launch_tp = std::chrono::steady_clock::now();
bool entered_running = false;
bool need_restart = false; bool need_restart = false;
// 5) 主循环：检测停帧或 error
while (ctx->thread_running) while (ctx->thread_running)
{ {
auto now = std::chrono::steady_clock::now(); auto now = std::chrono::steady_clock::now();
int64_t now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count(); int64_t now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
int64_t last_ms = ctx->last_frame_ms.load(std::memory_order_relaxed); int64_t last_ms = ctx->last_frame_ms.load(std::memory_order_relaxed);
if (last_ms == 0) if (!entered_running)
{ {
auto startup_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - launch_tp).count(); if (last_ms != 0)
if (startup_ms > START_TIMEOUT_MS)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); entered_running = true;
ctx->status.running = false; LOG_INFO("[RTMP] " + key + " entered RUNNING state");
ctx->status.last_error = "No frames detected at startup"; }
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error); else if (std::chrono::duration_cast<std::chrono::milliseconds>(now - launch_tp).count() >
START_TIMEOUT_MS)
{
LOG_ERROR("[RTMP] " + key + " - no frames during startup");
need_restart = true; need_restart = true;
break; break;
} }
} }
else else
{ {
int64_t idle_ms = now_ms - last_ms; if (now_ms - last_ms > NO_FRAME_TIMEOUT_MS)
if (idle_ms > NO_FRAME_TIMEOUT_MS)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); LOG_ERROR("[RTMP] " + key + " - frame stalled");
ctx->status.running = false;
ctx->status.last_error = "Frame stalled for " + std::to_string(idle_ms) + " ms";
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
need_restart = true; need_restart = true;
break; break;
} }
else
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = true;
ctx->status.last_error.clear();
}
} }
// 错误消息检测 if (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 (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ERROR)
{ {
GError* err = nullptr;
gst_message_parse_error(msg, &err, nullptr);
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false;
ctx->status.last_error = err ? err->message : "GStreamer error";
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
if (err) g_error_free(err);
need_restart = true; need_restart = true;
}
else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_EOS)
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false;
ctx->status.last_error = "EOS (End of stream)";
LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
need_restart = true;
}
gst_message_unref(msg); gst_message_unref(msg);
if (need_restart) break; break;
}
} }
// 清理 & 重建
gst_element_set_state(pipeline, GST_STATE_NULL); gst_element_set_state(pipeline, GST_STATE_NULL);
if (bus) gst_object_unref(bus); gst_object_unref(bus);
gst_object_unref(pipeline); gst_object_unref(pipeline);
if (ctx->thread_running) if (ctx->thread_running && need_restart)
{ {
LOG_WARN("[RTMP] Restarting " + key + " in 3s..."); LOG_WARN("[RTMP] Restarting " + key + " in 3s...");
std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS)); std::this_thread::sleep_for(std::chrono::seconds(3));
} }
} }
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false;
}
LOG_INFO("[RTMP] Stream thread exited for " + key); LOG_INFO("[RTMP] Stream thread exited for " + key);
} }
// ========== 启停与状态 ========== // ========== 启停 ==========
void RTMPManager::start_all() void RTMPManager::start_all()
{ {
LOG_INFO("[RTMP] Starting enabled record streams...");
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 (const auto& cam : g_app_config.cameras) for (const auto& cam : g_app_config.cameras)
{ {
// 跳过未启用摄像头 if (!cam.enabled) continue;
if (!cam.enabled)
{
LOG_INFO("[RTMP] Skip disabled camera: " + cam.name);
continue;
}
auto key = make_key(cam.name); auto key = make_key(cam.name);
if (streams.find(key) != streams.end())
{
LOG_INFO("[RTMP] Stream already running: " + key);
continue;
}
auto ctx = std::make_unique<StreamContext>(); auto ctx = std::make_unique<StreamContext>();
ctx->thread_running.store(true); ctx->thread_running.store(true);
@ -402,61 +255,9 @@ void RTMPManager::stop_all()
streams.clear(); streams.clear();
} }
bool RTMPManager::is_streaming(const std::string& cam_name)
{
std::lock_guard<std::mutex> lock(streams_mutex);
auto it = streams.find(make_key(cam_name));
if (it == streams.end()) return false;
auto& ctx = *(it->second);
std::lock_guard<std::mutex> lk(ctx.status_mutex);
return ctx.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";
return "rtsp://" + ip + ":18554/" + cam_name; return "rtsp://" + ip + ":18554/" + cam_name;
} }
// ========== 汇总状态 ==========
std::vector<RTMPManager::ChannelInfo> RTMPManager::get_all_channels_status()
{
std::vector<ChannelInfo> result;
std::lock_guard<std::mutex> lock(streams_mutex);
for (size_t i = 0; i < g_app_config.cameras.size(); ++i)
{
const auto& cam = g_app_config.cameras[i];
auto key = make_key(cam.name);
ChannelInfo ch;
ch.loc = static_cast<int>(i);
ch.url.clear();
ch.running = false;
ch.reason = "Not started";
auto it = streams.find(key);
if (it != streams.end())
{
auto& ctx = *(it->second);
std::lock_guard<std::mutex> lk(ctx.status_mutex);
auto& status = it->second->status;
ch.running = status.running;
if (status.running)
{
ch.url = get_stream_url(cam.name);
ch.reason.clear();
}
else
{
ch.reason = status.last_error.empty() ? "Unknown error" : status.last_error;
}
}
result.push_back(ch);
}
return result;
}