使用 C++ 构建简单的 RTSP 流媒体转发服务器 | 极客日志

C++

使用 C++ 构建简单的 RTSP 流媒体转发服务器

本文介绍如何使用 C++ 和 wepoll 库构建一个简单的 RTSP 流媒体转发服务器。项目包含 EventPoller（并发封装）、RtspServer（服务接入）、RtspSession（会话管理）、SdpParser（SDP 解析）和 MediaPusherManager（推拉流管理）五个核心类。支持桌面推流及播放器拉流播放，采用 TCP 传输协议，原封不动转发 RTP 包。代码结构清晰，适合学习 RTSP 协议交互流程。

念念不忘发布于 2026/3/29更新于 2026/4/141 浏览

使用效果举例

运行程序代码后，打开 cmd，执行以下代码进行桌面推流:

ffmpeg -f gdigrab -i desktop -c:v libx264 -f rtsp -rtsp_transport tcp rtsp://127.0.0.1/live/test

这会将流推送给流媒体，创建了一条流媒体的转发流 rtsp://127.0.0.1/live/test。随后使用 PotPlayer 或者 VLC Player 拉取该流，即可成功播放转发流。

注意：这个流媒体是简单的 RTSP 流媒体转发程序，不做 RTP 和 RTCP 包的分析和重构处理，仅原封不动地将发送至流媒体的 RTSP 流转发出去。

相关免费在线工具

Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。在线工具，Base64 字符串编码/解码在线工具，online
Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。在线工具，Base64 文件转换器在线工具，online
Markdown 转 HTML
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML 转 Markdown 互为补充。在线工具，Markdown 转 HTML在线工具，online
HTML 转 Markdown
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。在线工具，HTML 转 Markdown在线工具，online
JSON 压缩
通过删除不必要的空白来缩小和压缩JSON。在线工具，JSON 压缩在线工具，online
JSON美化和格式化
将JSON字符串修饰为友好的可读格式。在线工具，JSON美化和格式化在线工具，online

#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include <winsock2.h>
#include <ws2tcpip.h>
#include <iphlpapi.h>
#pragma comment(lib, "Ws2_32.lib")
#pragma comment(lib, "Iphlpapi.lib")
#include <string>
#include <map>
#include <vector>
#include <iostream>
#include "EventPoller.h"
#include "RtspServer.h"

int main() {
    bool isRet;
    // 检查网络 WSA 是否可用，不可用则退出
    WORD wVersionRequested = MAKEWORD(2, 2);
    WSADATA wsaData;
    isRet = (0 == WSAStartup(wVersionRequested, &wsaData));
    if (false == isRet) return 1;

    EventPoller eventPoller;
    auto rtspServer = std::make_shared<RtspServer>(&eventPoller);
    if (rtspServer->Start("127.0.0.1", 554)) {
        std::cout << "已就绪，请试着用 ffmpeg 运行如下命令进行推流:ffmpeg -f gdigrab -i desktop -c:v libx264 -f rtsp -rtsp_transport tcp rtsp://127.0.0.1/live/test\n";
        std::cout << "推流成功后，可以使用某个播放器 (比如 VlcPlayer/PotPlayer) 对 rtsp://127.0.0.1/live/test 进行拉流以查看转发效果\n";
        eventPoller.Exec();
        rtspServer->Stop();
    }
    rtspServer.reset();
    WSACleanup();
    return 0;
}

#pragma once
#include <string>
#include <functional>
#include <mutex>
typedef void* HANDLE;

/// <summary>
/// epoll 处理线程：这是对 epoll 三方库的进一步封装，它为网络套接字服务，负责单线程内处理千万级以上的套接字连接
/// 每个类实例都封装了一条线程用于接收并处理套接字的相关业务
/// </summary>
class EventPoller {
public:
    using PollEventCB = std::function<void(int event)>; // 套接字的监听回调
    // 提供给 wepoll 的监听事件类型
    typedef enum {
        Event_Read = 1 << 0, // 读事件，对标 wepoll.h 的 EPOLLIN
        Event_Write = 1 << 2, // 写事件，对标 wepoll.h 的 EPOLLOUT
        Event_Error = 1 << 3, // 错误事件，对标 wepoll.h 的 EPOLLERR
    } Poll_Event;

    /// <summary>
    /// 构造
    /// </summary>
    EventPoller();
    ~EventPoller();

    /// <summary>
    /// 添加事件监听
    /// </summary>
    /// <param name="fd">监听的套接字</param>
    /// <param name="event">事件类型，例如 Event_Read | Event_Write</param>
    /// <param name="cb">事件回调 functional</param>
    /// <returns>是否成功</returns>
    bool AddEvent(int fd, int event, PollEventCB cb);

    /// <summary>
    /// 删除事件监听
    /// </summary>
    /// <param name="fd">监听的套接字</param>
    /// <returns>是否成功</returns>
    bool DelEvent(int fd);

    /// <summary>
    /// 执行事件轮询死循环
    /// </summary>
    void Exec();

    /// <summary>
    /// 获取当前线程下所有 socket 共享的读缓存
    /// </summary>
    /// <returns></returns>
    std::vector<char>& GetSharedBuffer();

private:
    bool m_isRun = false; // 处理线程是否正在执行
    HANDLE m_hEpoll = NULL; // 本例中管控 socket 的 epoll 的 fd
    std::unordered_map<int, std::shared_ptr<PollEventCB>> m_mapEventCB; // 监听的套接字和他的处理回调 <套接字，处理回调>
    std::vector<char> m_vecSharedBuffer; // 当前线程下所有 socket 共享的读缓存
};

#include "EventPoller.h"
#include "wepoll.h"
#include "Windows.h"

// 可管理的最大套接字数量，对标 select 的最大管理数量，但其实这个值并没有用，传进 epoll_create 里会被忽略
#define EPOLL_SIZE 1024

EventPoller::EventPoller() {
    m_vecSharedBuffer.resize(32 * 4 * 1024); // 128k 共享缓存
    // 给本例创建对应的 epoll
    m_hEpoll = epoll_create(EPOLL_SIZE);
}

EventPoller::~EventPoller() {}

bool EventPoller::AddEvent(int fd, int event, PollEventCB cb) {
    // 如果是当前线程进行的事件添加则直接放进 map 里
    struct epoll_event ev = { 0 };
    ev.events = event;
    ev.data.fd = fd;
    const int iRet = epoll_ctl(m_hEpoll, EPOLL_CTL_ADD, fd, &ev);
    if (-1 == iRet) {
        return false;
    }
    // 记录要监听的套接字和其对应的处理回调
    m_mapEventCB.emplace(fd, std::make_shared<PollEventCB>(std::move(cb)));
    return true;
}

