temp

src/rtmp_manager.cpp

@@ -82,24 +82,11 @@ 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);
 
-    GstElement *pipeline = create_pipeline(cam, type);
+    const int MAX_RETRIES = 5;
-    if (!pipeline)
+    int retry_count = 0;
-    {
-        update_status(key, {false, StreamResult::PIPELINE_ERROR, "Failed to create pipeline"});
-        LOG_ERROR("[RTMP] " + key + " pipeline creation failed. Exiting stream loop.");
-        return;
-    }
-
-    GstBus *bus = gst_element_get_bus(pipeline);
-    gst_element_set_state(pipeline, GST_STATE_PLAYING);
-    update_status(key, {true, StreamResult::OK, ""});
-    LOG_INFO("[RTMP] " + key + " is streaming.");
 
     while (true)
     {
-        GstMessage *msg = gst_bus_timed_pop_filtered(bus, GST_CLOCK_TIME_NONE,
-                                                     static_cast<GstMessageType>(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
-
         {
             std::lock_guard<std::mutex> lock(streams_mutex);
             auto it = streams.find(key);
@@ -107,42 +94,88 @@ void RTMPManager::stream_loop(Camera cam, StreamType type)
                 break;
         }
 
-        if (!msg)
+        GstElement *pipeline = create_pipeline(cam, type);
+        if (!pipeline)
+        {
+            update_status(key, {false, StreamResult::PIPELINE_ERROR, "Failed to create pipeline"});
+            if (++retry_count > MAX_RETRIES)
+                break;
+            std::this_thread::sleep_for(std::chrono::seconds(1));
             continue;
-
-        if (GST_MESSAGE_TYPE(msg) == 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});
-            if (err)
-                g_error_free(err);
-            if (debug)
-                g_free(debug);
-            gst_message_unref(msg);
-            break; // 出错立即退出
-        }
-        else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_EOS)
-        {
-            LOG_WARN("[RTMP] EOS on " + key);
-            update_status(key, {false, StreamResult::EOS_RECEIVED, "EOS"});
-            gst_message_unref(msg);
-            break; // EOS 退出
         }
 
-        gst_message_unref(msg);
+        GstBus *bus = gst_element_get_bus(pipeline);
+        gst_element_set_state(pipeline, GST_STATE_PLAYING);
+        update_status(key, {true, StreamResult::OK, ""});
+        LOG_INFO("[RTMP] " + key + " is streaming.");
+
+        bool stop_flag = false;
+        GstMessage *msg = nullptr;
+
+        while (!stop_flag)
+        {
+            // 等待消息，但设置 100ms 超时
+            msg = gst_bus_timed_pop_filtered(
+                bus, 100 * GST_MSECOND,
+                static_cast<GstMessageType>(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
+
+            {
+                std::lock_guard<std::mutex> lock(streams_mutex);
+                auto it = streams.find(key);
+                if (it == streams.end() || !it->second->running.load())
+                {
+                    stop_flag = true;
+                    break;
+                }
+            }
+
+            if (!msg)
+                continue;
+
+            if (GST_MESSAGE_TYPE(msg) == 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});
+                if (err)
+                    g_error_free(err);
+                if (debug)
+                    g_free(debug);
+                stop_flag = true;
+            }
+            else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_EOS)
+            {
+                LOG_WARN("[RTMP] EOS on " + key);
+                update_status(key, {false, StreamResult::EOS_RECEIVED, "EOS"});
+                stop_flag = true;
+            }
+
+            gst_message_unref(msg);
+        }
+
+        gst_element_set_state(pipeline, GST_STATE_NULL);
+        if (bus)
+            gst_object_unref(bus);
+        gst_object_unref(pipeline);
+
+        if (!stop_flag)
+            break;
+
+        if (++retry_count > MAX_RETRIES)
+        {
+            LOG_ERROR("[RTMP] " + key + " reached max retries. Giving up.");
+            break;
+        }
+
+        LOG_WARN("[RTMP] Reconnecting " + key + " in 1s...");
+        std::this_thread::sleep_for(std::chrono::seconds(1));
     }
 
-    gst_element_set_state(pipeline, GST_STATE_NULL);
-    if (bus)
-        gst_object_unref(bus);
-    gst_object_unref(pipeline);
-
-    LOG_INFO("[RTMP] Stream loop ended for " + key);
     update_status(key, {false, StreamResult::UNKNOWN, "Stream loop exited"});
+    LOG_INFO("[RTMP] Stream loop ended for " + key);
 }
 
 void RTMPManager::start_camera(const Camera &cam, StreamType type)