277 lines
9.2 KiB
C++
277 lines
9.2 KiB
C++
// rtmp_manager.cpp
|
|
#include "rtmp_manager.hpp"
|
|
|
|
#include <arpa/inet.h>
|
|
#include <ifaddrs.h>
|
|
#include <netinet/in.h>
|
|
#include <sys/stat.h>
|
|
|
|
#include <chrono>
|
|
#include <thread>
|
|
|
|
// ========== 工具函数 ==========
|
|
static bool device_exists(const std::string& path)
|
|
{
|
|
struct stat st;
|
|
return (stat(path.c_str(), &st) == 0);
|
|
}
|
|
|
|
std::string get_ip_address(const std::string& ifname)
|
|
{
|
|
struct ifaddrs* ifaddr = nullptr;
|
|
if (getifaddrs(&ifaddr) == -1) return "";
|
|
|
|
char ip[INET_ADDRSTRLEN] = {0};
|
|
for (auto* ifa = ifaddr; ifa; ifa = ifa->ifa_next)
|
|
{
|
|
if (!ifa->ifa_addr) continue;
|
|
if (ifa->ifa_addr->sa_family == AF_INET && ifname == ifa->ifa_name)
|
|
{
|
|
auto* addr = &((struct sockaddr_in*)ifa->ifa_addr)->sin_addr;
|
|
inet_ntop(AF_INET, addr, ip, sizeof(ip));
|
|
break;
|
|
}
|
|
}
|
|
freeifaddrs(ifaddr);
|
|
return ip;
|
|
}
|
|
|
|
// ========== 静态成员 ==========
|
|
std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
|
|
std::mutex RTMPManager::streams_mutex;
|
|
|
|
// ========== 初始化 ==========
|
|
void RTMPManager::init()
|
|
{
|
|
gst_init(nullptr, nullptr);
|
|
LOG_INFO("[RTMP] GStreamer initialized.");
|
|
}
|
|
|
|
std::string RTMPManager::make_key(const std::string& name) { return name + "_main"; }
|
|
|
|
// ========== 创建 RTMP Pipeline ==========
|
|
GstElement* RTMPManager::create_pipeline(const Camera& cam)
|
|
{
|
|
const std::string rtmp_url = "rtmp://127.0.0.1:1935/" + cam.name;
|
|
|
|
std::string pipeline_str = "v4l2src device=" + cam.device +
|
|
" io-mode=dmabuf "
|
|
|
|
// ⭐ 强制打时间戳（关键）
|
|
"! identity do-timestamp=true "
|
|
|
|
// 原始 4:3 输入
|
|
"! video/x-raw,format=NV12,width=1280,height=960,framerate=30/1 "
|
|
|
|
// 裁成 16:9
|
|
"! videocrop top=120 bottom=120 "
|
|
|
|
// 缩放到目标分辨率
|
|
"! videoscale "
|
|
"! video/x-raw,width=" +
|
|
std::to_string(cam.width) + ",height=" + std::to_string(cam.height) +
|
|
" "
|
|
|
|
// 缓冲（只在这里允许丢帧）
|
|
"! queue max-size-buffers=12 max-size-time=0 leaky=downstream "
|
|
|
|
// 编码
|
|
"! mpph264enc rc-mode=cbr "
|
|
"bps=" +
|
|
std::to_string(cam.bitrate) +
|
|
" "
|
|
"gop=30 "
|
|
"header-mode=each-idr "
|
|
"profile=main "
|
|
|
|
// 解析 + 帧探针位置
|
|
"! h264parse name=parse config-interval=1 "
|
|
"! video/x-h264,stream-format=avc,alignment=au "
|
|
|
|
// RTMP
|
|
"! flvmux streamable=true "
|
|
"! rtmpsink location=\"" +
|
|
rtmp_url +
|
|
"\" "
|
|
"sync=false async=false";
|
|
|
|
LOG_INFO("[PIPELINE] " + pipeline_str);
|
|
|
|
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;
|
|
}
|
|
|
|
// ========== 主推流循环 ==========
|
|
void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
|
|
{
|
|
const std::string key = make_key(cam.name);
|
|
|
|
constexpr int64_t START_TIMEOUT_MS = 12000; // 启动宽限
|
|
constexpr int64_t NO_FRAME_TIMEOUT_MS = 15000; // 运行期宽限
|
|
|
|
while (ctx->thread_running)
|
|
{
|
|
if (!device_exists(cam.device))
|
|
{
|
|
LOG_WARN("[RTMP] " + key + " - device not found");
|
|
std::this_thread::sleep_for(std::chrono::seconds(3));
|
|
continue;
|
|
}
|
|
|
|
GstElement* pipeline = create_pipeline(cam);
|
|
if (!pipeline)
|
|
{
|
|
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);
|
|
|
|
ctx->last_frame_ms.store(0, std::memory_order_relaxed);
|
|
|
|
// ⭐ 帧探针挂在 h264parse src
|
|
if (GstElement* parse = gst_bin_get_by_name(GST_BIN(pipeline), "parse"))
|
|
{
|
|
if (GstPad* pad = gst_element_get_static_pad(parse, "src"))
|
|
{
|
|
gst_pad_add_probe(
|
|
pad, GST_PAD_PROBE_TYPE_BUFFER,
|
|
[](GstPad*, GstPadProbeInfo*, gpointer data) -> GstPadProbeReturn
|
|
{
|
|
auto* ts = static_cast<std::atomic<int64_t>*>(data);
|
|
auto now = std::chrono::steady_clock::now().time_since_epoch();
|
|
ts->store(std::chrono::duration_cast<std::chrono::milliseconds>(now).count(),
|
|
std::memory_order_relaxed);
|
|
return GST_PAD_PROBE_OK;
|
|
},
|
|
&ctx->last_frame_ms, nullptr);
|
|
gst_object_unref(pad);
|
|
}
|
|
gst_object_unref(parse);
|
|
}
|
|
|
|
LOG_INFO("[RTMP] Starting stream: " + key);
|
|
gst_element_set_state(pipeline, GST_STATE_PLAYING);
|
|
|
|
GstState state;
|
|
if (gst_element_get_state(pipeline, &state, nullptr, 5 * GST_SECOND) != GST_STATE_CHANGE_SUCCESS ||
|
|
state != GST_STATE_PLAYING)
|
|
{
|
|
LOG_WARN("[RTMP] " + key + " failed to enter PLAYING");
|
|
gst_element_set_state(pipeline, GST_STATE_NULL);
|
|
gst_object_unref(bus);
|
|
gst_object_unref(pipeline);
|
|
std::this_thread::sleep_for(std::chrono::seconds(3));
|
|
continue;
|
|
}
|
|
|
|
LOG_INFO("[RTMP] " + key + " confirmed PLAYING");
|
|
|
|
auto launch_tp = std::chrono::steady_clock::now();
|
|
bool entered_running = false;
|
|
bool need_restart = false;
|
|
|
|
while (ctx->thread_running)
|
|
{
|
|
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 last_ms = ctx->last_frame_ms.load(std::memory_order_relaxed);
|
|
|
|
if (!entered_running)
|
|
{
|
|
if (last_ms != 0)
|
|
{
|
|
entered_running = true;
|
|
LOG_INFO("[RTMP] " + key + " entered RUNNING state");
|
|
}
|
|
else if (std::chrono::duration_cast<std::chrono::milliseconds>(now - launch_tp).count() >
|
|
START_TIMEOUT_MS)
|
|
{
|
|
// ⭐ 启动阶段允许慢，不重启
|
|
LOG_WARN("[RTMP] " + key + " - no frames yet during startup (tolerated)");
|
|
}
|
|
}
|
|
else
|
|
{
|
|
if (now_ms - last_ms > NO_FRAME_TIMEOUT_MS)
|
|
{
|
|
LOG_ERROR("[RTMP] " + key + " - frame stalled");
|
|
need_restart = true;
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (GstMessage* msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND,
|
|
(GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS)))
|
|
{
|
|
need_restart = true;
|
|
gst_message_unref(msg);
|
|
break;
|
|
}
|
|
}
|
|
|
|
gst_element_set_state(pipeline, GST_STATE_NULL);
|
|
gst_object_unref(bus);
|
|
gst_object_unref(pipeline);
|
|
|
|
if (ctx->thread_running && need_restart)
|
|
{
|
|
LOG_WARN("[RTMP] Restarting " + key + " in 3s...");
|
|
std::this_thread::sleep_for(std::chrono::seconds(3));
|
|
}
|
|
}
|
|
|
|
LOG_INFO("[RTMP] Stream thread exited for " + key);
|
|
}
|
|
|
|
// ========== 启停 ==========
|
|
void RTMPManager::start_all()
|
|
{
|
|
std::lock_guard<std::mutex> lock(streams_mutex);
|
|
|
|
int delay_ms = 0;
|
|
for (const auto& cam : g_app_config.cameras)
|
|
{
|
|
if (!cam.enabled) continue;
|
|
|
|
auto key = make_key(cam.name);
|
|
auto ctx = std::make_unique<StreamContext>();
|
|
ctx->thread_running.store(true);
|
|
|
|
ctx->thread = std::thread(
|
|
[cam, ptr = ctx.get(), delay_ms]()
|
|
{
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
|
|
stream_loop(cam, ptr);
|
|
});
|
|
|
|
streams.emplace(key, std::move(ctx));
|
|
delay_ms += 200;
|
|
}
|
|
}
|
|
|
|
void RTMPManager::stop_all()
|
|
{
|
|
std::lock_guard<std::mutex> lock(streams_mutex);
|
|
for (auto& kv : streams) kv.second->thread_running.store(false);
|
|
for (auto& kv : streams)
|
|
if (kv.second->thread.joinable()) kv.second->thread.join();
|
|
streams.clear();
|
|
}
|
|
|
|
std::string RTMPManager::get_stream_url(const std::string& cam_name)
|
|
{
|
|
std::string ip = get_ip_address("enP2p33s0");
|
|
if (ip.empty()) ip = "127.0.0.1";
|
|
return "rtsp://" + ip + ":18554/" + cam_name;
|
|
}
|