1

include/rtmp_manager.hpp

@@ -4,9 +4,6 @@
 #include <gst/gst.h>
 
 #include <atomic>
-#include <functional>
-#include <future>
-#include <memory>
 #include <mutex>
 #include <string>
 #include <thread>
@@ -14,101 +11,39 @@
 #include <vector>
 
 #include "app_config.hpp"
-#include "data_manager.hpp"
+#include "logger.hpp"
 
 class RTMPManager
 {
    public:
-    enum class StreamResult
-    {
-        OK,
-        PIPELINE_ERROR,
-        CONNECTION_FAIL,
-        EOS_RECEIVED,
-        TIMEOUT,
-        UNKNOWN
-    };
-
     struct StreamStatus
     {
         bool running{false};
-        StreamResult last_result{StreamResult::UNKNOWN};
         std::string last_error;
     };
 
-    struct StreamResultInfo
-    {
-        int loc{-1};
-        std::string url;
-        int result{1};  // 0 success, 1 fail
-        std::string reason;
-    };
-
-    using StreamCallback = std::function<void(const std::string &key, StreamStatus)>;
-
     static void init();
-    static void set_status_callback(StreamCallback cb);
+    static void start_all();
-
-    // 单路接口
-    static StreamResultInfo start_camera(const Camera &cam, StreamType type);
-    static StreamResultInfo stop_camera(const std::string &cam_name, StreamType type);
-
-    // 批量接口：处理 VideoPushRequest
-    static std::vector<StreamResultInfo> process_push_request(const VideoPushRequest &req);
-
     static void stop_all();
-    // 自动启动所有录像流（上电调用一次）
+    static bool is_streaming(const std::string &cam_name);
-    static void start_all_record_streams();
+    static std::string get_stream_url(const std::string &cam_name);
-    static bool is_streaming(const std::string &cam_name, StreamType type);
+    static std::vector<std::pair<std::string, bool>> get_all_status();
-    static bool is_any_streaming();
-    static std::string get_stream_url(const std::string &cam_name, StreamType type);
 
    private:
     struct StreamContext
     {
         std::atomic<bool> running{false};
         std::thread thread;
-
-        // 启动阶段的 promise，用于首次启动结果反馈（7s 内）
-        std::promise<StreamStatus> start_promise;
-        std::atomic<bool> start_promise_set{false};
-
         StreamStatus status;
-
-        StreamContext() = default;
-        StreamContext(const StreamContext &) = delete;
-        StreamContext &operator=(const StreamContext &) = delete;
-
-        ~StreamContext()
-        {
-            try
-            {
-                running.store(false);
-                if (thread.joinable())
-                {
-                    thread.join();  // 防止 std::terminate()
-                }
-            }
-            catch (...)
-            {
-                // 安全兜底，不抛异常
-            }
-        }
     };
 
-    static std::string make_stream_key(const std::string &cam_name, StreamType type);
+    static void stream_loop(Camera cam, StreamContext *ctx);
-    static GstElement *create_pipeline(const Camera &cam, StreamType type);
+    static GstElement *create_pipeline(const Camera &cam);
-    static void stream_loop(Camera cam, StreamType type, StreamContext *ctx);
+    static std::string make_key(const std::string &name);
-    static bool do_stream_once(const Camera &cam, StreamType type, StreamContext *ctx);
-
-    static void update_status(const std::string &key, const StreamStatus &status);
 
     static std::unordered_map<std::string, std::unique_ptr<StreamContext>> streams;
     static std::mutex streams_mutex;
-    static StreamCallback status_callback;
 
-    // 自动重试参数
+    // 参数
-    // MAIN（record）为永久重试；SUB（live）使用限次重试
+    static constexpr int RETRY_BASE_DELAY_MS = 3000;
-    static constexpr int LIVE_MAX_RETRIES = 5;
-    static constexpr int RETRY_BASE_DELAY_MS = 2000;  // 2s -> 可指数退避
 };

src/mqtt_client_wrapper.cpp

@@ -24,23 +24,36 @@ static void send_heartbeat()
     auto now = std::chrono::system_clock::now();
     auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
 
