diff --git a/include/rtmp_manager.hpp b/include/rtmp_manager.hpp index 69137b7..572fc93 100644 --- a/include/rtmp_manager.hpp +++ b/include/rtmp_manager.hpp @@ -11,6 +11,7 @@ #include #include #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"; diff --git a/src/rtmp_manager.cpp b/src/rtmp_manager.cpp index b99894c..ca28ad3 100644 --- a/src/rtmp_manager.cpp +++ b/src/rtmp_manager.cpp @@ -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 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(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,28 +159,33 @@ 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; + } } gst_message_unref(msg); @@ -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)); }