From 2ee7f67b5b386e23553694604a6baf2e041aef1d Mon Sep 17 00:00:00 2001 From: baiyfcu Date: Mon, 31 May 2021 16:51:43 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dwindows=20set=20FD=201024?= =?UTF-8?q?=E6=9C=AA=E7=94=9F=E6=95=88BUG#898?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 3rdpart/ZLToolKit | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/3rdpart/ZLToolKit b/3rdpart/ZLToolKit index ca4985c2..7aed59ec 160000 --- a/3rdpart/ZLToolKit +++ b/3rdpart/ZLToolKit @@ -1 +1 @@ -Subproject commit ca4985c2dc34426363c5ab4eb82f941bcd31cdc2 +Subproject commit 7aed59ecf3da8a710dd04dc4f19c95a07be3e58d From 92736db5b2835c3f631ce14dc2d83f98e16da9c3 Mon Sep 17 00:00:00 2001 From: ziyue <1213642868@qq.com> Date: Tue, 8 Jun 2021 11:29:32 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E9=80=82=E9=85=8Dzltoolkit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 3rdpart/ZLToolKit | 2 +- server/WebApi.cpp | 9 ++++++--- src/Http/WebSocketSession.h | 6 +++--- src/Rtp/RtpSender.cpp | 2 +- src/Rtp/RtpSession.cpp | 4 ++-- src/Rtp/RtpSession.h | 2 +- src/Rtsp/RtpMultiCaster.cpp | 2 +- src/Rtsp/RtspPlayer.cpp | 4 ++-- src/Rtsp/RtspPusher.cpp | 4 ++-- src/Rtsp/RtspSession.cpp | 8 ++++---- tests/test_wsServer.cpp | 4 ++-- 11 files changed, 25 insertions(+), 22 deletions(-) diff --git a/3rdpart/ZLToolKit b/3rdpart/ZLToolKit index ca4985c2..58d0e6a4 160000 --- a/3rdpart/ZLToolKit +++ b/3rdpart/ZLToolKit @@ -1 +1 @@ -Subproject commit ca4985c2dc34426363c5ab4eb82f941bcd31cdc2 +Subproject commit 58d0e6a4488a290e966b2a0535445dc75af0529c diff --git a/server/WebApi.cpp b/server/WebApi.cpp index cbb2f6d9..2ccb077b 100755 --- 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" @@ -538,7 +539,7 @@ void installWebApi() { uint16_t local_port = allArgs["local_port"].as(); 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; } @@ -577,8 +578,8 @@ void installWebApi() { string &peer_ip = allArgs["peer_ip"]; size_t count_hit = 0; - list session_list; - SessionMap::Instance().for_each_session([&](const string &id,const TcpSession::Ptr &session){ + list session_list; + SessionMap::Instance().for_each_session([&](const string &id,const Session::Ptr &session){ if(local_port != 0 && local_port != session->get_local_port()){ return; } @@ -1038,6 +1039,8 @@ void installWebApi() { val["data"]["TcpServer"] = (Json::UInt64)(ObjectStatistic::count()); val["data"]["TcpSession"] = (Json::UInt64)(ObjectStatistic::count()); + val["data"]["UdpServer"] = (Json::UInt64)(ObjectStatistic::count()); + val["data"]["UdpSession"] = (Json::UInt64)(ObjectStatistic::count()); val["data"]["TcpClient"] = (Json::UInt64)(ObjectStatistic::count()); val["data"]["Socket"] = (Json::UInt64)(ObjectStatistic::count()); diff --git a/src/Http/WebSocketSession.h b/src/Http/WebSocketSession.h index d6fb6d4a..03872abf 100644 --- 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(server).shared_from_this(); + _weak_server = const_cast(server).shared_from_this(); } protected: @@ -226,7 +226,7 @@ protected: private: string _payload_cache; string _payload_section; - weak_ptr _weak_server; + weak_ptr _weak_server; TcpSession::Ptr _session; Creator _creator; }; diff --git a/src/Rtp/RtpSender.cpp b/src/Rtp/RtpSender.cpp index 0ca406b1..a0079801 100644 --- 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(); } }); diff --git a/src/Rtp/RtpSession.cpp b/src/Rtp/RtpSession.cpp index 4eb79120..868f1b51 100644 --- a/src/Rtp/RtpSession.cpp +++ b/src/Rtp/RtpSession.cpp @@ -17,8 +17,8 @@ namespace mediakit{ const string RtpSession::kStreamID = "stream_id"; -void RtpSession::attachServer(const TcpServer &server) { - _stream_id = const_cast(server)[kStreamID]; +void RtpSession::attachServer(const Server &server) { + _stream_id = const_cast(server)[kStreamID]; } RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) { diff --git a/src/Rtp/RtpSession.h b/src/Rtp/RtpSession.h index 2999b2b9..9e656d36 100644 --- a/src/Rtp/RtpSession.h +++ b/src/Rtp/RtpSession.h @@ -28,7 +28,7 @@ public: 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: // 通知其停止推流 diff --git a/src/Rtsp/RtpMultiCaster.cpp b/src/Rtsp/RtpMultiCaster.cpp index 2b2f7eba..36cef4fd 100644 --- 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()); diff --git a/src/Rtsp/RtspPlayer.cpp b/src/Rtsp/RtspPlayer.cpp index f6c461fa..f3b7004f 100644 --- 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()); diff --git a/src/Rtsp/RtspPusher.cpp b/src/Rtsp/RtspPusher.cpp index c5fb253f..9ce2d56b 100644 --- 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 weakSelf = dynamic_pointer_cast(shared_from_this()); diff --git a/src/Rtsp/RtspSession.cpp b/src/Rtsp/RtspSession.cpp index 40124477..c14a7978 100644 --- 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()); } diff --git a/tests/test_wsServer.cpp b/tests/test_wsServer.cpp index 6590baae..c8e802b5 100644 --- 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 { From c4817c6d5f8f28b8b4c1287c0ff2b11fb37b6c11 Mon Sep 17 00:00:00 2001 From: ziyue <1213642868@qq.com> Date: Tue, 8 Jun 2021 14:03:25 +0800 Subject: [PATCH 3/3] =?UTF-8?q?GB28181=E5=8D=95=E7=AB=AF=E5=8F=A3=E6=8E=A5?= =?UTF-8?q?=E6=94=B6=E6=B5=81=E6=94=AF=E6=8C=81=E5=A4=9A=E7=BA=BF=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 3rdpart/ZLToolKit | 2 +- src/Rtp/RtpServer.cpp | 59 +++++++++++++++++++++++++----------------- src/Rtp/RtpServer.h | 9 +++---- src/Rtp/RtpSession.cpp | 57 +++++++++++++++++++++++++++------------- src/Rtp/RtpSession.h | 9 +++++-- 5 files changed, 85 insertions(+), 51 deletions(-) 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