cxh/kunlang_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-10-17 16:27:04 +08:00
parent 9f5a71de51
commit 0b53cf04d0
1 changed files with 97 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
src/rtmp_manager.cpp
View File

@ -65,37 +65,40 @@ std::string RTMPManager::make_key(const std::string &name) { return name + "_mai
// ========== 创建推流管线 ==========
GstElement *RTMPManager::create_pipeline(const Camera &cam)
{
int width = cam.width;
int height = cam.height;
int fps = cam.fps;
int bitrate = cam.bitrate;
const int width = cam.width;
const int height = cam.height;
const int fps = cam.fps;
const int bitrate = cam.bitrate;
std::string stream_name = cam.name + "_main";
std::string app = "record";
std::string location = "rtmp://127.0.0.1:1935/" + app + "/" + stream_name + "?vhost=" + app;
const std::string stream_name = cam.name + "_main";
const std::string app = "record";
const std::string location = "rtmp://127.0.0.1:1935/" + app + "/" + stream_name + "?vhost=" + app;
// fpsdisplaysink 探测首帧
std::string pipeline_str =
"v4l2src device=" + cam.device + " ! video/x-raw,width=" + std::to_string(width) +
",height=" + std::to_string(height) + ",framerate=" + std::to_string(fps) +
"/1 ! videoconvert ! video/x-raw,format=NV12 "
// 给关键元素起名字：src, vc, enc, par, mux, sink, tee
std::string pipeline_str = "v4l2src name=src device=" + cam.device +
" ! video/x-raw,width=" + std::to_string(width) + ",height=" + std::to_string(height) +
",framerate=" + std::to_string(fps) +
"/1 "
"! videoconvert name=vc ! video/x-raw,format=NV12 "
"! tee name=t "
"t. ! queue ! mpph264enc bps=" +
"t. ! queue "
"! mpph264enc name=enc bps=" +
std::to_string(bitrate) + " gop=" + std::to_string(fps) +
" ! h264parse ! flvmux name=mux streamable=true "
" "
"! h264parse name=par "
"! flvmux name=mux streamable=true "
"audiotestsrc wave=silence ! audioconvert ! audioresample ! voaacenc ! aacparse ! mux. "
"mux. ! rtmpsink location=\"" +
"mux. ! rtmpsink name=sink location=\"" +
location +
"\" sync=false "
"t. ! queue ! fpsdisplaysink name=fpsprobe text-overlay=false video-sink=fakesink sync=false";
// LOG_INFO("[RTMP] Pipeline: " + pipeline_str);
// 预留第二路 tee 分支（以后加叠加/探测都行），先丢掉
"t. ! queue ! fakesink sync=false";
GError *error = nullptr;
GstElement *pipeline = gst_parse_launch(pipeline_str.c_str(), &error);
if (error)
{
LOG_ERROR("[RTMP] Pipeline creation failed: " + std::string(error->message));
LOG_ERROR(std::string("[RTMP] Pipeline creation failed: ") + error->message);
g_error_free(error);
return nullptr;
}
@ -105,93 +108,79 @@ GstElement *RTMPManager::create_pipeline(const Camera &cam)
// ========== 主推流循环 ==========
void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
{
std::string key = make_key(cam.name);
int consecutive_failures = 0;
const std::string key = make_key(cam.name);
while (ctx->running)
{
// 检查设备是否存在
if (!device_exists(cam.device))
// 设备是否存在
struct stat st{};
if (stat(cam.device.c_str(), &st) != 0)
{
ctx->status.running = false;
ctx->status.last_error = "Device not found: " + cam.device;
LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
consecutive_failures++;
std::this_thread::sleep_for(std::chrono::seconds(3));
continue;
}
// 创建 GStreamer 管线
// 创建管线
GstElement *pipeline = create_pipeline(cam);
if (!pipeline)
{
ctx->status.running = false;
ctx->status.last_error = "Pipeline creation failed";
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
consecutive_failures++;
std::this_thread::sleep_for(std::chrono::seconds(3));
continue;
}
gst_element_set_name(pipeline, key.c_str());
GstBus *bus = gst_element_get_bus(pipeline);
// ======== Pad Probe 探测帧 ==========
// 在 videoconvert 的 src pad 上挂 probe 判断是否有帧流过
bool got_frame = false;
GstElement *src = gst_bin_get_by_name(GST_BIN(pipeline), "v4l2src0");
if (!src) src = gst_bin_get_by_interface(GST_BIN(pipeline), GST_TYPE_ELEMENT);
if (src)
{
GstPad *pad = gst_element_get_static_pad(src, "src");
GstElement *vc = gst_bin_get_by_name(GST_BIN(pipeline), "vc");
if (vc)
{
GstPad *pad = gst_element_get_static_pad(vc, "src");
if (pad)
{
gst_pad_add_probe(
pad, GST_PAD_PROBE_TYPE_BUFFER,
[](GstPad *, GstPadProbeInfo *, gpointer user_data) -> GstPadProbeReturn
{
auto *flag = static_cast<bool *>(user_data);
*flag = true; // 收到帧数据
*static_cast<bool *>(user_data) = true;
return GST_PAD_PROBE_OK;
},
&got_frame, nullptr);
gst_object_unref(pad);
}
gst_object_unref(src);
else
{
LOG_WARN("[RTMP] " + key + " - vc has no src pad?");
}
gst_object_unref(vc);
}
else
{
LOG_WARN("[RTMP] " + key + " - cannot find element 'vc' for pad-probe");
}
}
// ======== 启动推流 ========
gst_element_set_state(pipeline, GST_STATE_PLAYING);
// 启动
LOG_INFO("[RTMP] Starting stream: " + key);
gst_element_set_state(pipeline, GST_STATE_PLAYING);
// 等待进入 PLAYING（最长 5s）
bool confirmed_running = false;
bool need_restart = false;
auto start_time = std::chrono::steady_clock::now();
// ======== 等待 pipeline 确认启动 ========
{
bool reached_playing = false;
auto check_start = std::chrono::steady_clock::now();
while (std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - check_start)
.count() < 2000)
{
GstState state;
gst_element_get_state(pipeline, &state, nullptr, 200 * GST_MSECOND);
if (state == GST_STATE_PLAYING)
{
reached_playing = true;
break;
}
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
if (reached_playing)
GstState state = GST_STATE_NULL, pending = GST_STATE_NULL;
// 注意：第三个参数单位是纳秒，5s=5*GST_SECOND
if (gst_element_get_state(pipeline, &state, &pending, 5 * GST_SECOND) == GST_STATE_CHANGE_SUCCESS &&
state == GST_STATE_PLAYING)
{
confirmed_running = true;
ctx->status.running = true;
ctx->status.last_error.clear();
confirmed_running = true;
LOG_INFO("[RTMP] " + key + " confirmed running (PLAYING).");
LOG_INFO("[RTMP] " + key + " confirmed PLAYING");
}
else
{
@ -210,17 +199,19 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
continue;
}
// ======== 主循环监听消息 ========
// 运行中：5s 内必须看到帧，否则认定无信号重启
const auto start_t = std::chrono::steady_clock::now();
bool need_restart = false;
while (ctx->running)
{
GstMessage *msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND,
(GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
// 检查是否超时无帧（5 秒）
// 先检查帧
if (!got_frame)
{
auto elapsed =
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start_time).count();
if (!got_frame && elapsed > 5)
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start_t)
.count();
if (elapsed > 5)
{
ctx->status.running = false;
ctx->status.last_error = "No frames detected (no video signal)";
@ -228,6 +219,17 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
need_restart = true;
break;
}
}
else
{
// 一旦有帧，保持 running=true
ctx->status.running = true;
ctx->status.last_error.clear();
}
// 监听错误/EOS
GstMessage *msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND,
(GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
if (!msg) continue;
@ -238,7 +240,7 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
GError *err = nullptr;
gst_message_parse_error(msg, &err, nullptr);
ctx->status.running = false;
ctx->status.last_error = err ? std::string(err->message) : "GStreamer error";
ctx->status.last_error = err ? err->message : "GStreamer error";
LOG_ERROR("[RTMP] " + key + " stream error: " + ctx->status.last_error);
if (err) g_error_free(err);
need_restart = true;
@ -253,21 +255,19 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
default:
break;
}
gst_message_unref(msg);
if (need_restart) break;
}
// ======== 收尾与重启 ========
// 收尾
gst_element_set_state(pipeline, GST_STATE_NULL);
if (bus) gst_object_unref(bus);
gst_object_unref(pipeline);
if (ctx->running)
{
consecutive_failures++;
LOG_WARN("[RTMP] Restarting " + key + " in 3s... (fail count " + std::to_string(consecutive_failures) +
")");
LOG_WARN("[RTMP] Restarting " + key + " in 3s...");
std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS));
}
}