-    g_streaming.store(RTMPManager::is_any_streaming());
+    // 获取当前所有通道状态
+    auto status_list = RTMPManager::get_all_status();
+
+    int total = static_cast<int>(status_list.size());
+    int running_count = 0;
+    nlohmann::json channels = nlohmann::json::array();
+
+    for (int i = 0; i < total; ++i)
+    {
+        const auto &[key, running] = status_list[i];
+        if (running) running_count++;
+
+        nlohmann::json item;
+        item["loc"] = i;  // 摄像头位置索引
+        item["url"] = RTMPManager::get_stream_url(g_app_config.cameras[i].name);
+        item["running"] = running;
+        channels.push_back(item);
+    }
 
     nlohmann::json hb;
     hb["timestamp"] = ms;
-    hb["status"] = g_streaming ? 1 : 0;  // 0：等待中，1：推流中
+    hb["status"] = (running_count == total) ? 1 : (running_count == 0 ? 0 : 2);
+    hb["channels"] = channels;
 
     mqtt_client->publish(g_app_config.mqtt.topics.heartbeat_up, hb.dump());
+    LOG_INFO("[MQTT] Sent video heartbeat: " + hb.dump());
 }
 
 // MQTT 回调
-static void on_mqtt_connected()
+static void on_mqtt_connected() { LOG_INFO("[MQTT] Connected to broker: " + g_app_config.mqtt.server_ip); }
-{
-    LOG_INFO("[MQTT] Connected to broker: " + g_app_config.mqtt.server_ip);
-
-    const auto &topics = g_app_config.mqtt.topics;
-    mqtt_client->subscribe(topics.video_down);
-}
 
 static void on_mqtt_disconnected() { LOG_WARN("[MQTT] Disconnected from broker: " + g_app_config.mqtt.server_ip); }
 
