temp

.clang-format

@@ -0,0 +1,4 @@
+BasedOnStyle: Google
+IndentWidth: 4
+ColumnLimit: 120
+BreakBeforeBraces: Allman

include/rtmp_manager.hpp

@@ -22,8 +22,8 @@ public:
         PIPELINE_ERROR,
         CONNECTION_FAIL,
         EOS_RECEIVED,
-        UNKNOWN,
+        TIMEOUT,
-        TIMEOUT
+        UNKNOWN
     };
 
     struct StreamStatus
@@ -55,17 +55,19 @@ private:
         StreamContext &operator=(const StreamContext &) = delete;
     };
 
+    // helper
+    static std::string make_stream_key(const std::string &cam_name, StreamType type);
+    static GstElement *create_pipeline(const Camera &cam, StreamType type);
+    static void update_status(const std::string &key, const StreamStatus &status);
+
+    // publish a single-camera response on mqtt (uses global mqtt_client)
+    static void publish_stream_status(const std::string &cam_name, StreamType type, const std::string &seqNo, bool ok, const std::string &reason);
+
+    // actual loop (internal) - note it takes seqNo so it can report back
+    static void stream_loop(Camera cam, StreamType type, const std::string &seqNo);
+
+    // storage
     static std::unordered_map<std::string, std::unique_ptr<StreamContext>> streams;
     static std::mutex streams_mutex;
     static StreamCallback status_callback;
-
-    static void stream_loop(Camera cam, StreamType type, const std::string &seqNo);
-    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 publish_stream_status(const std::string &cam_name, 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";
-    }
 };

src/rtmp_manager.cpp

@@ -1,11 +1,16 @@
 // rtsp_manager.cpp
 #include "rtmp_manager.hpp"
-#include "logger.hpp"
+
 #include <chrono>
-#include <thread>
 #include <iostream>
+#include <thread>
 #include <utility>
 
+#include "logger.hpp"
+
+static inline std::string stream_type_suffix(StreamType type) { return (type == StreamType::MAIN) ? "_main" : "_sub"; }
+
+// static members
 std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
 std::mutex RTMPManager::streams_mutex;
 RTMPManager::StreamCallback RTMPManager::status_callback = nullptr;
@@ -21,21 +26,16 @@ void RTMPManager::init()
     LOG_INFO("[RTMP] GStreamer initialized.");
 }
 
-void RTMPManager::set_status_callback(StreamCallback cb)
+void RTMPManager::set_status_callback(StreamCallback cb) { status_callback = std::move(cb); }
-{
-    status_callback = std::move(cb);
-}
 
 void RTMPManager::update_status(const std::string &key, const StreamStatus &status)
 {
     {
         std::lock_guard<std::mutex> lock(streams_mutex);
         auto it = streams.find(key);
-        if (it != streams.end())
+        if (it != streams.end()) it->second->status = status;
-            it->second->status = status;
     }
-    if (status_callback)
+    if (status_callback) status_callback(key, status);
-        status_callback(key, status);
 }
 
 GstElement *RTMPManager::create_pipeline(const Camera &cam, StreamType type)
@@ -51,18 +51,15 @@ GstElement *RTMPManager::create_pipeline(const Camera &cam, StreamType type)
     }
 
     std::string stream_name = cam.name + stream_type_suffix(type);
