Merge branch 'master' of https://gitee.com/xia-chu/ZLMediaKit into dev

edf9129f · ziyue · c3354a3d · 511aa220 · 0c371464 · edf9129f
Commit edf9129f authored Jun 08, 2021 by ziyue
--- a/ZLToolKit @ 0c371464
+++ b/ZLToolKit @ 0c371464
-Subproject commit ca4985c2dc34426363c5ab4eb82f941bcd31cdc2
+Subproject commit 0c37146426d8fb0a6d98472739ec3bcc683922db
--- a/server/WebApi.cpp
+++ b/server/WebApi.cpp
@@ -25,6 +25,7 @@
 #include "Http/HttpRequester.h"
 #include "Http/HttpSession.h"
 #include "Network/TcpServer.h"
+#include "Network/UdpServer.h"
 #include "Player/PlayerProxy.h"
 #include "Util/MD5.h"
 #include "WebApi.h"
@@ -569,7 +570,7 @@ void installWebApi() {
        uint16_t local_port = allArgs["local_port"].as<uint16_t>();
        string &peer_ip = allArgs["peer_ip"];
-        SessionMap::Instance().for_each_session([&](const string &id,const TcpSession::Ptr &session){
+        SessionMap::Instance().for_each_session([&](const string &id,const Session::Ptr &session){
            if(local_port != 0 && local_port != session->get_local_port()){
                return;
            }
@@ -608,8 +609,8 @@ void installWebApi() {
        string &peer_ip = allArgs["peer_ip"];
        size_t count_hit = 0;
-        list<TcpSession::Ptr> session_list;
+        list<Session::Ptr> session_list;
-        SessionMap::Instance().for_each_session([&](const string &id,const TcpSession::Ptr &session){
+        SessionMap::Instance().for_each_session([&](const string &id,const Session::Ptr &session){
            if(local_port != 0 && local_port != session->get_local_port()){
                return;
            }
@@ -1069,6 +1070,8 @@ void installWebApi() {
        val["data"]["TcpServer"] = (Json::UInt64)(ObjectStatistic<TcpServer>::count());
        val["data"]["TcpSession"] = (Json::UInt64)(ObjectStatistic<TcpSession>::count());
+        val["data"]["UdpServer"] = (Json::UInt64)(ObjectStatistic<UdpServer>::count());
+        val["data"]["UdpSession"] = (Json::UInt64)(ObjectStatistic<UdpSession>::count());
        val["data"]["TcpClient"] = (Json::UInt64)(ObjectStatistic<TcpClient>::count());
        val["data"]["Socket"] = (Json::UInt64)(ObjectStatistic<Socket>::count());

--- a/src/Http/WebSocketSession.h
+++ b/src/Http/WebSocketSession.h
@@ -104,9 +104,9 @@ public:
        }
    }
-    void attachServer(const TcpServer &server) override{
+    void attachServer(const Server &server) override{
        HttpSessionType::attachServer(server);
-        _weak_server = const_cast<TcpServer &>(server).shared_from_this();
+        _weak_server = const_cast<Server &>(server).shared_from_this();
    }
 protected:
@@ -226,7 +226,7 @@ protected:
 private:
    string _payload_cache;
    string _payload_section;
-    weak_ptr<TcpServer> _weak_server;
+    weak_ptr<Server> _weak_server;
    TcpSession::Ptr _session;
    Creator _creator;
 };

--- a/src/Rtp/RtpSender.cpp
+++ b/src/Rtp/RtpSender.cpp
@@ -54,7 +54,7 @@ void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp,
                cb(local_port, SockException());
                auto strong_self = weak_self.lock();
                if (strong_self) {
-                    strong_self->_socket->setSendPeerAddr(&addr);
+                    strong_self->_socket->bindPeerAddr(&addr);
                    strong_self->onConnect();
                }
            });

--- a/src/Rtp/RtpServer.cpp
+++ b/src/Rtp/RtpServer.cpp
@@ -86,55 +86,69 @@ private:
 void RtpServer::start(uint16_t local_port, const string &stream_id,  bool enable_tcp, const char *local_ip) {
    //创建udp服务器
-    Socket::Ptr udp_server = Socket::createSocket(nullptr, true);
+    Socket::Ptr rtp_socket = Socket::createSocket(nullptr, true);
-    Socket::Ptr rtcp_server = Socket::createSocket(nullptr, true);
+    Socket::Ptr rtcp_socket = Socket::createSocket(nullptr, true);
    if (local_port == 0) {
        //随机端口，rtp端口采用偶数
-        auto pair = std::make_pair(udp_server, rtcp_server);
+        auto pair = std::make_pair(rtp_socket, rtcp_socket);
        makeSockPair(pair, local_ip);
        //取偶数端口
-        udp_server = pair.first;
+        rtp_socket = pair.first;
-        rtcp_server = pair.second;
+        rtcp_socket = pair.second;
-    } else if (!udp_server->bindUdpSock(local_port, local_ip)) {
+    } else if (!rtp_socket->bindUdpSock(local_port, local_ip)) {
        //用户指定端口
        throw std::runtime_error(StrPrinter << "创建rtp端口 " << local_ip << ":" << local_port << " 失败:" << get_uv_errmsg(true));
-    } else if(!rtcp_server->bindUdpSock(udp_server->get_local_port() + 1, local_ip)) {
+    } else if(!rtcp_socket->bindUdpSock(rtp_socket->get_local_port() + 1, local_ip)) {
        // rtcp端口
        throw std::runtime_error(StrPrinter << "创建rtcp端口 " << local_ip << ":" << local_port << " 失败:" << get_uv_errmsg(true));
    }
    //设置udp socket读缓存
-    SockUtil::setRecvBuf(udp_server->rawFD(), 4 * 1024 * 1024);
+    SockUtil::setRecvBuf(rtp_socket->rawFD(), 4 * 1024 * 1024);
    TcpServer::Ptr tcp_server;
    if (enable_tcp) {
        //创建tcp服务器
-        tcp_server = std::make_shared<TcpServer>(udp_server->getPoller());
+        tcp_server = std::make_shared<TcpServer>(rtp_socket->getPoller());
        (*tcp_server)[RtpSession::kStreamID] = stream_id;
-        tcp_server->start<RtpSession>(udp_server->get_local_port(), local_ip);
+        (*tcp_server)[RtpSession::kIsUDP] = 0;
+        tcp_server->start<RtpSession>(rtp_socket->get_local_port(), local_ip);
    }
+    //创建udp服务器
+    UdpServer::Ptr udp_server;
    RtpProcess::Ptr process;
    if (!stream_id.empty()) {
        //指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
+        //由于是一个端口一个流，单线程处理即可
        process = RtpSelector::Instance().getProcess(stream_id, true);
-        RtcpHelper::Ptr helper = std::make_shared<RtcpHelper>(std::move(rtcp_server), 90000);
+        RtcpHelper::Ptr helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), 90000);
        helper->startRtcp();
-        udp_server->setOnRead([udp_server, process, helper](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
+        rtp_socket->setOnRead([rtp_socket, process, helper](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
            helper->onRecvRtp(buf, addr, addr_len);
-            process->inputRtp(true, udp_server, buf->data(), buf->size(), addr);
+            process->inputRtp(true, rtp_socket, buf->data(), buf->size(), addr);
        });
    } else {
-        //未指定流id，一个端口多个流，通过ssrc来分流
+#if 1
+        //单端口多线程接收多个流，根据ssrc区分流
+        udp_server = std::make_shared<UdpServer>(rtp_socket->getPoller());
+        (*udp_server)[RtpSession::kIsUDP] = 1;
+        udp_server->start<RtpSession>(rtp_socket->get_local_port(), local_ip);
+        rtp_socket = nullptr;
+#else
+        //单端口单线程接收多个流
        auto &ref = RtpSelector::Instance();
-        udp_server->setOnRead([&ref, udp_server](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
+        rtp_socket->setOnRead([&ref, rtp_socket](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
-            ref.inputRtp(udp_server, buf->data(), buf->size(), addr);
+            ref.inputRtp(rtp_socket, buf->data(), buf->size(), addr);
        });
+#endif
    }
-    _on_clearup = [udp_server, process, stream_id]() {
+    _on_clearup = [rtp_socket, process, stream_id]() {
-        //去除循环引用
+        if (rtp_socket) {
-        udp_server->setOnRead(nullptr);
+            //去除循环引用
+            rtp_socket->setOnRead(nullptr);
+        }
        if (process) {
            //删除rtp处理器
            RtpSelector::Instance().delProcess(stream_id, process.get());
@@ -143,6 +157,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id,  bool enable
    _tcp_server = tcp_server;
    _udp_server = udp_server;
+    _rtp_socket = rtp_socket;
    _rtp_process = process;
 }
@@ -152,12 +167,8 @@ void RtpServer::setOnDetach(const function<void()> &cb){
    }
 }
-EventPoller::Ptr RtpServer::getPoller() {
-    return _udp_server->getPoller();
-}
 uint16_t RtpServer::getPort() {
-    return _udp_server ? _udp_server->get_local_port() : 0;
+    return _udp_server ? _udp_server->getPort() : _rtp_socket->get_local_port();
 }
 }//namespace mediakit

--- a/src/Rtp/RtpServer.h
+++ b/src/Rtp/RtpServer.h
@@ -15,6 +15,7 @@
 #include <memory>
 #include "Network/Socket.h"
 #include "Network/TcpServer.h"
+#include "Network/UdpServer.h"
 #include "RtpSession.h"
 using namespace std;
@@ -48,17 +49,13 @@ public:
    uint16_t getPort();
    /**
-     * 获取绑定的线程
-     */
-    EventPoller::Ptr getPoller();
-    /**
     * 设置RtpProcess onDetach事件回调
     */
    void setOnDetach(const function<void()> &cb);
 protected:
-    Socket::Ptr _udp_server;
+    Socket::Ptr _rtp_socket;
+    UdpServer::Ptr _udp_server;
    TcpServer::Ptr _tcp_server;
    RtpProcess::Ptr _rtp_process;
    function<void()> _on_clearup;

--- a/src/Rtp/RtpSession.cpp
+++ b/src/Rtp/RtpSession.cpp
@@ -16,16 +16,27 @@
 namespace mediakit{
 const string RtpSession::kStreamID = "stream_id";
+const string RtpSession::kIsUDP = "is_udp";
-void RtpSession::attachServer(const TcpServer &server) {
-    _stream_id = const_cast<TcpServer &>(server)[kStreamID];
+void RtpSession::attachServer(const Server &server) {
+    _stream_id = const_cast<Server &>(server)[kStreamID];
+    _is_udp = const_cast<Server &>(server)[kIsUDP];
+    if (_is_udp) {
+        //设置udp socket读缓存
+        SockUtil::setRecvBuf(getSock()->rawFD(), 4 * 1024 * 1024);
+        _statistic_udp = std::make_shared<ObjectStatistic<UdpSession> >();
+    } else {
+        _statistic_tcp = std::make_shared<ObjectStatistic<TcpSession> >();
+    }
 }
-RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) {
+RtpSession::RtpSession(const Socket::Ptr &sock) : Session(sock) {
    DebugP(this);
-    socklen_t addr_len = sizeof(addr);
+    socklen_t addr_len = sizeof(_addr);
-    getpeername(sock->rawFD(), &addr, &addr_len);
+    getpeername(sock->rawFD(), &_addr, &addr_len);
 }
 RtpSession::~RtpSession() {
    DebugP(this);
    if(_process){
@@ -35,6 +46,10 @@ RtpSession::~RtpSession() {
 void RtpSession::onRecv(const Buffer::Ptr &data) {
    try {
+        if (_is_udp) {
+            onRtpPacket(data->data(), data->size());
+            return;
+        }
        RtpSplitter::input(data->data(), data->size());
    } catch (SockException &ex) {
        shutdown(ex);
@@ -58,19 +73,21 @@ void RtpSession::onManager() {
 }
 void RtpSession::onRtpPacket(const char *data, size_t len) {
-    if (_search_rtp) {
+    if (!_is_udp) {
-        //搜索上下文期间，数据丢弃
+        if (_search_rtp) {
-        if (_search_rtp_finished) {
+            //搜索上下文期间，数据丢弃
-            //下个包开始就是正确的rtp包了
+            if (_search_rtp_finished) {
-            _search_rtp_finished = false;
+                //下个包开始就是正确的rtp包了
-            _search_rtp = false;
+                _search_rtp_finished = false;
+                _search_rtp = false;
+            }
+            return;
+        }
+        if (len > 1024 * 10) {
+            _search_rtp = true;
+            WarnL << "rtp包长度异常(" << len << ")，发送端可能缓存溢出并覆盖，开始搜索ssrc以便恢复上下文";
+            return;
        }
-        return;
-    }
-    if (len > 1024 * 10) {
-        _search_rtp = true;
-        WarnL << "rtp包长度异常(" << len << ")，发送端可能缓存溢出并覆盖，开始搜索ssrc以便恢复上下文";
-        return;
    }
    if (!_process) {
        if (!RtpSelector::getSSRC(data, len, _ssrc)) {
@@ -85,10 +102,14 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
        _process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
    }
    try {
-        _process->inputRtp(false, getSock(), data, len, &addr);
+        _process->inputRtp(false, getSock(), data, len, &_addr);
    } catch (RtpReceiver::BadRtpException &ex) {
-        WarnL << ex.what() << "，开始搜索ssrc以便恢复上下文";
+        if (!_is_udp) {
-        _search_rtp = true;
+            WarnL << ex.what() << "，开始搜索ssrc以便恢复上下文";
+            _search_rtp = true;
+        } else {
+            throw;
+        }
    } catch (...) {
        throw;
    }

--- a/src/Rtp/RtpSession.h
+++ b/src/Rtp/RtpSession.h
@@ -20,15 +20,17 @@ using namespace toolkit;
 namespace mediakit{
-class RtpSession : public TcpSession , public RtpSplitter , public MediaSourceEvent{
+class RtpSession : public Session , public RtpSplitter , public MediaSourceEvent{
 public:
    static const string kStreamID;
+    static const string kIsUDP;
    RtpSession(const Socket::Ptr &sock);
    ~RtpSession() override;
    void onRecv(const Buffer::Ptr &) override;
    void onError(const SockException &err) override;
    void onManager() override;
-    void attachServer(const TcpServer &server) override;
+    void attachServer(const Server &server) override;
 protected:
    // 通知其停止推流
@@ -41,13 +43,16 @@ protected:
    const char *onSearchPacketTail(const char *data, size_t len) override;
 private:
+    bool _is_udp = false;
    bool _search_rtp = false;
    bool _search_rtp_finished = false;
    uint32_t _ssrc = 0;
    Ticker _ticker;
    string _stream_id;
-    struct sockaddr addr;
+    struct sockaddr _addr;
    RtpProcess::Ptr _process;
+    std::shared_ptr<ObjectStatistic<TcpSession> > _statistic_tcp;
+    std::shared_ptr<ObjectStatistic<UdpSession> > _statistic_udp;
 };
 }//namespace mediakit

--- a/src/Rtsp/RtpMultiCaster.cpp
+++ b/src/Rtsp/RtpMultiCaster.cpp
@@ -127,7 +127,7 @@ RtpMultiCaster::RtpMultiCaster(SocketHelper &helper, const string &local_ip, con
        //组播目标地址
        peer.sin_addr.s_addr = htonl(*_multicast_ip);
        bzero(&(peer.sin_zero), sizeof peer.sin_zero);
-        _udp_sock[i]->setSendPeerAddr((struct sockaddr *) &peer);
+        _udp_sock[i]->bindPeerAddr((struct sockaddr *) &peer);
    }
    _rtp_reader = src->getRing()->attach(helper.getPoller());

--- a/src/Rtsp/RtspPlayer.cpp
+++ b/src/Rtsp/RtspPlayer.cpp
@@ -306,7 +306,7 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int track_idx) {
            rtpto.sin_port = ntohs(rtp_port);
            rtpto.sin_family = AF_INET;
            rtpto.sin_addr.s_addr = inet_addr(get_peer_ip().data());
-            pRtpSockRef->setSendPeerAddr((struct sockaddr *)&(rtpto));
+            pRtpSockRef->bindPeerAddr((struct sockaddr *)&(rtpto));
            //发送rtp打洞包
            pRtpSockRef->send("\xce\xfa\xed\xfe", 4);
@@ -314,7 +314,7 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int track_idx) {
            rtpto.sin_port = ntohs(rtcp_port);
            rtpto.sin_family = AF_INET;
            rtpto.sin_addr.s_addr = inet_addr(get_peer_ip().data());
-            pRtcpSockRef->setSendPeerAddr((struct sockaddr *)&(rtpto));
+            pRtcpSockRef->bindPeerAddr((struct sockaddr *)&(rtpto));
        }
        auto srcIP = inet_addr(get_peer_ip().data());

--- a/src/Rtsp/RtspPusher.cpp
+++ b/src/Rtsp/RtspPusher.cpp
@@ -313,13 +313,13 @@ void RtspPusher::handleResSetup(const Parser &parser, unsigned int track_idx) {
        rtpto.sin_port = ntohs(rtp_port);
        rtpto.sin_family = AF_INET;
        rtpto.sin_addr.s_addr = inet_addr(get_peer_ip().data());
-        rtp_sock->setSendPeerAddr((struct sockaddr *) &(rtpto));
+        rtp_sock->bindPeerAddr((struct sockaddr *) &(rtpto));
        //设置rtcp发送目标，为后续发送rtcp做准备
        rtpto.sin_port = ntohs(rtcp_port);
        rtpto.sin_family = AF_INET;
        rtpto.sin_addr.s_addr = inet_addr(get_peer_ip().data());
-        rtcp_sock->setSendPeerAddr((struct sockaddr *)&(rtpto));
+        rtcp_sock->bindPeerAddr((struct sockaddr *)&(rtpto));
        auto srcIP = inet_addr(get_peer_ip().data());
        weak_ptr<RtspPusher> weakSelf = dynamic_pointer_cast<RtspPusher>(shared_from_this());

--- a/src/Rtsp/RtspSession.cpp
+++ b/src/Rtsp/RtspSession.cpp
@@ -686,14 +686,14 @@ void RtspSession::handleReq_Setup(const Parser &parser) {
        peerAddr.sin_port = htons(ui16RtpPort);
        peerAddr.sin_addr.s_addr = inet_addr(get_peer_ip().data());
        bzero(&(peerAddr.sin_zero), sizeof peerAddr.sin_zero);
-        pr.first->setSendPeerAddr((struct sockaddr *) (&peerAddr));
+        pr.first->bindPeerAddr((struct sockaddr *) (&peerAddr));
        //设置rtcp发送目标地址
        peerAddr.sin_family = AF_INET;
        peerAddr.sin_port = htons(ui16RtcpPort);
        peerAddr.sin_addr.s_addr = inet_addr(get_peer_ip().data());
        bzero(&(peerAddr.sin_zero), sizeof peerAddr.sin_zero);
-        pr.second->setSendPeerAddr((struct sockaddr *) (&peerAddr));
+        pr.second->bindPeerAddr((struct sockaddr *) (&peerAddr));
        //尝试获取客户端nat映射地址
        startListenPeerUdpData(trackIdx);
@@ -925,13 +925,13 @@ void RtspSession::onRcvPeerUdpData(int interleaved, const Buffer::Ptr &buf, cons
        } else if (!_udp_connected_flags.count(interleaved)) {
            //这是rtsp播放器的rtp打洞包
            _udp_connected_flags.emplace(interleaved);
-            _rtp_socks[interleaved / 2]->setSendPeerAddr(&addr);
+            _rtp_socks[interleaved / 2]->bindPeerAddr(&addr);
        }
    } else {
        //rtcp包
        if (!_udp_connected_flags.count(interleaved)) {
            _udp_connected_flags.emplace(interleaved);
-            _rtcp_socks[(interleaved - 1) / 2]->setSendPeerAddr(&addr);
+            _rtcp_socks[(interleaved - 1) / 2]->bindPeerAddr(&addr);
        }
        onRtcpPacket((interleaved - 1) / 2, _sdp_track[(interleaved - 1) / 2], buf->data(), buf->size());
    }

--- a/tests/test_wsServer.cpp
+++ b/tests/test_wsServer.cpp
@@ -30,7 +30,7 @@ public:
        DebugL;
    }
-    void attachServer(const TcpServer &server) override{
+    void attachServer(const Server &server) override{
        DebugL << getIdentifier() << " " << TcpSession::getIdentifier();
    }
    void onRecv(const Buffer::Ptr &buffer) override {
@@ -57,7 +57,7 @@ public:
        DebugL;
    }
-    void attachServer(const TcpServer &server) override{
+    void attachServer(const Server &server) override{
        DebugL << getIdentifier() << " " << TcpSession::getIdentifier();
    }
    void onRecv(const Buffer::Ptr &buffer) override {