@@ -50,74 +63,8 @@ static void on_mqtt_message_received(const std::string &topic, const std::string
     {
         auto j = nlohmann::json::parse(message);
 
-        if (topic != g_app_config.mqtt.topics.video_down || j["type"] != "request") return;
-
         LOG_INFO("[MQTT] Received message on topic [" + topic + "], len = " + std::to_string(message.size()));
         LOG_INFO("[MQTT] Message content: " + j.dump(-1));
-
-        VideoPushRequest req;
-        req.seqNo = j.value("seqNo", "");
-
-        auto data = j["data"];
-        if (data.is_object())
-        {
-            VideoPushRequest::DataItem item;
-            item.switchVal = data.value("switch", 0);
-            item.streamType = data.value("streamType", 1);  // ✅ 默认子码流
-            if (item.streamType == 0) item.streamType = 1;  // ✅ 防止平台误发 main
-            item.channels = data.value("channels", std::vector<int>{});
-            req.data.push_back(item);
-        }
-        else if (data.is_array())
-        {
-            for (auto &d : data)
-            {
-                VideoPushRequest::DataItem item;
-                item.switchVal = d.value("switch", 0);
-                item.streamType = data.value("streamType", 1);  // ✅ 默认子码流
-                if (item.streamType == 0) item.streamType = 1;  // ✅ 防止平台误发 main
-                item.channels = d.value("channels", std::vector<int>{});
-                req.data.push_back(item);
-            }
-        }
-
-        // 异步执行推流任务
-        std::thread(
-            [req]()
-            {
-                try
-                {
-                    auto results = RTMPManager::process_push_request(req);
-
-                    // 组装 MQTT 回复
-                    nlohmann::json reply;
-                    reply["type"] = "response";
-                    reply["seqNo"] = req.seqNo;
-                    reply["data"] = nlohmann::json::array();
-
-                    for (const auto &r : results)
-                    {
-                        nlohmann::json item;
-                        item["loc"] = r.loc;
-                        item["url"] = r.url;
-                        item["result"] = r.result;
-                        item["reason"] = r.reason;
-                        reply["data"].push_back(item);
-                    }
-
-                    mqtt_client->publish(g_app_config.mqtt.topics.video_down, reply.dump(), 1);
-                    LOG_INFO("[MQTT] Sent RTMP response: " + reply.dump());
-                }
-                catch (const std::exception &e)
-                {
-                    LOG_ERROR(std::string("[MQTT] Exception in RTMP thread: ") + e.what());
-                }
-                catch (...)
-                {
-                    LOG_ERROR("[MQTT] Unknown exception in RTMP thread");
-                }
-            })
-            .detach();
     }
     catch (const std::exception &e)
     {

src/rtmp_manager.cpp

@@ -6,25 +6,12 @@
 #include <net/if.h>
 #include <netinet/in.h>
 #include <sys/stat.h>
-#include <sys/types.h>
-#include <unistd.h>
 
 #include <chrono>
 #include <cstring>
-#include <iostream>
-#include <string>
 #include <thread>
 
-#include "logger.hpp"
+// 动态获取指定网卡 IPv4 地址
-
-// ---------- 小工具 ----------
-
-static bool device_exists(const std::string &path)
-{
-    struct stat st;
-    return (stat(path.c_str(), &st) == 0);
-}
-
 static std::string get_ip_address(const std::string &ifname)
 {
     struct ifaddrs *ifaddr, *ifa;
@@ -55,19 +42,13 @@ static std::string get_ip_address(const std::string &ifname)
     return "";  // 没找到
 }
 
-static inline std::string stream_type_suffix(StreamType type) { return (type == StreamType::MAIN) ? "_main" : "_sub"; }
-
-// ---------- 静态成员 ----------
-
 std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
 std::mutex RTMPManager::streams_mutex;
-RTMPManager::StreamCallback RTMPManager::status_callback = nullptr;
 
-// ---------- 内部工具 ----------
+static bool device_exists(const std::string &path)
-
-std::string RTMPManager::make_stream_key(const std::string &cam_name, StreamType type)
 {
-    return cam_name + stream_type_suffix(type);
+    struct stat st;
+    return (stat(path.c_str(), &st) == 0);
 }
 
 void RTMPManager::init()
@@ -76,258 +57,114 @@ void RTMPManager::init()
     LOG_INFO("[RTMP] GStreamer initialized.");
 }
 
-void RTMPManager::set_status_callback(StreamCallback cb) { status_callback = std::move(cb); }
+std::string RTMPManager::make_key(const std::string &name) { return name + "_main"; }
 
-void RTMPManager::update_status(const std::string &key, const StreamStatus &status)
+GstElement *RTMPManager::create_pipeline(const Camera &cam)
 {
-    {
+    int width = cam.width;
-        std::lock_guard<std::mutex> lock(streams_mutex);
+    int height = cam.height;
-        auto it = streams.find(key);
+    int fps = cam.fps;
-        if (it != streams.end()) it->second->status = status;
+    int bitrate = cam.bitrate;
-    }
-    if (status_callback) status_callback(key, status);
-}
 
-// 决定 app：MAIN -> record（录像），SUB -> live（按需）
+    std::string stream_name = cam.name + "_main";
-static inline std::string pick_app(StreamType type) { return (type == StreamType::MAIN) ? "record" : "live"; }
+    std::string app = "record";
 
-GstElement *RTMPManager::create_pipeline(const Camera &cam, StreamType type)
+    std::string location = "rtmp://127.0.0.1:1935/" + app + "/" + stream_name + "?vhost=" + app;
-{
-    int width = cam.width, height = cam.height, fps = cam.fps, bitrate = cam.bitrate;
-    if (type == StreamType::SUB)
-    {
-        bitrate = std::max(300000, bitrate / 2);
-    }
-
-    const std::string stream_name = cam.name + stream_type_suffix(type);
-    const std::string app = (type == StreamType::MAIN) ? "record" : "live";  // 🎯 主码流录像，子码流远程推流
-
-    // 👇 确保明确指定 vhost，避免 “no vhost 127.0.0.1” 错误
-    const std::string location = "rtmp://127.0.0.1:1935/" + app + "/" + stream_name + "?vhost=" + app;
 
     std::string pipeline_str = "v4l2src device=" + cam.device + " ! video/x-raw,width=" + std::to_string(width) +
                                ",height=" + std::to_string(height) + ",framerate=" + std::to_string(fps) +
-                               "/1 "
+                               "/1 ! videoconvert ! video/x-raw,format=NV12 "
-                               "! videoconvert ! video/x-raw,format=NV12 "
                                "! mpph264enc bps=" +
                                std::to_string(bitrate) + " gop=" + std::to_string(fps) +
-                               " "
+                               " ! h264parse ! flvmux name=mux streamable=true "
-                               "! h264parse "
-                               "! flvmux name=mux streamable=true "
                                "audiotestsrc wave=silence ! audioconvert ! audioresample ! voaacenc ! aacparse ! mux. "
-                               "mux. ! rtmpsink name=rtmp_sink location=\"" +
+                               "mux. ! rtmpsink location=\"" +
                                location + "\" sync=false";
 
-    LOG_INFO("[RTMP] Creating pipeline for '" + stream_name + "' → app '" + app + "', url=" + location);
+    LOG_INFO("[RTMP] Pipeline: " + pipeline_str);
 
     GError *error = nullptr;
     GstElement *pipeline = gst_parse_launch(pipeline_str.c_str(), &error);
     if (error)
     {
-        LOG_ERROR("[RTMP] Pipeline parse error: " + std::string(error->message));
+        LOG_ERROR("[RTMP] Pipeline creation failed: " + std::string(error->message));
         g_error_free(error);
         return nullptr;
     }
-
     return pipeline;
 }
 
-// 摄像头在配置中的索引（用于 reply.loc）
+void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
-static int get_camera_index(const std::string &name)
 {
-    for (size_t i = 0; i < g_app_config.cameras.size(); ++i)
+    std::string key = make_key(cam.name);
-        if (g_app_config.cameras[i].name == name) return static_cast<int>(i);
-    return -1;
-}
 
-// ---------- 对外接口：启动/停止 ----------
+    while (ctx->running)
-
-RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, StreamType type)
-{
-    StreamResultInfo res;
-    res.loc = get_camera_index(cam.name);
-    res.url = get_stream_url(cam.name, type);
-
-    if (res.loc < 0 || res.loc >= static_cast<int>(g_app_config.cameras.size()))
     {
-        res.result = 1;
+        if (!device_exists(cam.device))
-        res.reason = "Invalid channel index";
-        return res;
-    }
-
-    // 主码流（record）允许设备缺失时也进入“重试守护”，所以这里只在 SUB 时严格检查
-    if (type == StreamType::SUB && !device_exists(cam.device))
-    {
-        res.result = 1;
-        res.reason = "Device not found: " + cam.device;
-        return res;
-    }
-
-    const std::string key = make_stream_key(cam.name, type);
-
-    std::unique_ptr<StreamContext> ctx;
-    std::future<StreamStatus> status_future;
-
-    {
-        std::lock_guard<std::mutex> lock(streams_mutex);
-        auto it = streams.find(key);
-
-        // 若旧线程仍在运行，先安全停止并移除
-        if (it != streams.end())
         {
-            if (it->second->running.load())
+            ctx->status.running = false;
-            {
+            ctx->status.last_error = "Device not found: " + cam.device;
-                LOG_WARN("[RTMP] Restarting existing stream: " + key);
+            LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
-                it->second->running.store(false);
+            std::this_thread::sleep_for(std::chrono::seconds(3));
-                if (it->second->thread.joinable()) it->second->thread.join();
+            continue;
-            }
-            streams.erase(it);
         }
 
-        ctx = std::make_unique<StreamContext>();
+        GstElement *pipeline = create_pipeline(cam);
-        ctx->running.store(true);
+        if (!pipeline)
-        status_future = ctx->start_promise.get_future();
+        {
+            ctx->status.running = false;
+            ctx->status.last_error = "Failed to create pipeline";
+            std::this_thread::sleep_for(std::chrono::seconds(3));
+            continue;
+        }
 
-        StreamContext *ctx_ptr = ctx.get();
+        GstBus *bus = gst_element_get_bus(pipeline);
-        ctx->thread = std::thread(
+        gst_element_set_state(pipeline, GST_STATE_PLAYING);
-            [cam, type, ctx_ptr]()
+        ctx->status.running = true;
+        ctx->status.last_error.clear();
+        LOG_INFO("[RTMP] Started stream for " + key);
+
+        bool need_restart = false;
+        while (ctx->running)
+        {
+            GstMessage *msg = gst_bus_timed_pop_filtered(bus, 500 * GST_MSECOND,
+                                                         (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
+
+            if (!msg) continue;
+
+            switch (GST_MESSAGE_TYPE(msg))
             {
-                try
+                case GST_MESSAGE_ERROR:
                 {
-                    RTMPManager::stream_loop(cam, type, ctx_ptr);
+                    GError *err = nullptr;
+                    gst_message_parse_error(msg, &err, nullptr);
+                    ctx->status.running = false;
+                    ctx->status.last_error = err ? err->message : "Unknown GStreamer error";
+                    LOG_ERROR("[RTMP] " + key + " stream error: " + ctx->status.last_error);
+                    if (err) g_error_free(err);
+                    need_restart = true;
+                    break;
                 }
-                catch (const std::exception &e)
+                case GST_MESSAGE_EOS:
-                {
+                    ctx->status.running = false;
-                    LOG_ERROR("[RTMP] Exception in stream thread: " + std::string(e.what()));
+                    ctx->status.last_error = "End of stream";
-                }
+                    need_restart = true;
-                catch (...)
+                    break;
-                {
+                default:
-                    LOG_ERROR("[RTMP] Unknown exception in stream thread");
+                    break;
-                }
+            }
-            });
+            gst_message_unref(msg);
 
-        streams.emplace(key, std::move(ctx));
+            if (need_restart) break;
-    }
-
-    // 等待首帧或错误，最多 7 秒
-    if (status_future.wait_for(std::chrono::seconds(7)) == std::future_status::ready)
-    {
-        const StreamStatus s = status_future.get();
-        res.result = s.running ? 0 : 1;
-        res.reason = s.running ? "Started OK" : (s.last_error.empty() ? "Failed to start" : s.last_error);
-    }
-    else
-    {
-        res.result = 1;
-        res.reason = "Timeout waiting for stream";
-    }
-
-    return res;
-}
-
-RTMPManager::StreamResultInfo RTMPManager::stop_camera(const std::string &cam_name, StreamType type)
-{
-    StreamResultInfo res;
-    res.loc = get_camera_index(cam_name);
-    res.url = get_stream_url(cam_name, type);
-
-    if (res.loc < 0 || res.loc >= static_cast<int>(g_app_config.cameras.size()))
-    {
-        res.result = 1;
-        res.reason = "Invalid channel index";
-        return res;
-    }
-
-    std::unique_ptr<StreamContext> ctx;
-    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())
-        {
-            res.result = 0;
-            res.reason = "Already stopped (no active stream)";
-            return res;
         }
 
-        ctx = std::move(it->second);
+        gst_element_set_state(pipeline, GST_STATE_NULL);
-        streams.erase(it);
+        gst_object_unref(bus);
-    }
+        gst_object_unref(pipeline);
 
-    bool was_running = ctx->running.load();
+        if (ctx->running)
-    ctx->running.store(false);
-    if (ctx->thread.joinable()) ctx->thread.join();
-
-    if (was_running)
-    {
-        res.result = 0;
-        res.reason = "Stopped manually";
-    }
-    else
-    {
-        res.result = 1;
-        res.reason = "Not running (stream never started)";
-    }
-
-    return res;
-}
-
-// ---------- 核心：推流循环（带重试策略） ----------
-
-void RTMPManager::stream_loop(Camera cam, StreamType type, StreamContext *ctx)
-{
-    const std::string key = make_stream_key(cam.name, type);
-
-    auto try_set_start = [&](const StreamStatus &st)
-    {
-        bool expected = false;
-        if (ctx && ctx->start_promise_set.compare_exchange_strong(expected, true))
         {
-            try
+            LOG_WARN("[RTMP] Restarting " + key + " in 3s...");
-            {
-                ctx->start_promise.set_value(st);
-            }
-            catch (...)
-            {
-            }
-        }
-    };
-
-    // MAIN（record）= 永久自动重试；SUB（live）= 限次重试
-    if (type == StreamType::MAIN)
-    {
-        int backoff_ms = RETRY_BASE_DELAY_MS;  // 可指数退避
-        while (ctx->running.load())
-        {
-            const bool ok = do_stream_once(cam, type, ctx);
-            if (!ok && ctx->running.load())
-            {
-                LOG_WARN("[RTMP] MAIN(record) stream lost, retry in " + std::to_string(backoff_ms) + "ms: " + cam.name);
-                std::this_thread::sleep_for(std::chrono::milliseconds(backoff_ms));
-                backoff_ms = std::min(backoff_ms * 2, 60000);  // up to 60s
-            }
-            else
-            {
-                backoff_ms = RETRY_BASE_DELAY_MS;  // 成功后重置退避
-            }
-        }
-    }
-    else  // SUB / live
-    {
-        int retries = 0;
-        while (ctx->running.load())
-        {
-            const bool ok = do_stream_once(cam, type, ctx);
-            if (ok) break;  // 正常退出（被 stop 或 EOS）
-            retries++;
-            if (retries >= LIVE_MAX_RETRIES || !ctx->running.load())
-            {
-                LOG_ERROR("[RTMP] SUB(live) stream failed after retries, giving up: " + cam.name);
-                break;
-            }
-            LOG_WARN("[RTMP] SUB(live) retry " + std::to_string(retries) + "/" + std::to_string(LIVE_MAX_RETRIES) +
-                     " in " + std::to_string(RETRY_BASE_DELAY_MS) + "ms: " + cam.name);
             std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS));
         }
     }
