适配zltoolkit

This commit is contained in:
ziyue 2021-06-08 11:29:32 +08:00
parent 5eb00d5fab
commit 92736db5b2
11 changed files with 25 additions and 22 deletions

@ -1 +1 @@
Subproject commit ca4985c2dc34426363c5ab4eb82f941bcd31cdc2 Subproject commit 58d0e6a4488a290e966b2a0535445dc75af0529c

View File

@ -25,6 +25,7 @@
#include "Http/HttpRequester.h" #include "Http/HttpRequester.h"
#include "Http/HttpSession.h" #include "Http/HttpSession.h"
#include "Network/TcpServer.h" #include "Network/TcpServer.h"
#include "Network/UdpServer.h"
#include "Player/PlayerProxy.h" #include "Player/PlayerProxy.h"
#include "Util/MD5.h" #include "Util/MD5.h"
#include "WebApi.h" #include "WebApi.h"
@ -538,7 +539,7 @@ void installWebApi() {
uint16_t local_port = allArgs["local_port"].as<uint16_t>(); uint16_t local_port = allArgs["local_port"].as<uint16_t>();
string &peer_ip = allArgs["peer_ip"]; 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()){ if(local_port != 0 && local_port != session->get_local_port()){
return; return;
} }
@ -577,8 +578,8 @@ void installWebApi() {
string &peer_ip = allArgs["peer_ip"]; string &peer_ip = allArgs["peer_ip"];
size_t count_hit = 0; 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()){ if(local_port != 0 && local_port != session->get_local_port()){
return; return;
} }
@ -1038,6 +1039,8 @@ void installWebApi() {
val["data"]["TcpServer"] = (Json::UInt64)(ObjectStatistic<TcpServer>::count()); val["data"]["TcpServer"] = (Json::UInt64)(ObjectStatistic<TcpServer>::count());
val["data"]["TcpSession"] = (Json::UInt64)(ObjectStatistic<TcpSession>::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"]["TcpClient"] = (Json::UInt64)(ObjectStatistic<TcpClient>::count());
val["data"]["Socket"] = (Json::UInt64)(ObjectStatistic<Socket>::count()); val["data"]["Socket"] = (Json::UInt64)(ObjectStatistic<Socket>::count());

View File

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

View File

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

View File

@ -17,8 +17,8 @@ namespace mediakit{
const string RtpSession::kStreamID = "stream_id"; const string RtpSession::kStreamID = "stream_id";
void RtpSession::attachServer(const TcpServer &server) { void RtpSession::attachServer(const Server &server) {
_stream_id = const_cast<TcpServer &>(server)[kStreamID]; _stream_id = const_cast<Server &>(server)[kStreamID];
} }
RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) { RtpSession::RtpSession(const Socket::Ptr &sock) : TcpSession(sock) {

View File

@ -28,7 +28,7 @@ public:
void onRecv(const Buffer::Ptr &) override; void onRecv(const Buffer::Ptr &) override;
void onError(const SockException &err) override; void onError(const SockException &err) override;
void onManager() override; void onManager() override;
void attachServer(const TcpServer &server) override; void attachServer(const Server &server) override;
protected: protected:
// 通知其停止推流 // 通知其停止推流

View File

@ -127,7 +127,7 @@ RtpMultiCaster::RtpMultiCaster(SocketHelper &helper, const string &local_ip, con
//组播目标地址 //组播目标地址
peer.sin_addr.s_addr = htonl(*_multicast_ip); peer.sin_addr.s_addr = htonl(*_multicast_ip);
bzero(&(peer.sin_zero), sizeof peer.sin_zero); 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()); _rtp_reader = src->getRing()->attach(helper.getPoller());

View File

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

View File

@ -313,13 +313,13 @@ void RtspPusher::handleResSetup(const Parser &parser, unsigned int track_idx) {
rtpto.sin_port = ntohs(rtp_port); rtpto.sin_port = ntohs(rtp_port);
rtpto.sin_family = AF_INET; rtpto.sin_family = AF_INET;
rtpto.sin_addr.s_addr = inet_addr(get_peer_ip().data()); 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做准备 //设置rtcp发送目标为后续发送rtcp做准备
rtpto.sin_port = ntohs(rtcp_port); rtpto.sin_port = ntohs(rtcp_port);
rtpto.sin_family = AF_INET; rtpto.sin_family = AF_INET;
rtpto.sin_addr.s_addr = inet_addr(get_peer_ip().data()); 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()); auto srcIP = inet_addr(get_peer_ip().data());
weak_ptr<RtspPusher> weakSelf = dynamic_pointer_cast<RtspPusher>(shared_from_this()); weak_ptr<RtspPusher> weakSelf = dynamic_pointer_cast<RtspPusher>(shared_from_this());

View File

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

View File

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