-    std::string pipeline_str =
+    std::string pipeline_str = "v4l2src device=" + cam.device +
-        "v4l2src device=" + cam.device +
+                               " ! video/x-raw,format=NV12,width=" + std::to_string(width) +
-        " ! video/x-raw,format=NV12,width=" + std::to_string(width) +
+                               ",height=" + std::to_string(height) + ",framerate=" + std::to_string(fps) +
-        ",height=" + std::to_string(height) +
+                               "/1 ! queue max-size-buffers=1 leaky=downstream "
-        ",framerate=" + std::to_string(fps) +
+                               "! mpph264enc bps=" +
-        "/1 ! queue max-size-buffers=1 leaky=downstream "
+                               std::to_string(bitrate) + " gop=" + std::to_string(fps) +
-        "! mpph264enc bps=" +
+                               " ! h264parse ! flvmux streamable=true name=mux "
-        std::to_string(bitrate) +
+                               "! rtmpsink location=\"rtmp://127.0.0.1/live/" +
-        " gop=" + std::to_string(fps) +
+                               stream_name + " live=1\" sync=false";
-        " ! h264parse ! flvmux streamable=true name=mux "
-        "! rtmpsink location=\"rtmp://127.0.0.1/live/" +
-        stream_name + " live=1\" sync=false";
 
     LOG_INFO("[RTMP] Creating pipeline for '" + stream_name + "': " + pipeline_str);
 
@@ -77,7 +74,8 @@ GstElement *RTMPManager::create_pipeline(const Camera &cam, StreamType type)
     return pipeline;
 }
 
-void RTMPManager::publish_stream_status(const std::string &cam_name, StreamType type, const std::string &seqNo, bool ok, const std::string &reason)
+void RTMPManager::publish_stream_status(const std::string &cam_name, StreamType type, const std::string &seqNo, bool ok,
+                                        const std::string &reason)
 {
     // 找 camera index
     int cam_index = -1;
@@ -94,7 +92,7 @@ void RTMPManager::publish_stream_status(const std::string &cam_name, StreamType
 
     nlohmann::json ch_resp;
     ch_resp["loc"] = cam_index;
-    ch_resp["url"] = ok && cam_ptr ? get_stream_url(cam_ptr->name, type) : "";
+    ch_resp["url"] = (ok && cam_ptr) ? get_stream_url(cam_ptr->name, type) : "";
     ch_resp["result"] = ok ? 0 : 1;
     ch_resp["reason"] = reason;
 
@@ -103,10 +101,9 @@ void RTMPManager::publish_stream_status(const std::string &cam_name, StreamType
     reply["seqNo"] = seqNo;
     reply["data"] = nlohmann::json::array({ch_resp});
 
-    mqtt_client->publish(g_app_config.mqtt.topics.video_down, reply.dump());
+    if (mqtt_client) mqtt_client->publish(g_app_config.mqtt.topics.video_down, reply.dump());
 }
 
-// 修改 stream_loop，新增 seqNo 参数
 void RTMPManager::stream_loop(Camera cam, StreamType type, const std::string &seqNo)
 {
     std::string key = make_stream_key(cam.name, type);
@@ -115,13 +112,14 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, const std::string &se
     const int MAX_RETRIES = 0;
     int retry_count = 0;
 
+    const int FIRST_FRAME_TIMEOUT_SEC = 5;  // 等待第一帧最大秒数
+
     while (true)
     {
         {
             std::lock_guard<std::mutex> lock(streams_mutex);
             auto it = streams.find(key);
-            if (it == streams.end() || !it->second->running.load())
+            if (it == streams.end() || !it->second->running.load()) break;
-                break;
         }
 
         GstElement *pipeline = create_pipeline(cam, type);
@@ -130,8 +128,7 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, const std::string &se
             std::string reason = "Failed to create pipeline (device unavailable or resolution unsupported)";
             update_status(key, {false, StreamResult::PIPELINE_ERROR, reason});
             publish_stream_status(cam.name, type, seqNo, false, reason);
-            if (++retry_count > MAX_RETRIES)
+            if (++retry_count > MAX_RETRIES) break;
-                break;
             std::this_thread::sleep_for(std::chrono::seconds(1));
             continue;
         }
@@ -141,8 +138,6 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, const std::string &se
 
         bool first_frame_received = false;
         bool stop_flag = false;
-
-        const int FIRST_FRAME_TIMEOUT_SEC = 5; // 等待第一帧最大秒数
         auto first_frame_start = std::chrono::steady_clock::now();
 
         while (!stop_flag)
@@ -161,6 +156,7 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, const std::string &se
                 }
             }
 
+            // check first-frame timeout
             if (!first_frame_received)
             {
                 auto elapsed = std::chrono::steady_clock::now() - first_frame_start;
@@ -174,58 +170,56 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, const std::string &se
                 }
             }
 
