kunlang_video/src/rtmp_manager.cpp

// rtsp_manager.cpp
#include "rtmp_manager.hpp"

#include <arpa/inet.h>
#include <ifaddrs.h>
#include <net/if.h>
#include <netinet/in.h>
#include <sys/stat.h>

#include <chrono>
#include <cstring>
#include <thread>

// 动态获取指定网卡 IPv4 地址
static std::string get_ip_address(const std::string &ifname)
{
    struct ifaddrs *ifaddr, *ifa;
    char ip[INET_ADDRSTRLEN] = {0};

    if (getifaddrs(&ifaddr) == -1)
    {
        perror("getifaddrs");
        return "";
    }

    for (ifa = ifaddr; ifa != nullptr; ifa = ifa->ifa_next)
    {
        if (ifa->ifa_addr == nullptr) continue;

        if (ifa->ifa_addr->sa_family == AF_INET && ifname == ifa->ifa_name)
        {
            void *addr = &((struct sockaddr_in *)ifa->ifa_addr)->sin_addr;
            if (inet_ntop(AF_INET, addr, ip, sizeof(ip)))
            {
                freeifaddrs(ifaddr);
                return ip;
            }
        }
    }

    freeifaddrs(ifaddr);
    return "";  // 没找到
}

std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
std::mutex RTMPManager::streams_mutex;

static bool device_exists(const std::string &path)
{
    struct stat st;
    return (stat(path.c_str(), &st) == 0);
}

void RTMPManager::init()
{
    gst_init(nullptr, nullptr);
    LOG_INFO("[RTMP] GStreamer initialized.");
}

std::string RTMPManager::make_key(const std::string &name) { return name + "_main"; }

GstElement *RTMPManager::create_pipeline(const Camera &cam)
{
    int width = cam.width;
    int height = cam.height;
    int fps = cam.fps;
    int bitrate = cam.bitrate;

    std::string stream_name = cam.name + "_main";
    std::string app = "record";

    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 ! videoconvert ! video/x-raw,format=NV12 "
                               "! mpph264enc bps=" +
                               std::to_string(bitrate) + " gop=" + std::to_string(fps) +
                               " ! h264parse ! flvmux name=mux streamable=true "
                               "audiotestsrc wave=silence ! audioconvert ! audioresample ! voaacenc ! aacparse ! mux. "
                               "mux. ! rtmpsink location=\"" +
                               location + "\" sync=false";

    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 creation failed: " + std::string(error->message));
        g_error_free(error);
        return nullptr;
    }
    return pipeline;
}

void RTMPManager::stream_loop(Camera cam, StreamContext *ctx)
{
    std::string key = make_key(cam.name);

    while (ctx->running)
    {
        if (!device_exists(cam.device))
        {
            ctx->status.running = false;
            ctx->status.last_error = "Device not found: " + cam.device;
            LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
            std::this_thread::sleep_for(std::chrono::seconds(3));
            continue;
        }

        GstElement *pipeline = create_pipeline(cam);
        if (!pipeline)
        {
            ctx->status.running = false;
            ctx->status.last_error = "Failed to create pipeline";
            std::this_thread::sleep_for(std::chrono::seconds(3));
            continue;
        }

        GstBus *bus = gst_element_get_bus(pipeline);
        gst_element_set_state(pipeline, GST_STATE_PLAYING);

        // -------- 等待状态切换结果 --------
        GstStateChangeReturn ret = gst_element_get_state(pipeline, nullptr, nullptr, 3 * GST_SECOND);

        if (ret != GST_STATE_CHANGE_SUCCESS)
        {
            ctx->status.running = false;
            ctx->status.last_error = "Pipeline failed to reach PLAYING";
            LOG_ERROR("[RTMP] " + key + " failed to start (no video signal?)");
            gst_element_set_state(pipeline, GST_STATE_NULL);
            gst_object_unref(bus);
            gst_object_unref(pipeline);
            std::this_thread::sleep_for(std::chrono::seconds(3));
            continue;
        }

        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))
            {
                case GST_MESSAGE_ERROR:
                {
                    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;
                }
                case GST_MESSAGE_EOS:
                    ctx->status.running = false;
                    ctx->status.last_error = "End of stream";
                    need_restart = true;
                    break;
                default:
                    break;
            }

            gst_message_unref(msg);
            if (need_restart) break;
        }

        gst_element_set_state(pipeline, GST_STATE_NULL);
        gst_object_unref(bus);
        gst_object_unref(pipeline);

        if (ctx->running)
        {
            LOG_WARN("[RTMP] Restarting " + key + " in 3s...");
            std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS));
        }
    }

    ctx->status.running = false;
}

void RTMPManager::start_all()
{
    LOG_INFO("[RTMP] Starting all record streams...");
    std::lock_guard<std::mutex> lock(streams_mutex);

    for (auto &cam : g_app_config.cameras)
    {
        auto key = make_key(cam.name);
        auto ctx = std::make_unique<StreamContext>();
        ctx->running.store(true);
        ctx->thread = std::thread([cam, ptr = ctx.get()]() { stream_loop(cam, ptr); });
        streams.emplace(key, std::move(ctx));
    }
}

void RTMPManager::stop_all()
{
    std::lock_guard<std::mutex> lock(streams_mutex);
    for (auto &kv : streams) kv.second->running.store(false);
    for (auto &kv : streams)
        if (kv.second->thread.joinable()) kv.second->thread.join();
    streams.clear();
}

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";  // 或用动态获取网卡 IP
    return "http://" + ip + ":1985/rtc/v1/whep/?app=record&stream=" + cam_name + "_main";
}

std::vector<std::pair<std::string, bool>> RTMPManager::get_all_status()
{
    std::vector<std::pair<std::string, bool>> result;
    std::lock_guard<std::mutex> lock(streams_mutex);
    for (auto &kv : streams) result.emplace_back(kv.first, kv.second->status.running);
    return result;
}