@@ -335,323 +172,49 @@ void RTMPManager::stream_loop(Camera cam, StreamType type, StreamContext *ctx)
     ctx->status.running = false;
 }
 
-// 单次推流：创建 pipeline -> PLAYING -> loop 直到 ERROR/EOS/stop
+void RTMPManager::start_all()
-bool RTMPManager::do_stream_once(const Camera &cam, StreamType type, StreamContext *ctx)
 {
-    const std::string key = make_stream_key(cam.name, type);
+    LOG_INFO("[RTMP] Starting all record streams...");
+    std::lock_guard<std::mutex> lock(streams_mutex);
 
-    auto try_set_start = [&](const StreamStatus &st)
+    for (auto &cam : g_app_config.cameras)
     {
-        bool expected = false;
+        auto key = make_key(cam.name);
-        if (ctx && ctx->start_promise_set.compare_exchange_strong(expected, true))
+        auto ctx = std::make_unique<StreamContext>();
-        {
+        ctx->running.store(true);
-            try
+        ctx->thread = std::thread([cam, ptr = ctx.get()]() { stream_loop(cam, ptr); });
-            {
+        streams.emplace(key, std::move(ctx));
-                ctx->start_promise.set_value(st);
-            }
-            catch (...)
-            {
-            }
-        }
-    };
-
-    RTMPManager::StreamStatus status;
-    status.running = false;
-    status.last_result = StreamResult::UNKNOWN;
-    status.last_error.clear();
-
-    GstElement *pipeline = create_pipeline(cam, type);
-    if (!pipeline)
-    {
-        status.last_result = StreamResult::PIPELINE_ERROR;
-        status.last_error = "Failed to create pipeline";
-        update_status(key, status);
-        try_set_start(status);
-        return false;
     }
-
-    GstBus *bus = gst_element_get_bus(pipeline);
-    gst_element_set_state(pipeline, GST_STATE_PLAYING);
-
-    bool started_reported = false;
-    auto t0 = std::chrono::steady_clock::now();
-    const int first_frame_timeout_s = 5;
-    const int connect_guard_ms = 2000;
-
-    bool need_restart = false;
-    bool ok = false;
-
-    while (true)
-    {
-        if (!ctx->running.load())
-        {
-            status.running = false;
-            status.last_result = StreamResult::UNKNOWN;
-            status.last_error = "Stream stopped manually";
-            update_status(key, status);
-            try_set_start(status);
-            break;
-        }
-
-        // 首帧/连通性超时保护
-        if (!started_reported)
-        {
-            auto elapsed_s =
-                std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - t0).count();
-            if (elapsed_s > first_frame_timeout_s)
-            {
-                status.running = false;
-                status.last_result = StreamResult::TIMEOUT;
-                status.last_error = "No frames/connection established within timeout";
-                update_status(key, status);
-                try_set_start(status);
-                need_restart = true;
-                break;
-            }
-        }
-
-        GstMessage *msg =
-            gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND,
-                                       (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS |
-                                                        GST_MESSAGE_STATE_CHANGED | GST_MESSAGE_STREAM_STATUS));
-
-        if (!msg)
-        {
-            // 过了 connect_guard_ms 且 rtmpsink 进入 PLAYING，认为启动成功
-            if (!started_reported)
-            {
-                auto elapsed_ms =
-                    std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - t0)
-                        .count();
-                if (elapsed_ms >= connect_guard_ms)
-                {
-                    GstElement *sink = gst_bin_get_by_name(GST_BIN(pipeline), "rtmp_sink");
-                    if (sink)
-                    {
-                        GstState cur, pending;
-                        if (gst_element_get_state(sink, &cur, &pending, 0) == GST_STATE_CHANGE_SUCCESS &&
-                            cur == GST_STATE_PLAYING)
-                        {
-                            status.running = true;
-                            ctx->status.running = true;
-                            status.last_result = StreamResult::OK;
-                            status.last_error.clear();
-                            update_status(key, status);
-                            try_set_start(status);
-                            started_reported = true;
-                            LOG_INFO("[RTMP] Guard-start OK for '" + cam.name + "'");
-                        }
-                        gst_object_unref(sink);
-                    }
-                }
-            }
-            continue;
-        }
-
-        switch (GST_MESSAGE_TYPE(msg))
-        {
-            case GST_MESSAGE_STREAM_STATUS:
-            {
-                GstStreamStatusType type_ss;
-                GstElement *owner = nullptr;
-                gst_message_parse_stream_status(msg, &type_ss, &owner);
-                if (!started_reported && owner)
-                {
-                    const gchar *oname = GST_OBJECT_NAME(owner);
-                    if (oname && std::strcmp(oname, "rtmp_sink") == 0)
-                    {
-                        if (type_ss == GST_STREAM_STATUS_TYPE_CREATE || type_ss == GST_STREAM_STATUS_TYPE_ENTER ||
-                            type_ss == GST_STREAM_STATUS_TYPE_START)
-                        {
-                            status.running = true;
-                            ctx->status.running = true;
-                            status.last_result = StreamResult::OK;
-                            status.last_error.clear();
-                            update_status(key, status);
-                            try_set_start(status);
-                            started_reported = true;
-                            LOG_INFO(std::string("[RTMP] Connected (STREAM_STATUS) for '") + cam.name + "'");
-                        }
-                    }
-                }
-                break;
-            }
-
-            case GST_MESSAGE_STATE_CHANGED:
-                // 可用于扩展日志；不以 pipeline 的 PLAYING 作为成功判据
-                break;
-
-            case GST_MESSAGE_ERROR:
-            {
-                GError *err = nullptr;
-                gchar *dbg = nullptr;
-                gst_message_parse_error(msg, &err, &dbg);
-
-                status.running = false;
-                status.last_result = StreamResult::CONNECTION_FAIL;
-                status.last_error = err ? err->message : "GStreamer error";
-
-                LOG_ERROR("[RTMP] Stream error from '" + cam.name + "': " + status.last_error);
-
-                ctx->status.running = false;
-                need_restart = true;
-
-                if (err) g_error_free(err);
-                if (dbg) g_free(dbg);
-                update_status(key, status);
-                try_set_start(status);
-                break;
-            }
-
-            case GST_MESSAGE_EOS:
-            {
-                status.running = false;
-                status.last_result = StreamResult::EOS_RECEIVED;
-                status.last_error = "EOS received";
-                update_status(key, status);
-                try_set_start(status);
-                need_restart = true;
-                break;
-            }
-
-            default:
-                break;
-        }
-
-        gst_message_unref(msg);
-
-        if (need_restart) break;
-        if (started_reported)
-        {
-            // 正常运行中，继续监听
-        }
-    }
-
-    // 收尾
-    gst_element_set_state(pipeline, GST_STATE_NULL);
-    if (bus) gst_object_unref(bus);
-    gst_object_unref(pipeline);
-
-    ok = started_reported && !need_restart;  // 若在 stop 前一直运行且未触发重启则认为 ok
-    return ok;
 }
 