-            if (!msg)
+            if (!msg) continue;
-                continue;
 
             switch (GST_MESSAGE_TYPE(msg))
             {
-            case GST_MESSAGE_ERROR:
+                case GST_MESSAGE_ERROR:
-            {
-                GError *err = nullptr;
-                gchar *debug = nullptr;
-                gst_message_parse_error(msg, &err, &debug);
-                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(cam.name, type, seqNo, false, err_msg);
-                if (err)
-                    g_error_free(err);
-                if (debug)
-                    g_free(debug);
-                stop_flag = true;
-                break;
-            }
-            case GST_MESSAGE_EOS:
-                LOG_WARN("[RTMP] EOS on " + key);
-                update_status(key, {false, StreamResult::EOS_RECEIVED, "EOS"});
-                publish_stream_status(cam.name, type, seqNo, false, "EOS received");
-                stop_flag = true;
-                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)
                 {
-                    first_frame_received = true;
+                    GError *err = nullptr;
-                    LOG_INFO("[RTMP] First frame received for " + key);
+                    gchar *debug = nullptr;
-                    update_status(key, {true, StreamResult::OK, ""});
+                    gst_message_parse_error(msg, &err, &debug);
-                    publish_stream_status(cam.name, type, seqNo, true, "Streaming OK");
+                    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(cam.name, type, seqNo, false, err_msg);
+                    if (err) g_error_free(err);
+                    if (debug) g_free(debug);
+                    stop_flag = true;
+                    break;
                 }
-                break;
+                case GST_MESSAGE_EOS:
-            }
+                    LOG_WARN("[RTMP] EOS on " + key);
+                    update_status(key, {false, StreamResult::EOS_RECEIVED, "EOS"});
+                    publish_stream_status(cam.name, type, seqNo, false, "EOS received");
+                    stop_flag = true;
+                    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)
+                    {
+                        first_frame_received = true;
+                        LOG_INFO("[RTMP] First frame received for " + key);
+                        update_status(key, {true, StreamResult::OK, ""});
+                        publish_stream_status(cam.name, type, seqNo, true, "Streaming OK");
+                    }
+                    break;
+                }
+                default:
+                    break;
             }
 
             gst_message_unref(msg);
         }
 
         gst_element_set_state(pipeline, GST_STATE_NULL);
-        if (bus)
+        if (bus) gst_object_unref(bus);
-            gst_object_unref(bus);
         gst_object_unref(pipeline);
 
-        if (!stop_flag)
+        if (!stop_flag) break;
-            break;
 
         if (++retry_count > MAX_RETRIES)
         {
@@ -236,6 +230,29 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, const std::string &se
         std::this_thread::sleep_for(std::chrono::seconds(1));
     }
 
+    // ensure we remove ourselves from streams map if still present and belong to this thread
+    {
+        std::lock_guard<std::mutex> lock(streams_mutex);
+        auto it = streams.find(key);
+        if (it != streams.end())
+        {
+            // If thread stored in map matches current thread id, erase it.
+            // This avoids racing with stop_camera which may already have moved/erased it.
+            try
+            {
+                if (it->second && it->second->thread.get_id() == std::this_thread::get_id())
+                {
+                    streams.erase(it);
+                }
+            }
+            catch (...)
+            {
+                // in case thread id comparisons throw (shouldn't), just ignore
+                streams.erase(it);
+            }
+        }
+    }
+
     update_status(key, {false, StreamResult::UNKNOWN, "Stream loop exited"});
     LOG_INFO("[RTMP] Stream loop ended for " + key);
 }
