cxh/sweeper_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-29 14:24:40 +08:00
parent 2b7ff4e056
commit d4531c99b0
1 changed files with 131 additions and 199 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

316
src/rtmp_manager.cpp
View File

@ -3,7 +3,6 @@
#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>
@ -11,179 +10,146 @@
#include <cstring> #include <cstring>
#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
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 = nullptr;
if (getifaddrs(&ifaddr) != 0) 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 = &reinterpret_cast<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;
std::mutex RTMPManager::streams_mutex; std::mutex RTMPManager::streams_mutex;
// ========== 初始化 ========== // =======================================================
// 初始化
// =======================================================
void RTMPManager::init() void RTMPManager::init()
{ {
gst_init(nullptr, nullptr); gst_init(nullptr, nullptr);
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 std::string stream = cam.name + "_main";
const int height = cam.height;
const int fps = cam.fps;
const int bitrate = cam.bitrate;
const std::string stream_name = cam.name + "_main";
const std::string app = "camera"; const std::string app = "camera";
// 两个不同的 SRS 实例：live 用于远控，record 用于录像
// const std::string live_rtmp = "rtmp://36.153.162.171:19435/" + app + "/" + stream_name + "?vhost=live";
const std::string live_rtmp = "rtmp://36.153.162.171:19435/" + app + "/" + stream_name + "?vhost=live";
const std::string record_rtmp = "rtmp://127.0.0.1:2935/" + app + "/" + stream_name + "?vhost=record";
std::string pipeline_str = "v4l2src device=" + cam.device + const std::string live_rtmp = "rtmp://36.153.162.171:19435/" + app + "/" + stream + "?vhost=live";
std::string pipeline_str = "v4l2src name=src device=" + cam.device +
" io-mode=dmabuf " " io-mode=dmabuf "
"! video/x-raw,format=NV12,width=" + "! video/x-raw,format=NV12,"
std::to_string(width) + ",height=" + std::to_string(height) + "width=" +
",framerate=" + std::to_string(fps) + std::to_string(cam.width) + ",height=" + std::to_string(cam.height) +
",framerate=" + std::to_string(cam.fps) +
"/1 " "/1 "
"! queue max-size-buffers=2 leaky=downstream " "! queue max-size-buffers=4 max-size-time=0 leaky=downstream "
"! mpph264enc " "! mpph264enc "
"bps=" + "bps=" +
std::to_string(bitrate) + std::to_string(cam.bitrate) +
" " " "
"gop=" + "gop=" +
std::to_string(fps) + std::to_string(cam.fps) +
" " " "
"rc-mode=cbr " "rc-mode=cbr "
"header-mode=each-idr " "header-mode=each-idr "
"! h264parse " "! h264parse config-interval=1 "
"! flvmux streamable=true " "! flvmux streamable=true "
"! rtmpsink location=\"" + "! rtmpsink location=\"" +
live_rtmp + "\" sync=false async=false"; live_rtmp +
"\" "
"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);
if (error) if (error)
{ {
LOG_ERROR(std::string("[RTMP] Pipeline creation failed: ") + error->message); LOG_ERROR("[RTMP] Pipeline creation failed: " + std::string(error->message));
g_error_free(error); g_error_free(error);
return nullptr; return nullptr;
} }
return pipeline; return pipeline;
} }
// ========== 主推流循环 ========== // =======================================================
// 主推流线程
// =======================================================
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->thread_running) while (ctx->thread_running)
{ {
// 1. 检查设备节点是否存在 // 1. 设备存在性
struct stat st{}; if (!device_exists(cam.device))
if (stat(cam.device.c_str(), &st) != 0) {
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false; ctx->status.running = false;
ctx->status.last_error = "Device not found: " + cam.device; ctx->status.last_error = "Device not found: " + cam.device;
LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error); }
LOG_WARN("[RTMP] " + key + " - device not found");
std::this_thread::sleep_for(std::chrono::seconds(3)); std::this_thread::sleep_for(std::chrono::seconds(3));
continue; continue;
} }
// 2. 创建 GStreamer 管线 // 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); std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false; ctx->status.running = false;
ctx->status.last_error = "Pipeline creation failed"; 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());
GstBus* bus = gst_element_get_bus(pipeline); GstBus* bus = gst_element_get_bus(pipeline);
// 3. 在 v4l2src 的 src pad 上挂探测 probe
bool got_frame = false;
{
GstElement* src = gst_bin_get_by_name(GST_BIN(pipeline), "src");
if (src)
{
GstPad* pad = gst_element_get_static_pad(src, "src");
if (pad)
{
gst_pad_add_probe(
pad, GST_PAD_PROBE_TYPE_BUFFER,
[](GstPad*, GstPadProbeInfo*, gpointer user_data) -> GstPadProbeReturn
{
*static_cast<bool*>(user_data) = true;
return GST_PAD_PROBE_OK;
},
&got_frame, nullptr);
gst_object_unref(pad);
}
else
{
LOG_WARN("[RTMP] " + key + " - src has no 'src' pad?");
}
gst_object_unref(src);
}
else
{
LOG_WARN("[RTMP] " + key + " - cannot find element 'src' for pad-probe");
}
}
// 4. 启动播放
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);
// 等待进入 PLAYING 状态（最长 5s） // 3. 等待进入 PLAYING
GstState state = GST_STATE_NULL, pending = GST_STATE_NULL; GstState state = GST_STATE_NULL;
bool confirmed_running = false; if (gst_element_get_state(pipeline, &state, nullptr, 5 * GST_SECOND) == GST_STATE_CHANGE_SUCCESS &&
if (gst_element_get_state(pipeline, &state, &pending, 5 * GST_SECOND) == GST_STATE_CHANGE_SUCCESS &&
state == GST_STATE_PLAYING) state == GST_STATE_PLAYING)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); std::lock_guard<std::mutex> lk(ctx->status_mutex);
confirmed_running = true;
ctx->status.running = true; ctx->status.running = true;
ctx->status.last_error.clear(); ctx->status.last_error.clear();
LOG_INFO("[RTMP] " + key + " confirmed PLAYING"); LOG_INFO("[RTMP] " + key + " confirmed PLAYING");
@ -192,95 +158,66 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false; ctx->status.running = false;
ctx->status.last_error = "Pipeline failed to confirm PLAYING"; ctx->status.last_error = "Failed to reach PLAYING";
LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error); LOG_ERROR("[RTMP] " + key + " failed to PLAY");
goto cleanup;
} }
if (!confirmed_running) // 4. 正常运行：只监听 ERROR / EOS
while (ctx->thread_running)
{ {
gst_element_set_state(pipeline, GST_STATE_NULL); GstMessage* msg = gst_bus_timed_pop_filtered(
if (bus) gst_object_unref(bus); bus, 300 * GST_MSECOND, static_cast<GstMessageType>(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
gst_object_unref(pipeline);
std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS)); if (!msg)
{
// 心跳：pipeline 没死就认为在跑
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = true;
continue; continue;
} }
// 5. 运行阶段：监测帧和错误 if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ERROR)
const auto start_t = std::chrono::steady_clock::now();
bool need_restart = false;
while (ctx->thread_running)
{
// 检查是否收到帧（前 5s 内）
if (!got_frame)
{
auto elapsed =
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start_t)
.count();
if (elapsed > 5)
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false;
ctx->status.last_error = "No frames detected (no video signal)";
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
need_restart = true;
break;
}
}
else
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
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;
switch (GST_MESSAGE_TYPE(msg))
{
case GST_MESSAGE_ERROR:
{ {
GError* err = nullptr; GError* err = nullptr;
gst_message_parse_error(msg, &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 + " stream error: " + ctx->status.last_error);
if (err) g_error_free(err);
need_restart = true;
break;
}
case GST_MESSAGE_EOS:
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false; ctx->status.running = false;
ctx->status.last_error = "End of stream (EOS)"; ctx->status.last_error = err ? err->message : "Unknown GStreamer error";
LOG_WARN("[RTMP] " + key + " reached EOS");
need_restart = true;
break;
}
default:
break;
} }
LOG_ERROR("[RTMP] " + key + " ERROR: " + (err ? err->message : "unknown"));
if (err) g_error_free(err);
gst_message_unref(msg); gst_message_unref(msg);
break;
if (need_restart) break;
} }
// 6. 收尾清理 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";
}
LOG_WARN("[RTMP] " + key + " EOS");
gst_message_unref(msg);
break;
}
gst_message_unref(msg);
}
cleanup:
gst_element_set_state(pipeline, GST_STATE_NULL); gst_element_set_state(pipeline, GST_STATE_NULL);
if (bus) gst_object_unref(bus); if (bus) gst_object_unref(bus);
gst_object_unref(pipeline); gst_object_unref(pipeline);
// 7. 若仍在运行状态，准备重启
if (ctx->thread_running) if (ctx->thread_running)
{ {
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));
} }
} }
@ -291,10 +228,12 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
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..."); LOG_INFO("[RTMP] Starting enabled 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;
@ -306,12 +245,8 @@ void RTMPManager::start_all()
continue; continue;
} }
auto key = make_key(cam.name); const auto key = make_key(cam.name);
if (streams.find(key) != streams.end()) if (streams.count(key)) continue;
{
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);
@ -332,29 +267,37 @@ 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->thread_running.store(false); for (auto& kv : streams) kv.second->thread_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();
} }
// =======================================================
// 状态 & URL
// =======================================================
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));
if (it == streams.end()) return false; if (it == streams.end()) return false;
auto& ctx = *(it->second); std::lock_guard<std::mutex> lk(it->second->status_mutex);
std::lock_guard<std::mutex> lk(ctx.status_mutex); return it->second->status.running;
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 "http://" + ip + ":1985/rtc/v1/whep/?app=camera&stream=" + cam_name + "_main&vhost=live";
return "http://" + ip + ":11985/rtc/v1/whep/?app=camera&stream=" + cam_name + "_main&vhost=live";
} }
// ========== 汇总状态 ==========
// =======================================================
// 汇总状态
// =======================================================
std::vector<RTMPManager::ChannelInfo> RTMPManager::get_all_channels_status() std::vector<RTMPManager::ChannelInfo> RTMPManager::get_all_channels_status()
{ {
std::vector<ChannelInfo> result; std::vector<ChannelInfo> result;
@ -363,40 +306,29 @@ 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); ChannelInfo ch{};
ChannelInfo ch;
ch.loc = static_cast<int>(i); ch.loc = static_cast<int>(i);
ch.url.clear();
ch.running = false;
if (!cam.enabled) if (!cam.enabled)
{ {
ch.running = false;
ch.reason = "Disabled by config"; ch.reason = "Disabled by config";
result.push_back(ch); result.push_back(ch);
continue; continue;
} }
auto it = streams.find(key); auto it = streams.find(make_key(cam.name));
if (it != streams.end()) if (it != streams.end())
{ {
auto& ctx = *(it->second); std::lock_guard<std::mutex> lk(it->second->status_mutex);
std::lock_guard<std::mutex> lk(ctx.status_mutex); ch.running = it->second->status.running;
ch.reason = it->second->status.last_error;
ch.running = ctx.status.running; if (ch.running) ch.url = get_stream_url(cam.name);
if (ctx.status.running)
{
ch.url = get_stream_url(cam.name);
ch.reason.clear();
} }
else else
{ {
ch.reason = ctx.status.last_error.empty() ? "Stopped" : ctx.status.last_error; ch.running = false;
} ch.reason = "Not started";
}
else
{
ch.reason = "Enabled but not started";
} }
result.push_back(ch); result.push_back(ch);