bool EventPoller::DelEvent(int fd) {
    int iRet = -1;
    // 仅删除监听中的 fd，如果没删除成功则意味着对应的 fd 在逻辑上已不受监听而无需 DEL。
    // 如果后续发现在监听回调里出现了不在监听中的 fd，则在监听回调里再进行 DEL，同时忽略那次的监听执行。
    if (m_mapEventCB.erase(fd)) {
        iRet = epoll_ctl(m_hEpoll, EPOLL_CTL_DEL, fd, nullptr);
    }
    return -1 != iRet;
}

void EventPoller::Exec() {
    m_isRun = true;
    int iMinDelay = 5000; // 等待事件的超时毫秒数
    struct epoll_event* events = new struct epoll_event[EPOLL_SIZE];
    while (m_isRun) {
        int ret = epoll_wait(m_hEpoll, events, EPOLL_SIZE, iMinDelay);
        if (ret <= 0) continue;
        for (int i = 0; i < ret; ++i) {
            struct epoll_event& ev = events[i];
            int fd = ev.data.fd;
            // 找找要被监听的事件 map 里有没有这个 fd，没有就移除掉这个残留的监听
            auto it = m_mapEventCB.find(fd);
            if (it == m_mapEventCB.end()) {
                epoll_ctl(m_hEpoll, EPOLL_CTL_DEL, fd, nullptr);
                continue;
            }
            (*(it->second))(ev.events);
        }
    }
    delete[] events;
}

std::vector<char>& EventPoller::GetSharedBuffer() {
    return m_vecSharedBuffer;
}

#pragma once
#include <memory>
#include <mutex>
#include <unordered_map>

class EventPoller;
class RtspSession;

class RtspServer : public std::enable_shared_from_this<RtspServer> {
public:
    explicit RtspServer(EventPoller* pEventPoller);

    /// <summary>
    /// 启动服务器
    /// </summary>
    /// <param name="szHost">主机</param>
    /// <param name="uPort">端口</param>
    /// <returns></returns>
    bool Start(const std::string& szHost, uint16_t uPort);

    // 停止服务器运行
    void Stop();

    // 关闭指定的 session 连接
    void CloseSession(int fd);

private:
    EventPoller* m_pEventPoller = nullptr; // 监听事件的实例
    std::string m_szHost; // 服务器监听 ip
    uint16_t m_uPort = 0; // 服务器端口
    int m_fd = -1; // 服务器套接字
    std::unordered_map<int, std::shared_ptr<RtspSession>> m_mapSession; // 保存所有的连接 session
    std::mutex m_mtxSession; // 保护 session 容器的互斥量
};

#pragma once
#include "RtspServer.h"
#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include <winsock2.h>
#include <ws2tcpip.h>
#include <iphlpapi.h>
#pragma comment(lib, "Ws2_32.lib")
#pragma comment(lib, "Iphlpapi.lib")
#include <string>
#include <map>
#include <vector>
#include <thread>
#include "EventPoller.h"
#include "RtspSession.h"

using namespace std;

RtspServer::RtspServer(EventPoller* pEventPoller) {
    m_pEventPoller = pEventPoller;
}