-// ---------- 其他工具 ----------
-
 void RTMPManager::stop_all()
-{
-    std::vector<std::unique_ptr<StreamContext>> ctxs;
-    {
-        std::lock_guard<std::mutex> lock(streams_mutex);
-        for (auto &kv : streams) kv.second->running.store(false);
-        for (auto &kv : streams) ctxs.push_back(std::move(kv.second));
-        streams.clear();
-    }
-
-    for (auto &ctx : ctxs)
-        if (ctx->thread.joinable()) ctx->thread.join();
-
-    LOG_INFO("[RTMP] stop_all completed.");
-}
-
-bool RTMPManager::is_streaming(const std::string &cam_name, StreamType type)
-{
-    std::lock_guard<std::mutex> lock(streams_mutex);
-    auto it = streams.find(make_stream_key(cam_name, type));
-    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) kv.second->running.store(false);
     for (auto &kv : streams)
-        if (kv.second->running.load()) return true;
+        if (kv.second->thread.joinable()) kv.second->thread.join();
-    return false;
+    streams.clear();
 }
 
-std::string RTMPManager::get_stream_url(const std::string &cam_name, StreamType type)
+bool RTMPManager::is_streaming(const std::string &cam_name)
 {
-    // 你可以把这里的网卡名替换成你的默认上行网卡
+    std::lock_guard<std::mutex> lock(streams_mutex);
+    auto it = streams.find(make_key(cam_name));
+    return (it != streams.end() && it->second->status.running);
+}
+
+std::string RTMPManager::get_stream_url(const std::string &cam_name)
+{
+    // 优先动态检测网卡 IP，失败则使用固定值
     std::string ip = get_ip_address("enP2p33s0");
-    if (ip.empty()) ip = "127.0.0.1";
+    if (ip.empty()) ip = "127.0.0.1";  // 或用动态获取网卡 IP
-
+    return "http://" + ip + ":1985/rtc/v1/whep/?app=record&stream=" + cam_name + "_main";
-    const std::string app = pick_app(type);
-    const std::string stream_name = make_stream_key(cam_name, type);
-
-    // 返回 WebRTC 拉流地址（WHEP）
-    return "http://" + ip + ":1985/rtc/v1/whep/?app=" + app + "&stream=" + stream_name;
 }
 
