diff --git a/src/Rtp/RtpServer.cpp b/src/Rtp/RtpServer.cpp index 42a4b73f..b53f9254 100644 --- a/src/Rtp/RtpServer.cpp +++ b/src/Rtp/RtpServer.cpp @@ -11,6 +11,7 @@ #if defined(ENABLE_RTPPROXY) #include "RtpServer.h" #include "RtpSelector.h" +#include "Rtcp/RtcpContext.h" namespace mediakit{ RtpServer::RtpServer() { @@ -22,20 +23,86 @@ RtpServer::~RtpServer() { } } +class RtcpHelper : public RtcpContext, public std::enable_shared_from_this { +public: + using Ptr = std::shared_ptr; + + RtcpHelper(Socket::Ptr rtcp_sock, uint32_t sample_rate) : RtcpContext(sample_rate, true){ + _rtcp_sock = std::move(rtcp_sock); + _sample_rate = sample_rate; + } + + void onRecvRtp(const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len){ + //统计rtp接受情况,用于发送rr包 + auto header = (RtpHeader *) buf->data(); + onRtp(ntohs(header->seq), ntohl(header->stamp) * uint64_t(1000) / _sample_rate, buf->size()); + sendRtcp(ntohl(header->ssrc), addr, addr_len); + } + + void startRtcp(){ + weak_ptr weak_self = shared_from_this(); + _rtcp_sock->setOnRead([weak_self](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) { + //用于接受rtcp打洞包 + auto strong_self = weak_self.lock(); + if (!strong_self) { + return; + } + if (!strong_self->_rtcp_addr) { + //只设置一次rtcp对端端口 + strong_self->_rtcp_addr = std::make_shared(); + memcpy(strong_self->_rtcp_addr.get(), addr, addr_len); + } + auto rtcps = RtcpHeader::loadFromBytes(buf->data(), buf->size()); + for (auto &rtcp : rtcps) { + strong_self->onRtcp(rtcp); + } + }); + } + +private: + void sendRtcp(uint32_t rtp_ssrc, struct sockaddr *addr, int addr_len){ + //每5秒发送一次rtcp + if (_ticker.elapsedTime() < 5000) { + return; + } + _ticker.resetTime(); + + auto rtcp_addr = _rtcp_addr.get(); + if (!rtcp_addr) { + //默认的,rtcp端口为rtp端口+1 + ((sockaddr_in *) addr)->sin_port = htons(ntohs(((sockaddr_in *) addr)->sin_port) + 1); + //未收到rtcp打洞包时,采用默认的rtcp端口 + rtcp_addr = addr; + } + _rtcp_sock->send(createRtcpRR(rtp_ssrc + 1, rtp_ssrc), rtcp_addr, addr_len); + } + +private: + Ticker _ticker; + Socket::Ptr _rtcp_sock; + uint32_t _sample_rate; + std::shared_ptr _rtcp_addr; +}; + 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); if (local_port == 0) { //随机端口,rtp端口采用偶数 - Socket::Ptr rtcp_server = Socket::createSocket(nullptr, true); auto pair = std::make_pair(udp_server, rtcp_server); makeSockPair(pair, local_ip); //取偶数端口 udp_server = pair.first; + rtcp_server = pair.second; } else if (!udp_server->bindUdpSock(local_port, local_ip)) { //用户指定端口 - throw std::runtime_error(StrPrinter << "bindUdpSock on " << local_ip << ":" << local_port << " failed:" << get_uv_errmsg(true)); + 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)) { + // 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); @@ -51,7 +118,10 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, bool enable if (!stream_id.empty()) { //指定了流id,那么一个端口一个流(不管是否包含多个ssrc的多个流,绑定rtp源后,会筛选掉ip端口不匹配的流) process = RtpSelector::Instance().getProcess(stream_id, true); - udp_server->setOnRead([udp_server, process](const Buffer::Ptr &buf, struct sockaddr *addr, int) { + RtcpHelper::Ptr helper = std::make_shared(std::move(rtcp_server), 90000); + helper->startRtcp(); + udp_server->setOnRead([udp_server, 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); }); } else {