bool RtspServer::Start(const std::string& szHost, uint16_t uPort) {
    m_szHost = szHost;
    m_uPort = uPort;
    m_fd = (int)socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    int opt = 1;
    // 设置套接字属性，此处处理服务器快速重启后的以前使用的套接字还在 TIME_WAIT 的问题，配置后可以直接接着使用
    setsockopt(m_fd, SOL_SOCKET, SO_REUSEADDR, (char*)&opt, static_cast<socklen_t>(sizeof(opt)));
    unsigned long ul = 1;
    // 设置为非阻塞模式
    ioctlsocket(m_fd, FIONBIO, &ul);
    struct sockaddr_in addr;
    ZeroMemory(&addr, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(uPort);
    addr.sin_addr.s_addr = INADDR_ANY;
    if (::bind(m_fd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
        DebugBreak();
    }
    if (::listen(m_fd, 1024) == -1) {
        DebugBreak();
    }
    // weak_self 的存在是为了防止发生回调时，本类的实例已经析构的情况
    // 这是因为 m_pEventPoller->DelEvent 只是停止监听新事件，但队列里的事件仍然有可能在实例析构后仍触发该回调
    // 此时必须忽略该回调，否则会访问野指针
    std::weak_ptr<RtspServer> weak_self = shared_from_this();
    // 这个事件处理主要用于 accept 新的套接字
    bool isRet = m_pEventPoller->AddEvent(m_fd, EventPoller::Event_Read | EventPoller::Event_Error, [weak_self](int event) {
        auto strong_self = weak_self.lock();
        if (nullptr == strong_self) return;
        const int fdServer = strong_self->m_fd;
        int fd;
        struct sockaddr_storage peer_addr;
        socklen_t addr_len = sizeof(peer_addr);
        // 这个 while 主要是用来尝试一次获取多个 accept 新连接
        while (true) {
            if (event & EventPoller::Event_Read) {
                fd = (int)accept(fdServer, (struct sockaddr*)&peer_addr, &addr_len);
                if (fd == -1) {
                    // accept 失败
                    int iRet = WSAGetLastError();
                    if (iRet == WSAEWOULDBLOCK) {
                        // 没有新的连接 accept 进来了
                        return;
                    }
                    DebugBreak();
                    return;
                }
                auto session = std::make_shared<RtspSession>(strong_self, strong_self->m_pEventPoller, fd);
                if (session->StartSession()) {
                    strong_self->m_mtxSession.lock();
                    strong_self->m_mapSession[fd] = session;
                    strong_self->m_mtxSession.unlock();
                }
            }
            if (event & EventPoller::Event_Error) {
                return;
            }
        }
    });
    // 没附加事件处理成功就直接停止服务器运行
    if (false == isRet) Stop();
    return isRet;
}

void RtspServer::Stop() {
    if (-1 != m_fd) {
        m_pEventPoller->DelEvent(m_fd);
        closesocket(m_fd);
        m_fd = -1;
    }
}

void RtspServer::CloseSession(int fd) {
    thread task([this, fd]() {
        m_mtxSession.lock();
        m_mapSession.erase(fd);
        m_mtxSession.unlock();
    });
    task.detach();
}

#pragma once
#include <vector>
#include <string>
#include <unordered_map>
#include <memory>
#include "SdpParser.h"

class EventPoller;
class RtspServer;

/// <summary>
/// 对于发往 RtspServer 的每一个连接，都会对应一个 RtspSession 实例，用于保存该连接的状态信息
/// </summary>
class RtspSession : public std::enable_shared_from_this<RtspSession> {
public:
    RtspSession(std::weak_ptr<RtspServer> pServer, EventPoller* pEventPoller, int fd);
    virtual ~RtspSession();

    // 附加事件监听从而启动该 session 的处理
    bool StartSession();

    // 向服务器通知关闭该 session
    void CloseSession();

    // 发送数据给该 session 对应的 fd
    bool SendData(const char* buffer, size_t uSize);

protected:
    // 处理给定的套接字的读取请求，尽可能地获取数据后交给 onRecv 进一步处理
    size_t onRead(std::vector<char>& buffer);

    // 接收并处理 fd 对应的数据
    void onRecv(char* buffer, size_t uSize);

private:
    // 解析请求头
    bool parseHeader();

    // 处理一次完整的请求
    void handleReq();

    // 处理 rtp 包
    void handleRtpPacket();

private:
    // 处理 options 方法，获取服务器能力
    void handleReq_Options(const std::unordered_map<std::string, std::string>& parser);
    // 处理 describe 方法，请求服务器 rtsp sdp 信息
    void handleReq_Describe(const std::unordered_map<std::string, std::string>& parser);
    // 处理 ANNOUNCE 方法，请求推流，附带 sdp
    void handleReq_ANNOUNCE(const std::unordered_map<std::string, std::string>& parser);
    // 处理 record 方法，开始推流
    void handleReq_RECORD(const std::unordered_map<std::string, std::string>& parser);
    // 处理 setup 方法，播放和推流协商 rtp 传输方式用
    void handleReq_SETUP(const std::unordered_map<std::string, std::string>& parser);
    // 处理 play 方法，开始或恢复播放
    void handleReq_PLAY(const std::unordered_map<std::string, std::string>& parser);
    // 处理 teardown 方法，结束播放
    void handleReq_TEARDOWN(const std::unordered_map<std::string, std::string>& parser);
    // 处理 SET_PARAMETER、GET_PARAMETER 方法，一般用于心跳保活
    void handleReq_SET_PARAMETER(const std::unordered_map<std::string, std::string>& parser);

    // 回复客户端
    bool sendRtspResponse(const std::string& res_code, std::multimap<std::string, std::string> header = {}, const std::string& sdp = "", const char* protocol = "RTSP/1.0");

    // 获取 track 下标
    size_t getTrackIndexByControlUrl(const std::string& control_url);

private:
    std::weak_ptr<RtspServer> m_pServer; // session 所属的服务器
    EventPoller* m_pEventPoller = nullptr; // 监听事件的实例
    int m_fd = -1; // 连接所对应的套接字
    std::vector<char> m_vecRecv; // 接收缓冲
    std::vector<char> m_vecHeader; // 接收结果：请求头
    std::vector<char> m_vecContent; // 接收结果：请求内容
    bool m_isHeader = true; // 当前是否仍在接收请求头
    int m_iContentLength = 0; // 附带的内容长度
    size_t m_uRtpPacketLength = 0; // 正在接收的 rtp 包长度
    std::unordered_map<std::string, std::string> m_mapHeader; // 请求头的 map
    // 收到的 seq，回复时一致
    int m_iCseq = 0;
    // Session 号
    std::string m_szSessionId;
    // 解析 sdp 的实例
    SdpParser* m_pSdpParser = nullptr;
    // sdp 里面有效的 track，包含音频或视频 (每样最多一个)
    std::vector<SdpTrack*> m_vecSdpTrack;
    // 完整链接（含后面参数）
    std::string m_szFullUrl;
    // 该链接是否是推流方
    bool m_isPusher = false;
    Rtsp::ERtpType m_eRtpType = Rtsp::ERtpType::INVALID;

    //////////////////
    // 拉流时的 session 使用变量
    RtspSession* m_pPusherRtspSession = nullptr; // 在拉取谁的推流
};

#include "RtspSession.h"
#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib, "Ws2_32.lib")
#include <string>
#include <map>
#include <vector>
#include <random>
#include <iomanip>
#include <unordered_map>
#include <sstream>
#include "EventPoller.h"
#include "MediaPusherManager.h"
#include "RtspServer.h"

using namespace std;

std::string makeRandStr(size_t uLen) {
    static constexpr char CCH[] = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
    static constexpr size_t CCH_LEN = sizeof(CCH) - 1;
    // 预计算长度，避免重复计算
    string ret;
    ret.resize(uLen);
    thread_local std::mt19937 rng(std::random_device{}());
    for (size_t i = 0; i < uLen; ++i) {
        ret[i] = CCH[rng % (sizeof(CCH) - 1)];
    }
    return ret;
}

// 获取当前时间的字符串
static string dateStr() {
    char buf[64];
    time_t tt = time(NULL);
    tm t;
    gmtime_s(&t, &tt);
    strftime(buf, sizeof buf, "%a, %b %d %Y %H:%M:%S GMT", &t);
    return buf;
}

RtspSession::RtspSession(std::weak_ptr<RtspServer> pServer, EventPoller* pEventPoller, int fd) : m_pServer(pServer), m_pEventPoller(pEventPoller), m_fd(fd) {
    unsigned long ul = 1;
    // 设置为非阻塞模式
    ioctlsocket(fd, FIONBIO, &ul);
    int opt = 1;
    setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char*)&opt, static_cast<socklen_t>(sizeof(opt)));
    int size = 262144;
    setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&size, sizeof(size));
    setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char*)&size, sizeof(size));
    linger m_sLinger;
    // 在调用 closesocket() 时还有数据未发送完，允许等待
    // 若 m_sLinger.l_onoff=0; 则调用 closesocket() 后强制关闭
    m_sLinger.l_onoff = false;
    m_sLinger.l_linger = 0;
    // 设置等待时间为 x 秒
    setsockopt(fd, SOL_SOCKET, SO_LINGER, (char*)&m_sLinger, sizeof(linger));
}

RtspSession::~RtspSession() {
    if (m_isPusher) MediaPusherManager::Instance().DelPusher(m_szFullUrl);
    else MediaPusherManager::Instance().DelRecvier(m_szFullUrl, this);
}

bool RtspSession::StartSession() {
    // 给客户端套接字接入监听
    std::weak_ptr<RtspSession> weak_self = shared_from_this();
    bool isRet = m_pEventPoller->AddEvent(m_fd, EventPoller::Event_Read | EventPoller::Event_Error, [weak_self](int event) {
        auto strong_self = weak_self.lock();
        if (!strong_self) {
            // 来到这里，意味着客户端本身已经析构，这次回调只是队列残留，忽略即可
            return;
        }
        if (event & EventPoller::Event_Read) {
            strong_self->onRead(strong_self->m_pEventPoller->GetSharedBuffer());
        }
        if (event & EventPoller::Event_Error) {
            strong_self->CloseSession();
        }
    });
    return isRet;
}

void RtspSession::CloseSession() {
    m_pEventPoller->DelEvent(m_fd);
    auto pServer = m_pServer.lock();
    if (nullptr == pServer) return;
    // 通知服务器把自己关了
    pServer->CloseSession(m_fd);
}

bool RtspSession::SendData(const char* buffer, size_t uSize) {
    const int iSendRet = ::send(m_fd, buffer, uSize, 0);
    return iSendRet > 0;
}

