1
This commit is contained in:
parent
f2c8cbfd9a
commit
23718bb584
@ -109,7 +109,9 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
|
||||
|
||||
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;
|
||||
@ -118,101 +120,76 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
|
||||
continue;
|
||||
}
|
||||
|
||||
// 创建 pipeline
|
||||
GstElement *pipeline = create_pipeline(cam);
|
||||
if (!pipeline)
|
||||
{
|
||||
ctx->status.running = false;
|
||||
ctx->status.last_error = "Failed to create pipeline";
|
||||
ctx->status.last_error = "Pipeline creation failed";
|
||||
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
|
||||
std::this_thread::sleep_for(std::chrono::seconds(3));
|
||||
continue;
|
||||
}
|
||||
|
||||
GstBus *bus = gst_element_get_bus(pipeline);
|
||||
gst_element_set_state(pipeline, GST_STATE_PLAYING);
|
||||
LOG_INFO("[RTMP] Starting stream: " + key);
|
||||
|
||||
// 等 pipeline 真正进入 PLAYING 状态
|
||||
GstStateChangeReturn ret = gst_element_get_state(pipeline, nullptr, nullptr, 3 * GST_SECOND);
|
||||
if (ret != GST_STATE_CHANGE_SUCCESS)
|
||||
{
|
||||
ctx->status.running = false;
|
||||
ctx->status.last_error = "Pipeline failed to reach PLAYING";
|
||||
LOG_ERROR("[RTMP] " + key + " failed to start (maybe no signal)");
|
||||
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;
|
||||
}
|
||||
|
||||
// 等首帧
|
||||
bool first_frame = false;
|
||||
auto start_time = std::chrono::steady_clock::now();
|
||||
while (!first_frame && std::chrono::steady_clock::now() - start_time < std::chrono::seconds(3))
|
||||
{
|
||||
GstMessage *msg = gst_bus_timed_pop_filtered(bus, 300 * GST_MSECOND,
|
||||
(GstMessageType)(GST_MESSAGE_ELEMENT | GST_MESSAGE_ERROR));
|
||||
if (!msg) continue;
|
||||
|
||||
if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ELEMENT)
|
||||
{
|
||||
const GstStructure *s = gst_message_get_structure(msg);
|
||||
if (s && g_str_has_prefix(gst_structure_get_name(s), "fpsdisplaysink"))
|
||||
{
|
||||
first_frame = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ERROR)
|
||||
{
|
||||
first_frame = false;
|
||||
break;
|
||||
}
|
||||
gst_message_unref(msg);
|
||||
}
|
||||
|
||||
if (!first_frame)
|
||||
{
|
||||
ctx->status.running = false;
|
||||
ctx->status.last_error = "No frames detected";
|
||||
LOG_ERROR("[RTMP] " + key + " - No frames detected (no video signal)");
|
||||
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;
|
||||
}
|
||||
|
||||
// 进入正常推流状态
|
||||
ctx->status.running = true;
|
||||
ctx->status.last_error.clear();
|
||||
LOG_INFO("[RTMP] Started stream for " + key);
|
||||
|
||||
bool got_frame = false;
|
||||
bool need_restart = false;
|
||||
auto start_time = std::chrono::steady_clock::now();
|
||||
|
||||
while (ctx->running)
|
||||
{
|
||||
GstMessage *msg = gst_bus_timed_pop_filtered(bus, 500 * GST_MSECOND,
|
||||
(GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
|
||||
GstMessage *msg = gst_bus_timed_pop_filtered(
|
||||
bus, 200 * GST_MSECOND, (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS | GST_MESSAGE_ELEMENT));
|
||||
|
||||
// 检查帧超时：3 秒内没有检测到帧
|
||||
auto elapsed =
|
||||
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start_time).count();
|
||||
if (!got_frame && elapsed > 3)
|
||||
{
|
||||
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;
|
||||
}
|
||||
|
||||
if (!msg) continue;
|
||||
|
||||
switch (GST_MESSAGE_TYPE(msg))
|
||||
{
|
||||
case GST_MESSAGE_ELEMENT:
|
||||
// 检测 fpsdisplaysink 输出帧
|
||||
if (gst_message_has_name(msg, "fpsprobe"))
|
||||
{
|
||||
got_frame = true;
|
||||
}
|
||||
break;
|
||||
|
||||
case GST_MESSAGE_ERROR:
|
||||
{
|
||||
GError *err = nullptr;
|
||||
gst_message_parse_error(msg, &err, nullptr);
|
||||
ctx->status.running = false;
|
||||
ctx->status.last_error = err ? err->message : "Unknown GStreamer error";
|
||||
ctx->status.last_error = err ? std::string(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:
|
||||
ctx->status.running = false;
|
||||
ctx->status.last_error = "End of stream";
|
||||
ctx->status.last_error = "End of stream (EOS)";
|
||||
LOG_WARN("[RTMP] " + key + " reached EOS");
|
||||
need_restart = true;
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
}
|
||||
@ -221,8 +198,9 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
|
||||
if (need_restart) break;
|
||||
}
|
||||
|
||||
// 清理资源
|
||||
gst_element_set_state(pipeline, GST_STATE_NULL);
|
||||
gst_object_unref(bus);
|
||||
if (bus) gst_object_unref(bus);
|
||||
gst_object_unref(pipeline);
|
||||
|
||||
if (ctx->running)
|
||||
@ -233,6 +211,7 @@ void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
|
||||
}
|
||||
|
||||
ctx->status.running = false;
|
||||
LOG_INFO("[RTMP] Stream thread exited for " + key);
|
||||
}
|
||||
|
||||
// ========== 启停与状态 ==========
|
||||
@ -251,20 +230,34 @@ void RTMPManager::start_all()
|
||||
}
|
||||
}
|
||||
|
||||
void RTMPManager::stop_all()
|
||||
void RTMPManager::start_all()
|
||||
{
|
||||
LOG_INFO("[RTMP] Starting all record streams...");
|
||||
std::lock_guard<std::mutex> lock(streams_mutex);
|
||||
for (auto &kv : streams) kv.second->running.store(false);
|
||||
for (auto &kv : streams)
|
||||
if (kv.second->thread.joinable()) kv.second->thread.join();
|
||||
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));
|
||||
return (it != streams.end() && it->second->status.running);
|
||||
int delay_ms = 0;
|
||||
for (auto &cam : g_app_config.cameras)
|
||||
{
|
||||
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>();
|
||||
ctx->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; // 每路错开 200ms
|
||||
}
|
||||
}
|
||||
|
||||
std::string RTMPManager::get_stream_url(const std::string &cam_name)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user