cxh/kunlang_video
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

temp

Browse Source
This commit is contained in:
cxh 2025-10-15 15:36:48 +08:00
parent 17cec83ae3
commit 90ab6e660a
2 changed files with 14 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/rtmp_manager.hpp
View File

@ -1,4 +1,4 @@
// rtsp_manager.hpp
// rtmp_manager.hpp
#pragma once
#include <gst/gst.h>
@ -73,9 +73,9 @@ 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, std::shared_ptr<StreamContext> ctx);
static void stream_loop(Camera cam, StreamType type, StreamContext *ctx);
static std::unordered_map<std::string, std::unique_ptr<StreamContext>> streams;
static std::mutex streams_mutex;
static StreamCallback status_callback;
};
};

90
src/rtmp_manager.cpp
View File

@ -1,4 +1,4 @@
// rtsp_manager.cpp
// rtmp_manager.cpp
#include "rtmp_manager.hpp"
#include <chrono>
@ -86,7 +86,8 @@ RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, Strea
res.url = get_stream_url(cam.name, type);
std::string key = make_stream_key(cam.name, type);
std::shared_ptr<StreamContext> ctx = std::make_shared<StreamContext>();
auto ctx = std::make_unique<StreamContext>();
StreamContext *ctx_ptr = ctx.get();
{
std::lock_guard<std::mutex> lock(streams_mutex);
@ -99,18 +100,13 @@ RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, Strea
}
ctx->running.store(true);
streams[key] = ctx;
streams[key] = std::move(ctx);
}
std::future<StreamResultInfo> fut = ctx->start_result.get_future();
std::future<StreamResultInfo> fut = ctx_ptr->start_result.get_future();
ctx->thread = std::thread(
[this, cam, type, ctx]()
{
stream_loop(cam, type, ctx); // stream_loop 接收 StreamContext
});
ctx_ptr->thread = std::thread([cam, type, ctx_ptr]() { RTMPManager::stream_loop(cam, type, ctx_ptr); });
// 等待 pipeline 初始化完成，最多等待 5 秒（可自定义）
if (fut.wait_for(std::chrono::seconds(5)) == std::future_status::ready)
{
res = fut.get();
@ -155,44 +151,10 @@ RTMPManager::StreamResultInfo RTMPManager::stop_camera(const std::string &cam_na
return res;
}
void RTMPManager::stop_all()
void RTMPManager::stream_loop(Camera cam, StreamType type, StreamContext *ctx)
{
std::vector<std::unique_ptr<StreamContext>> ctxs;
{
std::lock_guard<std::mutex> lock(streams_mutex);
for (auto &kv : streams) kv.second->running.store(false);
for (auto &kv : streams) ctxs.push_back(std::move(kv.second));
streams.clear();
}
constexpr int FIRST_FRAME_TIMEOUT_SEC = 5;
for (auto &ctx : ctxs)
if (ctx->thread.joinable()) ctx->thread.join();
LOG_INFO("[RTMP] stop_all completed.");
}
bool RTMPManager::is_streaming(const std::string &cam_name, StreamType type)
{
std::lock_guard<std::mutex> lock(streams_mutex);
auto it = streams.find(make_stream_key(cam_name, type));
return it != streams.end() && it->second->running.load();
}
bool RTMPManager::is_any_streaming()
{
std::lock_guard<std::mutex> lock(streams_mutex);
for (auto &kv : streams)
if (kv.second->running.load()) return true;
return false;
}
std::string RTMPManager::get_stream_url(const std::string &cam_name, StreamType type)
{
return "rtmp://127.0.0.1/live/" + make_stream_key(cam_name, type);
}
void RTMPManager::stream_loop(Camera cam, StreamType type, std::shared_ptr<StreamContext> ctx)
{
StreamResultInfo res;
res.loc = get_camera_index(cam.name);
res.url = get_stream_url(cam.name, type);
@ -217,7 +179,6 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, std::shared_ptr<Strea
while (true)
{
// 检查 stop_flag
{
std::lock_guard<std::mutex> lock(streams_mutex);
auto it = streams.find(key);
@ -237,7 +198,7 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, std::shared_ptr<Strea
{
res.result = 1;
res.reason = "No frames received within timeout";
ctx->start_result.set_value(res); // 返回失败
ctx->start_result.set_value(res);
break;
}
}
@ -265,24 +226,15 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, std::shared_ptr<Strea
ctx->start_result.set_value(res);
break;
case GST_MESSAGE_STATE_CHANGED:
{
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)
{
// 这里仅表示 pipeline 播放了，不算真正成功
}
// pipeline 播放状态变化，不算成功
break;
}
case GST_MESSAGE_ELEMENT:
// 这里可以通过 caps 或其他方式检测到第一帧到达
if (!first_frame_received)
{
first_frame_received = true;
res.result = 0;
res.reason = "First frame received, started OK";
ctx->start_result.set_value(res); // 返回成功
ctx->start_result.set_value(res);
}
break;
default:
@ -304,23 +256,3 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, std::shared_ptr<Strea
LOG_INFO("[RTMP] Stream loop ended for " + key);
}
{
std::lock_guard<std::mutex> lock(streams_mutex);
auto it = streams.find(key);
if (it != streams.end())
{
try
{
if (it->second && it->second->thread.get_id() == std::this_thread::get_id()) streams.erase(it);
}
catch (...)
{
streams.erase(it);
}
}
}
update_status(key, {false, StreamResult::UNKNOWN, "Stream loop exited"});
LOG_INFO("[RTMP] Stream loop ended for " + key);
}