size_t RtspSession::onRead(std::vector<char>& buffer) {
    size_t uReadSizeAll = 0, uReadSize = 0, count = 0;
    while (true) {
        uReadSize = (size_t)recv(m_fd, buffer.data(), buffer.size() - 1, 0);
        if (uReadSize == 0) {
            // 来到这里意味着套接字已被对方关闭，正常关闭回收就行
            CloseSession();
            return uReadSizeAll;
        }
        if (uReadSize == -1) {
            int iRetWSA = WSAGetLastError();
            if (iRetWSA != WSAEWOULDBLOCK) {
                CloseSession();
            }
            return uReadSizeAll;
        }
        buffer[uReadSize] = '\0';
        uReadSizeAll += uReadSize;
        // 交给子类实现具体处理这些数据
        onRecv(buffer.data(), uReadSize);
    }
}

void RtspSession::onRecv(char* buffer, size_t uSize) {
    size_t uIndex = 0;
    // 进入后续处理前，至少有 4 个缓存字节用于判断
    if (m_vecRecv.size() < 4) {
        size_t uLen = min(4 - m_vecRecv.size(), uSize);
        m_vecRecv.insert(m_vecRecv.end(), buffer, buffer + uLen);
        if (m_vecRecv.size() < 4) return;
        // 给定的数据不足，则记录已有的后就直接返回
        uSize -= uLen;
        buffer += uLen;
    }
    // 检查是否是要处理 rtp 包
    if (m_vecRecv[0] == '$') {
        char* data = m_vecRecv.data();
        size_t uDataLen = m_vecRecv.size();
        // 确定 rtp 包的接收大小
        if (0 == m_uRtpPacketLength) {
            m_uRtpPacketLength = ((((uint8_t*)data)[2] << 8) | ((uint8_t*)data)[3]) + 4;
        }
        size_t uLen = min(m_uRtpPacketLength - uDataLen, uSize);
        m_vecRecv.insert(m_vecRecv.end(), buffer, buffer + uLen);
        uSize -= uLen;
        buffer += uLen;
        // 是否已经足够一个 rtp 包可以开始处理了
        if (m_vecRecv.size() == m_uRtpPacketLength) {
            handleRtpPacket();
            if (uSize > 0) // 把剩下的数据也处理掉
                onRecv(buffer, uSize);
        }
        return;
    }
    // 处理请求，而非 rtp 包
    if (m_isHeader) {
        while (uIndex < uSize) {
            m_vecRecv.push_back(buffer[uIndex]);
            if (buffer[uIndex] == '\n') {
                // 检查是否以 rnrn 结尾
                const size_t vec_size = m_vecRecv.size();
                if (m_vecRecv[vec_size - 1] == '\n' && m_vecRecv[vec_size - 2] == '\r' && m_vecRecv[vec_size - 3] == '\n' && m_vecRecv[vec_size - 4] == '\r') {
                    m_vecRecv.push_back('\0');
                    m_vecHeader.swap(m_vecRecv);
                    m_vecRecv.clear();
                    ++uIndex;
                    // 解析请求头
                    if (parseHeader()) {
                        // 如果有附带内容，则继续解析
                        if (m_iContentLength > 0) {
                            m_isHeader = false;
                        }
                        else {
                            // 否则直接使用该请求头进行请求处理
                            handleReq();
                        }
                    }
                    else {
                        m_vecHeader.clear();
                    }
                    break; // 请求头获取结束
                }
            }
            ++uIndex;
        }
    }
    if (false == m_isHeader) {
        while (uIndex < uSize) {
            m_vecRecv.push_back(buffer[uIndex]);
            // 如果已经接收到了足够的内容
            if (m_vecRecv.size() == m_iContentLength) {
                m_vecRecv.push_back('\0');
                m_vecContent.swap(m_vecRecv);
                m_vecRecv.clear();
                m_mapHeader["reqContent"] = m_vecContent.data();
                m_isHeader = true;
                ++uIndex;
                handleReq();
                break; // 请求体获取结束
            }
            ++uIndex;
        }
    }
    // 如果发现还有数据则进行递归处理
    if (uSize - uIndex != 0)
        onRecv(buffer + uIndex, uSize - uIndex);
}

bool RtspSession::parseHeader() {
    /* OPTIONS rtsp://192.168.114.114:554/live/test RTSP/1.0
       CSeq: 1
       User-Agent: Lavf58.76.100 */
    m_mapHeader.clear();
    std::istringstream stream(m_vecHeader.data());
    stream >> m_mapHeader["reqMethod"];
    stream >> m_mapHeader["reqUrl"];
    stream >> m_mapHeader["reqProtocol"];
    std::string szUrl = m_mapHeader.at("reqUrl");
    size_t uSplitIndex = szUrl.find('?');
    if (std::string::npos != uSplitIndex) {
        m_mapHeader["reqUrl_Short"] = szUrl.substr(0, uSplitIndex);
        m_mapHeader["reqUrl_Args"] = szUrl.substr(uSplitIndex + 1);
    }
    else {
        m_mapHeader["reqUrl_Short"] = szUrl;
    }
    #define RM_R(STR) STR = STR.substr(0, STR.size()-1)
    std::string line;
    while (std::getline(stream, line)) {
        size_t pos = line.find(':');
        if (pos != std::string::npos) {
            std::string key = line.substr(0, pos);
            std::string value = line.substr(pos + 1);
            // 去掉首尾空格
            key.erase(0, key.find_first_not_of(" \t"));
            key.erase(key.find_last_not_of(" \t") + 1);
            value.erase(0, value.find_first_not_of(" \t"));
            value.erase(value.find_last_not_of(" \t") + 1);
            RM_R(value);
            m_mapHeader[key] = value;
        }
    }
    #undef RM_R
    if (0 == m_mapHeader.size()) {
        return false;
    }
    m_iContentLength = atoi(m_mapHeader["Content-Length"].c_str());
    return true;
}

void RtspSession::handleReq() {
    auto method = m_mapHeader["reqMethod"];
    m_iCseq = atoi(m_mapHeader["CSeq"].data());
    if (m_szFullUrl.empty()) {
        m_szFullUrl = m_mapHeader.at("reqUrl");
    }
    using rtsp_request_handler = void(RtspSession::*)(const std::unordered_map<std::string, std::string>& parser);
    static unordered_map<string, rtsp_request_handler> s_cmd_functions{
        {"OPTIONS", &RtspSession::handleReq_Options}, // 拉流推流都会用到
        {"DESCRIBE", &RtspSession::handleReq_Describe}, // 拉流会用到
        {"ANNOUNCE", &RtspSession::handleReq_ANNOUNCE}, // 推流会用到
        {"RECORD", &RtspSession::handleReq_RECORD}, // 推流会用到
        {"SETUP", &RtspSession::handleReq_SETUP}, // 拉流推流都会用到
        {"PLAY", &RtspSession::handleReq_PLAY},
        {"TEARDOWN", &RtspSession::handleReq_TEARDOWN}, // 拉流推流都会用到
        {"GET_PARAMETER", &RtspSession::handleReq_SET_PARAMETER}, // 保活
    };
    auto it = s_cmd_functions.find(method);
    if (it == s_cmd_functions.end()) {
        sendRtspResponse("403 Forbidden");
    }
    else {
        (this->*(it->second))(m_mapHeader);
    }
    m_vecHeader.clear();
    m_vecContent.clear();
    m_iContentLength = 0;
}

