temp

include/rtmp_manager.hpp

@@ -8,6 +8,7 @@
 #include <thread>
 #include <atomic>
 #include <functional>
+#include <memory>
 #include "app_config.hpp"
 
 class RTMPManager
@@ -47,9 +48,15 @@ private:
         std::atomic<bool> running{false};
         std::thread thread;
         StreamStatus status;
+        StreamContext() = default;
+        // non-copyable
+        StreamContext(const StreamContext &) = delete;
+        StreamContext &operator=(const StreamContext &) = delete;
+        // movable? we avoid moving by using unique_ptr
     };
 
-    static std::unordered_map<std::string, StreamContext> streams;
+    // store unique_ptr to avoid copy/move issues with atomic/thread
+    static std::unordered_map<std::string, std::unique_ptr<StreamContext>> streams;
     static std::mutex streams_mutex;
     static StreamCallback status_callback;
 

src/rtmp_manager.cpp

@@ -5,8 +5,10 @@
 #include <thread>
 #include <iostream>
 #include <algorithm>
+#include <utility> // for move
+#include <memory>
 
-std::unordered_map<std::string, RTMPManager::StreamContext> RTMPManager::streams;
+std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
 std::mutex RTMPManager::streams_mutex;
 RTMPManager::StreamCallback RTMPManager::status_callback = nullptr;
 
@@ -32,11 +34,13 @@ void RTMPManager::set_status_callback(StreamCallback 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())
-        it->second.status = status;
+            it->second->status = status;
+    }
     if (status_callback)
         status_callback(key, status);
 }
@@ -95,7 +99,8 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
     {
         {
             std::lock_guard<std::mutex> lock(streams_mutex);
-            if (!streams[key].running)
+            auto it = streams.find(key);
+            if (it == streams.end() || !it->second->running.load())
                 break;
         }
 
@@ -125,7 +130,8 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
 
             {
                 std::lock_guard<std::mutex> lock(streams_mutex);
-                if (!streams[key].running)
+                auto it = streams.find(key);
+                if (it == streams.end() || !it->second->running.load())
                 {
                     stop_flag = true;
                     break;
@@ -143,7 +149,9 @@ 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});
+                if (err)
                     g_error_free(err);
+                if (debug)
                     g_free(debug);
                 stop_flag = true;
             }
@@ -160,6 +168,7 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
         }
 
         gst_element_set_state(pipeline, GST_STATE_NULL);
+        if (bus)
             gst_object_unref(bus);
         gst_object_unref(pipeline);
 
@@ -183,26 +192,47 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
 void RTMPManager::start_camera(const Camera &cam, StreamType type)
 {
     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)
+        if (it != streams.end() && it->second->running.load())
         {
             LOG_WARN("[RTMP] " + key + " already streaming.");
             return;
         }
 
-    StreamContext ctx;
+        // create a new context on heap and keep a unique_ptr in map
-    ctx.running = true;
+        auto ctx = std::make_unique<StreamContext>();
-    ctx.thread = std::thread([cam, type]()
+        ctx->running.store(true);
-                             { stream_loop(cam, type); });
+        ctx->status = {true, StreamResult::OK, ""};
+
+        // start thread after moving unique_ptr into map (to avoid racing)
+        // insert placeholder first to reserve key, then start thread
         streams.emplace(key, std::move(ctx));
     }
 
+    // actually start thread outside the lock (we need to get pointer again)
+    {
+        std::lock_guard<std::mutex> lock(streams_mutex);
+        auto it = streams.find(key);
+        if (it != streams.end())
+        {
+            // start the thread that runs stream_loop; capture cam and type by value
+            it->second->thread = std::thread([cam, type]()
+                                             { stream_loop(cam, type); });
+        }
+    }
+
+    LOG_INFO("[RTMP] start_camera requested for " + key);
+}
+
 void RTMPManager::stop_camera(const std::string &cam_name, StreamType type)
 {
     std::string key = make_stream_key(cam_name, type);
-    std::unique_lock<std::mutex> lock(streams_mutex);
+
+    std::unique_ptr<StreamContext> taken_ctx;
+    {
+        std::lock_guard<std::mutex> lock(streams_mutex);
         auto it = streams.find(key);
         if (it == streams.end())
         {
@@ -210,32 +240,50 @@ void RTMPManager::stop_camera(const std::string &cam_name, StreamType type)
             return;
         }
 
-    it->second.running = false;
+        // mark it to stop and take ownership
-    auto th = std::move(it->second.thread);
+        it->second->running.store(false);
+        taken_ctx = std::move(it->second);
         streams.erase(it);
-    lock.unlock();
+    }
 
-    if (th.joinable())
+    // join outside lock
-        th.join();
+    if (taken_ctx && taken_ctx->thread.joinable())
+        taken_ctx->thread.join();
 
     update_status(key, {false, StreamResult::OK, "Stopped manually"});
+    LOG_INFO("[RTMP] stop_camera finished for " + key);
 }
 
 void RTMPManager::stop_all()
 {
+    std::vector<std::unique_ptr<StreamContext>> taken;
     std::vector<std::string> keys;
+
     {
         std::lock_guard<std::mutex> lock(streams_mutex);
         for (auto &kv : streams)
-            keys.push_back(kv.first);
-    }
-    for (auto &key : keys)
         {
-        if (key.find("_main") != std::string::npos)
+            keys.push_back(kv.first);
-            stop_camera(key.substr(0, key.size() - 5), StreamType::MAIN);
+            kv.second->running.store(false);
-        else
-            stop_camera(key.substr(0, key.size() - 4), StreamType::SUB);
         }
+        // move all contexts out
+        for (auto &k : keys)
+        {
+            auto it = streams.find(k);
+            if (it != streams.end())
+                taken.push_back(std::move(it->second));
+            streams.erase(k);
+        }
+    }
+
+    // join threads
+    for (size_t i = 0; i < taken.size(); ++i)
+    {
+        if (taken[i] && taken[i]->thread.joinable())
+            taken[i]->thread.join();
+    }
+
+    LOG_INFO("[RTMP] stop_all completed.");
 }
 
 bool RTMPManager::is_streaming(const std::string &cam_name, StreamType type)
@@ -243,14 +291,14 @@ bool RTMPManager::is_streaming(const std::string &cam_name, StreamType type)
     std::lock_guard<std::mutex> lock(streams_mutex);
     auto key = make_stream_key(cam_name, type);
     auto it = streams.find(key);
-    return it != streams.end() && it->second.running;
+    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)
+        if (kv.second->running.load())
             return true;
     return false;
 }