-// 批量处理
+std::vector<std::pair<std::string, bool>> RTMPManager::get_all_status()
-std::vector<RTMPManager::StreamResultInfo> RTMPManager::process_push_request(const VideoPushRequest &req)
 {
-    std::vector<StreamResultInfo> results;
+    std::vector<std::pair<std::string, bool>> result;
-    std::vector<std::future<StreamResultInfo>> futures;
+    std::lock_guard<std::mutex> lock(streams_mutex);
-
+    for (auto &kv : streams) result.emplace_back(kv.first, kv.second->status.running);
-    for (const auto &item : req.data)
+    return result;
-    {
-        StreamType type = (item.streamType == 0) ? StreamType::MAIN : StreamType::SUB;
-
-        for (int ch : item.channels)
-        {
-            if (ch < 0 || ch >= static_cast<int>(g_app_config.cameras.size())) continue;
-
-            const auto &cam = g_app_config.cameras[ch];
-
-            if (item.switchVal == 0)
-            {
-                futures.emplace_back(
-                    std::async(std::launch::async, [&, cam, type]() { return start_camera(cam, type); }));
-            }
-            else
-            {
-                futures.emplace_back(
-                    std::async(std::launch::async, [&, cam, type]() { return stop_camera(cam.name, type); }));
-            }
-        }
-    }
-
-    for (auto &f : futures)
-    {
-        try
-        {
-            results.push_back(f.get());
-        }
-        catch (const std::exception &e)
-        {
-            StreamResultInfo err;
-            err.result = 1;
-            err.reason = std::string("Exception: ") + e.what();
-            results.push_back(err);
-        }
-    }
-
-    return results;
-}
-
-void RTMPManager::start_all_record_streams()
-{
-    LOG_INFO("[RTMP] Auto-starting all record (MAIN) streams...");
-
-    for (const auto &cam : g_app_config.cameras)
-    {
-        // 检查是否已经存在
-        if (is_streaming(cam.name, StreamType::MAIN))
-        {
-            LOG_INFO("[RTMP] Record stream already running: " + cam.name);
-            continue;
-        }
-
-        auto res = start_camera(cam, StreamType::MAIN);
-        if (res.result == 0)
-        {
-            LOG_INFO("[RTMP] Record stream started for " + cam.name + ": " + res.reason);
-        }
-        else
-        {
-            LOG_WARN("[RTMP] Failed to start record stream for " + cam.name + ": " + res.reason);
-        }
-    }
-
-    LOG_INFO("[RTMP] Auto-start for record streams done.");
 }