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 11:27:53 +08:00
parent a955867531
commit 87c370ae42
2 changed files with 43 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/rtmp_manager.hpp
View File

@ -11,6 +11,7 @@
#include <memory>
#include <algorithm>
#include "app_config.hpp"
#include "mqtt_client_wrapper.hpp"
class RTMPManager
{
@ -61,6 +62,7 @@ private:
static GstElement *create_pipeline(const Camera &cam, StreamType type);
static std::string make_stream_key(const std::string &cam_name, StreamType type);
static void update_status(const std::string &key, const StreamStatus &status);
static void RTMPManager::publish_stream_status(const Camera &cam, StreamType type, const std::string &seqNo, bool ok, const std::string &reason);
static inline std::string stream_type_suffix(StreamType type)
{
return (type == StreamType::MAIN) ? "_main" : "_sub";

58
src/rtmp_manager.cpp
View File

@ -77,16 +77,33 @@ GstElement *RTMPManager::create_pipeline(const Camera &cam, StreamType type)
return pipeline;
}
void RTMPManager::publish_stream_status(const Camera &cam, StreamType type, const std::string &seqNo, bool ok, const std::string &reason)
{
nlohmann::json ch_resp;
ch_resp["loc"] = cam.index;
ch_resp["url"] = ok ? get_stream_url(cam.name, type) : "";
ch_resp["result"] = ok ? 0 : 1;
ch_resp["reason"] = reason;
nlohmann::json reply;
reply["type"] = "response";
reply["seqNo"] = seqNo;
reply["data"] = nlohmann::json::array({ch_resp});
mqtt_client->publish(g_app_config.mqtt.topics.video_down, reply.dump());
}
void RTMPManager::stream_loop(Camera cam, StreamType type)
{
std::string key = make_stream_key(cam.name, type);
LOG_INFO("[RTMP] Stream loop started for " + key);
const int MAX_RETRIES = 5;
const int MAX_RETRIES = 3;
int retry_count = 0;
while (true)
{
// 检查是否被 stop
{
std::lock_guard<std::mutex> lock(streams_mutex);
auto it = streams.find(key);
@ -97,7 +114,9 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
GstElement *pipeline = create_pipeline(cam, type);
if (!pipeline)
{
update_status(key, {false, StreamResult::PIPELINE_ERROR, "Failed to create pipeline"});
std::string reason = "Failed to create pipeline (device unavailable or resolution unsupported)";
update_status(key, {false, StreamResult::PIPELINE_ERROR, reason});
publish_stream_status(key, false, reason); // MQTT 上报
if (++retry_count > MAX_RETRIES)
break;
std::this_thread::sleep_for(std::chrono::seconds(1));
@ -107,14 +126,13 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
GstBus *bus = gst_element_get_bus(pipeline);
gst_element_set_state(pipeline, GST_STATE_PLAYING);
// 等待 pipeline 状态变为 PLAYING 或错误
bool first_playing = false;
bool first_frame_received = false;
bool stop_flag = false;
GstMessage *msg = nullptr;
// 等待消息，同时监控是否有第一帧
while (!stop_flag)
{
msg = gst_bus_timed_pop_filtered(
GstMessage *msg = gst_bus_timed_pop_filtered(
bus, 100 * GST_MSECOND,
static_cast<GstMessageType>(GST_MESSAGE_ERROR | GST_MESSAGE_EOS | GST_MESSAGE_STATE_CHANGED));
@ -131,7 +149,9 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
if (!msg)
continue;
if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ERROR)
switch (GST_MESSAGE_TYPE(msg))
{
case GST_MESSAGE_ERROR:
{
GError *err = nullptr;
gchar *debug = nullptr;
@ -139,27 +159,32 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
std::string err_msg = err ? err->message : "Unknown GStreamer error";
LOG_ERROR("[RTMP] Error in " + key + ": " + err_msg);
update_status(key, {false, StreamResult::CONNECTION_FAIL, err_msg});
publish_stream_status(key, false, err_msg); // MQTT 上报
if (err)
g_error_free(err);
if (debug)
g_free(debug);
stop_flag = true;
break;
}
else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_EOS)
{
case GST_MESSAGE_EOS:
LOG_WARN("[RTMP] EOS on " + key);
update_status(key, {false, StreamResult::EOS_RECEIVED, "EOS"});
publish_stream_status(key, false, "EOS received"); // MQTT 上报
stop_flag = true;
}
else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_STATE_CHANGED)
break;
case GST_MESSAGE_STATE_CHANGED:
{
GstState old_state, new_state, pending;
gst_message_parse_state_changed(msg, &old_state, &new_state, &pending);
if (GST_MESSAGE_SRC(msg) == GST_OBJECT(pipeline) && new_state == GST_STATE_PLAYING && !first_playing)
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)
{
first_playing = true;
first_frame_received = true;
LOG_INFO("[RTMP] First frame received for " + key);
update_status(key, {true, StreamResult::OK, ""});
LOG_INFO("[RTMP] " + key + " is streaming.");
publish_stream_status(key, true, "Streaming OK"); // MQTT 上报
}
break;
}
}
@ -180,7 +205,6 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
break;
}
LOG_WARN("[RTMP] Reconnecting " + key + " in 1s...");
std::this_thread::sleep_for(std::chrono::seconds(1));
}