@@ -245,24 +262,27 @@ void RTMPManager::start_camera(const Camera &cam, StreamType type, const std::st
 {
     std::string key = make_stream_key(cam.name, type);
 
-    std::lock_guard<std::mutex> lock(streams_mutex);
-    auto it = streams.find(key);
-    if (it != streams.end() && it->second->running.load())
     {
-        LOG_WARN("[RTMP] " + key + " already streaming.");
+        std::lock_guard<std::mutex> lock(streams_mutex);
-        // 上报已经在请求状态
+        auto it = streams.find(key);
-        publish_stream_status(cam.name, type, seqNo, false, "Already streaming");
+        if (it != streams.end() && it->second->running.load())
-        return;
+        {
+            LOG_WARN("[RTMP] " + key + " already streaming.");
+            // 按协议：如果已经在推，直接回报已在推（成功），并返回 URL
+            publish_stream_status(cam.name, type, seqNo, true, "Already streaming");
+            return;
+        }
+
+        // create and store context
+        auto ctx = std::make_unique<StreamContext>();
+        ctx->running.store(true);
+        ctx->status = {true, StreamResult::UNKNOWN, ""};
+        // create thread and place it into context. Need to assign thread before inserting to avoid race.
+        ctx->thread = std::thread([cam, type, seqNo]() { stream_loop(cam, type, seqNo); });
+
+        streams.emplace(key, std::move(ctx));
     }
 
-    auto ctx = std::make_unique<StreamContext>();
-    ctx->running.store(true);
-    ctx->thread = std::thread([cam, type, seqNo]()
-                              {
-                                  stream_loop(cam, type, seqNo); // stream_loop 内部处理 MQTT 上报
-                              });
-
-    streams[key] = std::move(ctx);
     LOG_INFO("[RTMP] start_camera requested for " + key);
 }
 
@@ -277,8 +297,8 @@ void RTMPManager::stop_camera(const std::string &cam_name, StreamType type, cons
         if (it == streams.end())
         {
             LOG_WARN("[RTMP] " + key + " not found.");
-            // 上报停止失败（未找到）
+            // 按协议：如果没有在推，则回复“未推”的成功响应（实际不做任何动作）
-            publish_stream_status(cam_name, type, seqNo, false, "Stream not found");
+            publish_stream_status(cam_name, type, seqNo, true, "Not streaming");
             return;
         }
 
@@ -287,8 +307,7 @@ void RTMPManager::stop_camera(const std::string &cam_name, StreamType type, cons
         streams.erase(it);
     }
 
-    if (ctx->thread.joinable())
+    if (ctx->thread.joinable()) ctx->thread.join();
-        ctx->thread.join();
 
     update_status(key, {false, StreamResult::OK, "Stopped manually"});
     LOG_INFO("[RTMP] stop_camera completed: " + key);
@@ -303,16 +322,13 @@ void RTMPManager::stop_all()
 
     {
         std::lock_guard<std::mutex> lock(streams_mutex);
-        for (auto &kv : streams)
+        for (auto &kv : streams) kv.second->running.store(false);
-            kv.second->running.store(false);
+        for (auto &kv : streams) ctxs.push_back(std::move(kv.second));
-        for (auto &kv : streams)
-            ctxs.push_back(std::move(kv.second));
         streams.clear();
     }
 
     for (auto &ctx : ctxs)
-        if (ctx->thread.joinable())
+        if (ctx->thread.joinable()) ctx->thread.join();
-            ctx->thread.join();
 
     LOG_INFO("[RTMP] stop_all completed.");
 }
@@ -328,8 +344,7 @@ bool RTMPManager::is_any_streaming()
 {
     std::lock_guard<std::mutex> lock(streams_mutex);
     for (auto &kv : streams)
-        if (kv.second->running.load())
+        if (kv.second->running.load()) return true;
-            return true;
     return false;
 }