void RtspSession::handleRtpPacket() {
    // ("{}字节 rtp 包到达", m_vecRecv.size());
    char* data = m_vecRecv.data();
    size_t uDataLen = m_vecRecv.size();
    // 直接转发
    MediaPusherManager::Instance().SendDataToRecvier(m_szFullUrl, data, uDataLen);
    m_vecRecv.clear();
    m_uRtpPacketLength = 0;
}

void RtspSession::handleReq_Options(const std::unordered_map<std::string, std::string>& parser) {
    /* [21444][2025-12-12 10:38:06.986][13696][RtspSession.cpp:21, onRecv][I] OPTIONS rtsp://192.168.114.114:554/live/test RTSP/1.0
       CSeq: 1
       User-Agent: Lavf58.76.100 */
    // 支持这些命令
    std::multimap<std::string, std::string> header;
    header.emplace("Public", "OPTIONS, DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE, ANNOUNCE, RECORD, SET_PARAMETER, GET_PARAMETER");
    sendRtspResponse("200 OK", header);
}

void RtspSession::handleReq_Describe(const std::unordered_map<std::string, std::string>& parser) {
    m_pPusherRtspSession = dynamic_cast<RtspSession*>(MediaPusherManager::Instance().AddRecvier(m_szFullUrl, this));
    if (nullptr == m_pPusherRtspSession) {
        static constexpr auto err = "该流在服务器上不存在";
        std::multimap<std::string, std::string> header;
        header.emplace("Content-Type", "text/plain");
        sendRtspResponse("406 Not Acceptable", header, err);
        // ("{}:{}", err, m_szFullUrl);
        return;
    }
    m_isPusher = false;
    // ("观看流拉流成功:{}", m_szFullUrl);
    // 找到了相应的 rtsp 流
    m_pSdpParser = new SdpParser(m_pPusherRtspSession->m_pSdpParser->GetSdp());
    m_vecSdpTrack = m_pSdpParser->GetAvailableTrack();
    if (m_vecSdpTrack.empty()) {
        // 该流无效
        static constexpr auto err = "sdp 中无有效 track，该流无效";
        std::multimap<std::string, std::string> header;
        header.emplace("Content-Type", "text/plain");
        sendRtspResponse("406 Not Acceptable", header, err);
        // ("{}:{}", err, m_szFullUrl);
        return;
    }
    m_szSessionId = makeRandStr(12);
    std::multimap<std::string, std::string> header;
    header.emplace("Content-Base", m_szFullUrl + "/");
    header.emplace("x-Accept-Retransmit", "our-retransmit");
    header.emplace("x-Accept-Dynamic-Rate", "1");
    sendRtspResponse("200 OK", header, m_pSdpParser->GetSdp());
}

void RtspSession::handleReq_ANNOUNCE(const std::unordered_map<std::string, std::string>& parser) {
    /* [21444][2025-12-12 10:38:10.129][13696][RtspSession.cpp:21, onRecv][I] ANNOUNCE rtsp://192.168.114.114:554/live/test RTSP/1.0
       Content-Type: application/sdp
       CSeq: 2
       User-Agent: Lavf58.76.100
       Content-Length: 296
       v=0 o=- 0 0 IN IP4 127.0.0.1 s=No Name c=IN IP4 192.168.114.114 t=0 0 a=tool:libavformat 58.76.100
       m=video 0 RTP/AVP 96 a=rtpmap:96 H264/90000 a=fmtp:96 packetization-mode=1; sprop-parameter-sets=Z/QAMpGbKAeAET8TCAAAH0gAB1MAeMGMsA==,aOvjxEhE; profile-level-id=F40032 a=control:streamid=0 */
    m_szFullUrl = parser.at("reqUrl");
    if (false == MediaPusherManager::Instance().AddPusher(m_szFullUrl, this)) {
        static constexpr auto err = "该流在服务器上已存在";
        std::multimap<std::string, std::string> header;
        header.emplace("Content-Type", "text/plain");
        sendRtspResponse("406 Not Acceptable", header, err);
        // ("{}:{}", err, m_szFullUrl);
        return;
    }
    m_isPusher = true;
    // ("流添加成功:{}", m_szFullUrl);
    m_pSdpParser = new SdpParser(parser.at("reqContent"));
    m_szSessionId = makeRandStr(12);
    m_vecSdpTrack = m_pSdpParser->GetAvailableTrack();
    if (m_vecSdpTrack.empty()) {
        static constexpr auto err = "sdp 中无有效 track";
        std::multimap<std::string, std::string> header;
        header.emplace("Content-Type", "text/plain");
        sendRtspResponse("403 Forbidden", header, err);
        // ("{}:{}", err, m_szFullUrl);
        return;
    }
    sendRtspResponse("200 OK");
}

void RtspSession::handleReq_RECORD(const std::unordered_map<std::string, std::string>& parser) {
    if (m_vecSdpTrack.empty() || parser.at("Session") != m_szSessionId) {
        std::multimap<std::string, std::string> header;
        header.emplace("Connection", "Close");
        sendRtspResponse("454 Session Not Found", header);
        // ("{}:{}", "454 Session Not Found", m_szFullUrl);
        return;
    }
    stringstream rtp_info;
    for (auto& track : m_vecSdpTrack) {
        if (track->isInited == false) {
            // 还有 track 没有 setup
            static constexpr auto err = "track not setuped";
            std::multimap<std::string, std::string> header;
            header.emplace("Content-Type", "text/plain");
            sendRtspResponse("403 Forbidden", header, err);
            // ("{}:{}", err, m_mediaTuple.ShortUrl());
            return;
        }
        rtp_info << "url=" << track->GetControlUrl(m_szFullUrl) << ",";
    }
    auto rtpStr = rtp_info.str();
    rtpStr.pop_back();
    std::multimap<std::string, std::string> header;
    header.emplace("RTP-Info", rtpStr);
    sendRtspResponse("200 OK", header);
}

