diff --git a/3rdpart/ZLToolKit b/3rdpart/ZLToolKit index 58d0e6a4..0c371464 160000 --- a/3rdpart/ZLToolKit +++ b/3rdpart/ZLToolKit @@ -1 +1 @@ -Subproject commit 58d0e6a4488a290e966b2a0535445dc75af0529c +Subproject commit 0c37146426d8fb0a6d98472739ec3bcc683922db diff --git a/src/Rtp/RtpServer.cpp b/src/Rtp/RtpServer.cpp index 75821c75..6462abac 100644 --- 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 rtcp_server = Socket::createSocket(nullptr, true); + Socket::Ptr rtp_socket = 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; - rtcp_server = pair.second; - } else if (!udp_server->bindUdpSock(local_port, local_ip)) { + rtp_socket = pair.first; + rtcp_socket = pair.second; + } 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(udp_server->getPoller()); + tcp_server = std::make_shared(rtp_socket->getPoller()); (*tcp_server)[RtpSession::kStreamID] = stream_id; - tcp_server->start(udp_server->get_local_port(), local_ip); + (*tcp_server)[RtpSession::kIsUDP] = 0; + tcp_server->start(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(std::move(rtcp_server), 90000); + RtcpHelper::Ptr helper = std::make_shared(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(rtp_socket->getPoller()); + (*udp_server)[RtpSession::kIsUDP] = 1; + udp_server->start(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) { - ref.inputRtp(udp_server, buf->data(), buf->size(), addr); + rtp_socket->setOnRead([&ref, rtp_socket](const Buffer::Ptr &buf, struct sockaddr *addr, int) { + ref.inputRtp(rtp_socket, buf->data(), buf->size(), addr); }); +#endif } - _on_clearup = [udp_server, process, stream_id]() { - //去除循环引用 - udp_server->setOnRead(nullptr); + _on_clearup = [rtp_socket, process, stream_id]() { + if (rtp_socket) { + //去除循环引用 + 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 &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 diff --git a/src/Rtp/RtpServer.h b/src/Rtp/RtpServer.h index 42ac7c6b..14b04b7d 100644 --- a/src/Rtp/RtpServer.h +++ b/src/Rtp/RtpServer.h @@ -15,6 +15,7 @@ #include #include "Network/Socket.h" #include "Network/TcpServer.h" +#include "Network/UdpServer.h" #include "RtpSession.h" using namespace std; @@ -47,18 +48,14 @@ public: */ uint16_t getPort(); - /** - * 获取绑定的线程 - */ - EventPoller::Ptr getPoller(); - /** * 设置RtpProcess onDetach事件回调 */ void setOnDetach(const function &cb); protected: - Socket::Ptr _udp_server; + Socket::Ptr _rtp_socket; + UdpServer::Ptr _udp_server; TcpServer::Ptr _tcp_server; RtpProcess::Ptr _rtp_process; function _on_clearup; diff --git a/src/Rtp/RtpSession.cpp b/src/Rtp/RtpSession.cpp index 868f1b51..880fd6cc 100644 --- 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 Server &server) { _stream_id = const_cast(server)[kStreamID]; + _is_udp = const_cast(server)[kIsUDP]; + + if (_is_udp) { + //设置udp socket读缓存 + SockUtil::setRecvBuf(getSock()->rawFD(), 4 * 1024 * 1024); + _statistic_udp = std::make_shared >(); + } else { + _statistic_tcp = std::make_shared >(); + } } -RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) { +RtpSession::RtpSession(const Socket::Ptr &sock) : Session(sock) { DebugP(this); - socklen_t addr_len = sizeof(addr); - getpeername(sock->rawFD(), &addr, &addr_len); + socklen_t addr_len = sizeof(_addr); + 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 (_search_rtp_finished) { - //下个包开始就是正确的rtp包了 - _search_rtp_finished = false; - _search_rtp = false; + if (!_is_udp) { + if (_search_rtp) { + //搜索上下文期间，数据丢弃 + if (_search_rtp_finished) { + //下个包开始就是正确的rtp包了 + _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(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以便恢复上下文"; - _search_rtp = true; + if (!_is_udp) { + WarnL << ex.what() << "，开始搜索ssrc以便恢复上下文"; + _search_rtp = true; + } else { + throw; + } } catch (...) { throw; } diff --git a/src/Rtp/RtpSession.h b/src/Rtp/RtpSession.h index 9e656d36..44565a58 100644 --- a/src/Rtp/RtpSession.h +++ b/src/Rtp/RtpSession.h @@ -20,9 +20,11 @@ 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; @@ -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 > _statistic_tcp; + std::shared_ptr > _statistic_udp; }; }//namespace mediakit