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-16 12:53:41 +08:00
parent 4954ab200e
commit 93c71e1193
2 changed files with 70 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
include/rtmp_manager.hpp
View File

@ -66,8 +66,12 @@ class RTMPManager
{
std::atomic<bool> running{false};
std::thread thread;
// 新增：把启动阶段的 promise 放到上下文里，避免悬垂引用
std::promise<StreamStatus> start_promise;
std::atomic<bool> start_promise_set{false};
StreamStatus status;
int retry_count{0};
StreamContext() = default;
StreamContext(const StreamContext &) = delete;
@ -76,8 +80,7 @@ class RTMPManager
static std::string make_stream_key(const std::string &cam_name, StreamType type);
static GstElement *create_pipeline(const Camera &cam, StreamType type);
static void stream_loop(Camera cam, StreamType type, StreamContext *ctx,
std::promise<StreamStatus> *status_promise);
static void stream_loop(Camera cam, StreamType type, StreamContext *ctx);
static void update_status(const std::string &key, const StreamStatus &status);

166
src/rtmp_manager.cpp
View File

@ -85,11 +85,10 @@ RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, Strea
res.loc = get_camera_index(cam.name);
res.url = get_stream_url(cam.name, type);
std::string key = make_stream_key(cam.name, type);
std::promise<StreamStatus> status_promise;
auto status_future = status_promise.get_future();
const std::string key = make_stream_key(cam.name, type);
std::unique_ptr<StreamContext> ctx;
std::future<StreamStatus> status_future;
{
std::lock_guard<std::mutex> lock(streams_mutex);
@ -104,20 +103,21 @@ RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, Strea
ctx = std::make_unique<StreamContext>();
ctx->running.store(true);
// 传入 StreamContext 指针
// 重要：在把 ctx 放入 map 之前，先从 ctx->start_promise 拿到 future
status_future = ctx->start_promise.get_future();
StreamContext *ctx_ptr = ctx.get();
ctx->thread = std::thread([cam, type, ctx_ptr, &status_promise]()
{ RTMPManager::stream_loop(cam, type, ctx_ptr, &status_promise); });
ctx->thread = std::thread([cam, type, ctx_ptr]() { RTMPManager::stream_loop(cam, type, ctx_ptr); });
streams.emplace(key, std::move(ctx));
}
// 等待首帧确认或错误，最多等待 7 秒
// 等待首帧或错误，最多 7 秒
if (status_future.wait_for(std::chrono::seconds(7)) == std::future_status::ready)
{
StreamStatus status = status_future.get();
res.result = status.running ? 0 : 1;
res.reason = status.last_error.empty() ? "Started OK" : status.last_error;
const StreamStatus s = status_future.get();
res.result = s.running ? 0 : 1;
res.reason = s.running ? "Started OK" : (s.last_error.empty() ? "Failed to start" : s.last_error);
}
else
{
@ -158,83 +158,68 @@ RTMPManager::StreamResultInfo RTMPManager::stop_camera(const std::string &cam_na
return res;
}
void RTMPManager::stream_loop(Camera cam, StreamType type, StreamContext *ctx,
std::promise<StreamStatus> *status_promise)
void RTMPManager::stream_loop(Camera cam, StreamType type, StreamContext *ctx)
{
std::string key = make_stream_key(cam.name, type);
const std::string key = make_stream_key(cam.name, type);
auto try_set_start = [&](const StreamStatus &st)
{
bool expected = false;
if (ctx && ctx->start_promise_set.compare_exchange_strong(expected, true))
{
try
{
ctx->start_promise.set_value(st);
}
catch (...)
{
}
}
};
StreamStatus status;
status.running = false;
status.last_result = StreamResult::UNKNOWN;
status.last_error.clear();
GstElement *pipeline = create_pipeline(cam, type);
if (!pipeline)
{
status.last_result = StreamResult::PIPELINE_ERROR;
status.last_error = "Failed to create pipeline";
if (status_promise)
{
try
{
status_promise->set_value(status);
}
catch (...)
{
}
}
try_set_start(status);
return;
}
GstBus *bus = gst_element_get_bus(pipeline);
gst_element_set_state(pipeline, GST_STATE_PLAYING);
bool first_frame_received = false;
auto start_time = std::chrono::steady_clock::now();
bool first_frame = false;
const auto t0 = std::chrono::steady_clock::now();
while (true)
{
// 检查超时
if (!first_frame_received &&
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start_time).count() > 5)
// 手动停止
if (!ctx->running.load())
{
status.running = false;
status.last_result = StreamResult::UNKNOWN;
status.last_error = "Stream stopped manually";
try_set_start(status); // 若启动阶段还没返回，这里补一次
break;
}
// 首帧超时（5s）
if (!first_frame &&
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - t0).count() > 5)
{
status.running = false;
status.last_result = StreamResult::TIMEOUT;
status.last_error = "No frames received within timeout";
if (status_promise)
{
try
{
status_promise->set_value(status);
}
catch (...)
{
}
status_promise = nullptr;
}
try_set_start(status);
break;
}
// 检查 stop 请求
{
std::lock_guard<std::mutex> lock(streams_mutex);
if (!ctx->running.load())
{
status.running = false;
status.last_result = StreamResult::UNKNOWN;
status.last_error = "Stream stopped manually";
if (status_promise)
{
try
{
status_promise->set_value(status);
}
catch (...)
{
}
status_promise = nullptr;
}
break;
}
}
GstMessage *msg = gst_bus_timed_pop_filtered(
bus, 100 * GST_MSECOND,
static_cast<GstMessageType>(GST_MESSAGE_ERROR | GST_MESSAGE_EOS | GST_MESSAGE_STATE_CHANGED));
@ -247,28 +232,16 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, StreamContext *ctx,
{
GstState old_state, new_state;
gst_message_parse_state_changed(msg, &old_state, &new_state, nullptr);
if (GST_MESSAGE_SRC(msg) == GST_OBJECT(pipeline) && new_state == GST_STATE_PLAYING &&
!first_frame_received)
if (GST_MESSAGE_SRC(msg) == GST_OBJECT(pipeline) && new_state == GST_STATE_PLAYING && !first_frame)
{
first_frame_received = true;
first_frame = true;
status.running = true;
status.last_result = StreamResult::OK;
status.last_error = "";
if (status_promise)
{
try
{
status_promise->set_value(status);
}
catch (...)
{
}
status_promise = nullptr;
}
status.last_error.clear();
try_set_start(status); // 首帧成功：通知 start_camera
}
break;
}
case GST_MESSAGE_ERROR:
{
GError *err = nullptr;
@ -277,46 +250,35 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, StreamContext *ctx,
status.last_result = StreamResult::CONNECTION_FAIL;
status.last_error = err ? err->message : "GStreamer error";
if (err) g_error_free(err);
if (status_promise)
{
try
{
status_promise->set_value(status);
}
catch (...)
{
}
status_promise = nullptr;
}
try_set_start(status);
break;
}
case GST_MESSAGE_EOS:
{
status.running = false;
status.last_result = StreamResult::EOS_RECEIVED;
status.last_error = "EOS received";
if (status_promise)
{
try
{
status_promise->set_value(status);
}
catch (...)
{
}
status_promise = nullptr;
}
try_set_start(status);
break;
}
default:
break;
}
gst_message_unref(msg);
// 错误/EOS 后退出循环
if (status.last_result == StreamResult::CONNECTION_FAIL || status.last_result == StreamResult::EOS_RECEIVED)
{
break;
}
}
gst_element_set_state(pipeline, GST_STATE_NULL);
gst_object_unref(bus);
if (bus) gst_object_unref(bus);
gst_object_unref(pipeline);
// 线程收尾：这里不要删 map 里的 ctx，stop_camera 已经负责转移并 join
}
void RTMPManager::stop_all()