void RtspSession::handleReq_SETUP(const std::unordered_map<std::string, std::string>& parser) {
    /* SETUP rtsp://192.168.114.114:554/live/test/streamid=0 RTSP/1.0
       Transport: RTP/AVP/TCP;unicast;interleaved=0-1;mode=record
       CSeq: 3
       User-Agent: Lavf58.76.100
       Session: uMPQjlcMcI7J *
    */
    // 处理 setup 命令，该函数可能进入多次
    m_szFullUrl = parser.at("reqUrl");
    int trackIdx = getTrackIndexByControlUrl(parser.at("reqUrl"));
    SdpTrack* trackRef = m_vecSdpTrack[trackIdx];
    if (trackRef->isInited) {
        // 已经初始化过该 Track
        static constexpr auto err = "不允许对同一个 track 进行两次 setup";
        std::multimap<std::string, std::string> header;
        header.emplace("Content-Type", "text/plain");
        sendRtspResponse("403 Forbidden", header, err);
        // ("{}:{}", err, m_mediaTuple.ShortUrl());
        return;
    }
    static auto getRtpTypeStr = [](const Rtsp::ERtpType type) {
        switch (type) {
        case Rtsp::ERtpType::TCP: return "TCP";
        case Rtsp::ERtpType::UDP: return "UDP";
        case Rtsp::ERtpType::MULTICAST: return "MULTICAST";
        default: return "Invalid";
        }
    };
    if (m_eRtpType == Rtsp::ERtpType::INVALID) {
        auto& strTransport = parser.at("Transport");
        auto rtpType = Rtsp::ERtpType::INVALID;
        if (strTransport.find("TCP") != string::npos) {
            rtpType = Rtsp::ERtpType::TCP;
        }
        else if (strTransport.find("multicast") != string::npos) {
            // rtpType = Rtsp::ERtpType::MULTICAST;
        }
        else {
            // rtpType = Rtsp::ERtpType::UDP;
        }
        if (Rtsp::ERtpType::INVALID == rtpType) {
            sendRtspResponse("461 Unsupported transport");
            return;
        }
        m_eRtpType = rtpType;
    }
    trackRef->isInited = true;
    // 现在初始化
    {
        // rtsp 推流时，interleaved 由推流者决定
        auto key_values = SdpParser::ParseArgs(parser.at("Transport"), ";", "=");
        int interleaved_rtp = -1, interleaved_rtcp = -1;
        if (2 == sscanf_s(key_values["interleaved"].data(), "%d-%d", &interleaved_rtp, &interleaved_rtcp)) {
            trackRef->uInterleaved = interleaved_rtp;
        }
        else {
            static constexpr auto err = "can not find interleaved when setup of rtp over tcp";
            std::multimap<std::string, std::string> header;
            header.emplace("Content-Type", "text/plain");
            sendRtspResponse("403 Forbidden", header, err);
            // ("{}:{}", err, m_mediaTuple.ShortUrl());
            return;
        }
        stringstream sdpResponse;
        sdpResponse << "RTP/AVP/TCP;unicast;" << "interleaved=" << (int)trackRef->uInterleaved << "-" << (int)trackRef->uInterleaved + 1 << ";" << "ssrc=00000000";
        std::multimap<std::string, std::string> header;
        header.emplace("Transport", sdpResponse.str());
        header.emplace("x-Transport-Options", "late-tolerance=1.400000");
        header.emplace("x-Dynamic-Rate", "1");
        sendRtspResponse("200 OK", header);
    }
}

void RtspSession::handleReq_PLAY(const std::unordered_map<std::string, std::string>& parser) {
    if (m_vecSdpTrack.empty() || parser.at("Session") != m_szSessionId) {
        std::multimap<std::string, std::string> header;
        header.emplace("Connection", "Close");
        sendRtspResponse("454 Session Not Found", header);
        // ("{}:{}", "454 Session Not Found", m_szFullUrl);
        return;
    }
    std::multimap<std::string, std::string> res_header;
    vector<ETrackType> inited_tracks;
    stringstream rtp_info;
    for (auto& track : m_vecSdpTrack) {
        if (track->isInited == false) {
            // 为支持播放器播放单一 track，不校验没有发 setup 的 track
            continue;
        }
        inited_tracks.emplace_back(track->type);
        rtp_info << "url=" << track->GetControlUrl(m_szFullUrl) << ";" << "seq=0;" << "rtptime=0" << ",";
    }
    auto rtpStr = rtp_info.str();
    rtpStr.pop_back();
    res_header.emplace("RTP-Info", rtpStr);
    // 已存在 Range 时不覆盖
    stringstream szPlayRange;
    szPlayRange << "npt=" << setiosflags(ios::fixed) << setprecision(2) << 0;
    res_header.emplace("Range", szPlayRange.str());
    sendRtspResponse("200 OK", res_header);
}

void RtspSession::handleReq_TEARDOWN(const std::unordered_map<std::string, std::string>& parser) {
    sendRtspResponse("200 OK");
    CloseSession();
}

void RtspSession::handleReq_SET_PARAMETER(const std::unordered_map<std::string, std::string>& parser) {
    sendRtspResponse("200 OK");
}

bool RtspSession::sendRtspResponse(const std::string& res_code, std::multimap<std::string, std::string> header, const std::string& sdp, const char* protocol) {
    header.emplace("CSeq", to_string(m_iCseq));
    if (!m_szSessionId.empty()) {
        header.emplace("Session", m_szSessionId);
    }
    header.emplace("Server", "服务器名字");
    header.emplace("Date", dateStr());
    if (!sdp.empty()) {
        header.emplace("Content-Length", to_string(sdp.size()));
        header.emplace("Content-Type", "application/sdp");
    }
    stringstream printer;
    printer << protocol << " " << res_code << "\r\n";
    for (auto& pr : header) {
        printer << pr.first << ": " << pr.second << "\r\n";
    }
    printer << "\r\n";
    if (!sdp.empty()) {
        printer << sdp;
    }
    const int iSendRet = ::send(m_fd, printer.str().data(), printer.str().size(), 0);
    return iSendRet > 0;
}

size_t RtspSession::getTrackIndexByControlUrl(const std::string& control_url) {
    for (size_t i = 0; i < m_vecSdpTrack.size(); ++i) {
        if (control_url.find(m_vecSdpTrack[i]->GetControlUrl(m_szFullUrl)) == 0) {
            return i;
        }
    }
    if (m_vecSdpTrack.size() == 1) {
        return 0;
    }
    DebugBreak();
}

#pragma once
#include <string>
#include <vector>
#include <map>

// 通道类型
enum class ETrackType {
    INVALID = -1, // 无效通道
    VIDEO = 0, // 视频通道
    AUDIO, // 音频通道
    TITLE, // 原始通道
    MIN = INVALID,
    MAX = TITLE,
};

namespace Rtsp {
    // RTSP 的 RTP 类型
    enum class ERtpType {
        INVALID = -1,
        TCP = 0,
        UDP = 1,
        MULTICAST = 2,
    };
}

// 一个通道的具体属性
struct SdpTrack {
    std::multimap<std::string, std::string> map_attr; // 通道属性
    ETrackType type = ETrackType::INVALID; // 这个 track 的类型
    std::string szControl; // 媒体流的控制属性。当存在属性 control 时可用
    uint8_t uInterleaved = 0; // rtp 的通道号
    bool isInited = false; // 该通道是否已经被 setup 过
    std::string GetControlUrl(const std::string& szUrl) const;
};

class SdpParser {
public:
    SdpParser(const std::string& szSdp);
    ~SdpParser();

    // 获取第一个指定类型的通道
    SdpTrack* GetTrack(ETrackType type) const;

    // 取第一个有效的视频通道和音频通道，最多两个元素
    std::vector<SdpTrack*> GetAvailableTrack() const;

    // 获取 sdp 的解析原文
    const std::string GetSdp() const {
        return m_szSdp;
    }

    // 解析参数为 map
    static std::map<std::string, std::string> ParseArgs(const std::string& str, const char* pair_delim, const char* key_delim);

private:
    std::string m_szSdp;
    std::vector<SdpTrack*> m_vecTrack;
};

#include "SdpParser.h"
#include <algorithm>

using namespace std;

static int GetClockRate(int pt) {
    switch (pt) {
    case 0: return 8000; // PCMU
    case 3: return 8000; // GSM
    case 4: return 8000; // G723
    case 5: return 8000; // DVI4_8000
    case 6: return 16000; // DVI4_16000
    case 7: return 8000; // LPC
    case 8: return 8000; // PCMA
    case 9: return 16000; // G722
    case 10: return 44100; // L16_Stereo
    case 11: return 44100; // L16_Mono
    case 12: return 8000; // QCELP
    case 13: return 8000; // CN
    case 14: return 44100; // MP3
    case 15: return 8000; // G728
    case 16: return 11025; // DVI4_11025
    case 17: return 22050; // DVI4_22050
    case 18: return 8000; // G729
    case 25: return 90000; // CelB (视频)
    case 26: return 90000; // JPEG (视频)
    case 28: return 90000; // nv (视频)
    case 31: return 90000; // H261 (视频)
    case 32: return 90000; // MPV (视频)
    case 33: return 90000; // MP2T (视频)
    case 34: return 90000; // H263 (视频)
    default: return 90000; // 默认返回 90000（视频默认采样率/时钟频率）
    }
}

static vector<string> split(const string& s, const char* delim) {
    vector<string> ret;
    size_t last = 0;
    auto index = s.find(delim, last);
    while (index != string::npos) {
        if (index - last > 0) {
            ret.push_back(s.substr(last, index - last));
        }
        last = index + strlen(delim);
        index = s.find(delim, last);
    }
    if (!s.size() || s.size() - last > 0) {
        ret.push_back(s.substr(last));
    }
    return ret;
}

// 去除前后的空格、回车符、制表符
static string& trim(string& s, const string& chars = " \r\n\t") {
    string map(0xFF, '\0');
    for (auto& ch : chars) {
        map[(unsigned char&)ch] = '\1';
    }
    while (s.size() && map.at((unsigned char&)s.back())) s.pop_back();
    while (s.size() && map.at((unsigned char&)s.front())) s.erase(0, 1);
    return s;
}

static string findSubString(const char* buf, const char* start, const char* end, size_t buf_size = 0) {
    if (buf_size <= 0) {
        buf_size = strlen(buf);
    }
    auto msg_start = buf;
    auto msg_end = buf + buf_size;
    size_t len = 0;
    if (start != NULL) {
        len = strlen(start);
        msg_start = strstr(buf, start);
    }
    if (msg_start == NULL) {
        return "";
    }
    msg_start += len;
    if (end != NULL) {
        msg_end = strstr(msg_start, end);
        if (msg_end == NULL) {
            return "";
        }
    }
    return string(msg_start, msg_end);
}

static ETrackType toTrackType(const string& str) {
    if (str == "") {
        return ETrackType::TITLE;
    }
    if (str == "video") {
        return ETrackType::VIDEO;
    }
    if (str == "audio") {
        return ETrackType::AUDIO;
    }
    return ETrackType::INVALID;
}

SdpParser::SdpParser(const std::string& szSdp) : m_szSdp(szSdp) {
    /***** v=0 #SDP 的版本号，目前固定为 0，是唯一的有效值。
       o=- 0 0 IN IP4 127.0.0.1 #o=<用户名：匿名> <会话 ID:默认> <版本号：默认> <网络类型：互联网（IN）> <地址类型：IPv4> <地址：本地回环>
       s=No Name # 会话的名称
       c=IN IP4 192.168.114.114 #媒体流的传输地址
       t=0 0 #表示会话永久有效
       a=tool:libavformat 58.76.100 #生成这份 SDP 的工具是 libavformat（FFmpeg 的核心格式处理库），版本号为 58.76.100
       m=video 0 RTP/AVP 96 #m=<媒体类型> <端口> <传输协议> <编码载荷类型:1-95 为标准载荷，96 + 为自定义>
       a=rtpmap:96 H264/90000 #关联载荷类型和具体编码格式
       a=fmtp:96 packetization-mode=1; sprop-parameter-sets=Z/QAMpGbKAeAET8TCAAAH0gAB1MAeMGMsA==,aOvjxEhE; profile-level-id=F40032
       a=control:streamid=0 *****/
    SdpTrack* track = new SdpTrack;
    track->type = ETrackType::TITLE;
    m_vecTrack.emplace_back(track);
    auto lines = split(m_szSdp, "\n");
    for (auto& line : lines) {
        trim(line);
        if (line.size() < 2 || line[1] != '=') {
            continue;
        }
        char opt = line[0];
        string opt_val = line.substr(2);
        switch (opt) {
        case 'm': {
            track = new SdpTrack;
            m_vecTrack.emplace_back(track);
            int pt, port, port_count;
            char rtp[16] = { 0 }, type[16] = { 0 };
            if (4 == sscanf_s(opt_val.data(), " %15[^ ] %d %15[^ ] %d", type, (unsigned)_countof(type), &port, rtp, (unsigned)_countof(rtp), &pt) ||
                5 == sscanf_s(opt_val.data(), " %15[^ ] %d/%d %15[^ ] %d", type, (unsigned)_countof(type), &port, &port_count, rtp, (unsigned)_countof(rtp), &pt)) {
                track->type = toTrackType(type);
            }
            break;
        }
        case 'a': {
            string attr = findSubString(opt_val.data(), nullptr, ":");
            if (attr.empty()) {
                track->map_attr.emplace(opt_val, "");
            }
            else {
                track->map_attr.emplace(attr, findSubString(opt_val.data(), ":", nullptr));
            }
            break;
        }
        }
    }
    for (auto& track_ptr : m_vecTrack) {
        auto& track = *track_ptr;
        auto it = track.map_attr.find("control");
        if (it != track.map_attr.end()) {
            track.szControl = it->second;
        }
    }
}

SdpParser::~SdpParser() {
    for (auto one : m_vecTrack) delete one;
    m_vecTrack.clear();
}

SdpTrack* SdpParser::GetTrack(ETrackType type) const {
    for (auto& track : m_vecTrack) {
        if (track->type == type) {
            return track;
        }
    }
    return nullptr;
}

std::vector<SdpTrack*> SdpParser::GetAvailableTrack() const {
    vector<SdpTrack*> ret;
    bool audio_added = false;
    bool video_added = false;
    for (auto& track : m_vecTrack) {
        if (track->type == ETrackType::AUDIO) {
            if (!audio_added) {
                ret.emplace_back(track);
                audio_added = true;
            }
            continue;
        }
        if (track->type == ETrackType::VIDEO) {
            if (!video_added) {
                ret.emplace_back(track);
                video_added = true;
            }
            continue;
        }
    }
    return ret;
}

std::map<std::string, std::string> SdpParser::ParseArgs(const string& str, const char* pair_delim, const char* key_delim) {
    map<string, string> ret;
    auto arg_vec = split(str, pair_delim);
    for (auto& key_val : arg_vec) {
        if (key_val.empty()) {
            // 忽略
            continue;
        }
        auto pos = key_val.find(key_delim);
        if (pos != string::npos) {
            std::string key(key_val, 0, pos);
            std::string val(key_val.substr(pos + strlen(key_delim)));
            trim(key);
            trim(val);
            ret.emplace(std::move(key), std::move(val));
        }
        else {
            trim(key_val);
            if (!key_val.empty()) {
                ret.emplace(std::move(key_val), "");
            }
        }
    }
    return ret;
}

std::string SdpTrack::GetControlUrl(const std::string& szUrl) const {
    if (szControl.find("://") != string::npos) {
        // 以 rtsp://开头
        return szControl;
    }
    return szUrl + "/" + szControl;
}

#pragma once
#include <string>
#include <map>
#include <list>
#include <mutex>

class RtspSession; // 具体推流类

class MediaPusher {
public:
    MediaPusher(const std::string& szUrl, RtspSession* pRtspSession);
    ~MediaPusher();

    // 添加一个接收端
    void AddRecvier(RtspSession* pRtspSession);

    // 删除一个接收端
    void DelRecvier(RtspSession* pRtspSession);

    // 发送数据给所有接收端
    bool SendDataToRecvier(const char* pData, size_t uSize);

    // 提供给拉流实例，用于获取推流对应的 tcp 连接中存储的各项信息，比如 sdp
    RtspSession* GetRtspSession() const {
        return m_pRtspSession;
    }

private:
    std::string m_szUrl; // 推流地址
    RtspSession* m_pRtspSession = nullptr; // 推流对应的 tcp 连接
    std::list<RtspSession*> m_listRtspSession;
};

// 全局媒体推流管理类
class MediaPusherManager {
public:
    static MediaPusherManager& Instance() {
        static MediaPusherManager instance;
        return instance;
    }

    // 添加一个推流地址，返回是否添加成功
    bool AddPusher(const std::string& szUrl, RtspSession* pRtspSession);

    // 删除一个推流地址，返回是否删除成功
    bool DelPusher(const std::string& szUrl);

    // 判断推流地址是否存在
    bool IsExist(const std::string& szUrl);

    // 添加一个接收端，返回对应的 tcp 连接 session，失败返回 nullptr
    RtspSession* AddRecvier(const std::string& szUrl, RtspSession* pRtspSession);

    // 删除一个接收端，返回是否删除成功
    bool DelRecvier(const std::string& szUrl, RtspSession* pRtspSession);

    // 发送数据给所有接收端
    bool SendDataToRecvier(const std::string& szUrl, const char* pData, size_t uSize);

private:
    std::map<const std::string, MediaPusher*> m_mapMediaPusher;
    std::mutex m_mtxMediaPusher;
};

#include "MediaPusherManager.h"
#include "RtspSession.h"

bool MediaPusherManager::AddPusher(const std::string& szUrl, RtspSession* pRtspSession) {
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) == m_mapMediaPusher.end()) {
            m_mapMediaPusher[szUrl] = new MediaPusher(szUrl, pRtspSession);
            return true;
        }
    }
    return false;
}

bool MediaPusherManager::DelPusher(const std::string& szUrl) {
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            delete m_mapMediaPusher.at(szUrl);
            m_mapMediaPusher.erase(szUrl);
            return true;
        }
    }
    return false;
}

bool MediaPusherManager::IsExist(const std::string& szUrl) {
    std::lock_guard lck(m_mtxMediaPusher);
    return m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end();
}

RtspSession* MediaPusherManager::AddRecvier(const std::string& szUrl, RtspSession* pRtspSession) {
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            auto pusher = m_mapMediaPusher.at(szUrl);
            pusher->AddRecvier(pRtspSession);
            return pusher->GetRtspSession();
        }
    }
    return nullptr;
}

bool MediaPusherManager::DelRecvier(const std::string& szUrl, RtspSession* pRtspSession) {
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            m_mapMediaPusher.at(szUrl)->DelRecvier(pRtspSession);
            return true;
        }
    }
    return false;
}

bool MediaPusherManager::SendDataToRecvier(const std::string& szUrl, const char* pData, size_t uSize) {
    {
        std::lock_guard lck(m_mtxMediaPusher);
        if (m_mapMediaPusher.find(szUrl) != m_mapMediaPusher.end()) {
            m_mapMediaPusher.at(szUrl)->SendDataToRecvier(pData, uSize);
            return true;
        }
    }
    return false;
}

MediaPusher::MediaPusher(const std::string& szUrl, RtspSession* pRtspSession) : m_szUrl(szUrl), m_pRtspSession(pRtspSession) {}

MediaPusher::~MediaPusher() {}

void MediaPusher::AddRecvier(RtspSession* pRtspSession) {
    m_listRtspSession.push_back(pRtspSession);
}

void MediaPusher::DelRecvier(RtspSession* pRtspSession) {
    m_listRtspSession.remove(pRtspSession);
}

bool MediaPusher::SendDataToRecvier(const char* pData, size_t uSize) {
    decltype(m_listRtspSession) removeList;
    for (auto one : m_listRtspSession) {
        if (false == one->SendData(pData, uSize)) {
            removeList.push_back(one);
        }
    }
    for (auto one : removeList) {
        this->DelRecvier(one);
        one->CloseSession();
    }
    return true;
}

使用 C++ 构建简单的 RTSP 流媒体转发服务器

使用效果举例

更多推荐文章

相关免费在线工具

开篇

过程

代码

1. main.cpp

2. EventPoller.h

3. EventPoller.cpp

4. RtspServer.h

5. RtspServer.cpp

6. RtspSession.h

7. RtspSession.cpp

8. SdpParser.h

9. SdpParser.cpp

10. MediaPusherManager.h

11. MediaPusherManager.cpp

使用 C++ 构建简单的 RTSP 流媒体转发服务器

使用效果举例

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

开篇

过程

代码

1. main.cpp

2. EventPoller.h

3. EventPoller.cpp

4. RtspServer.h

5. RtspServer.cpp

6. RtspSession.h

7. RtspSession.cpp

8. SdpParser.h

9. SdpParser.cpp

10. MediaPusherManager.h

11. MediaPusherManager.cpp