Compare commits
5 Commits
Author | SHA1 | Date | |
---|---|---|---|
![]() |
7eb242f5b8 | ||
![]() |
571a29eebb | ||
![]() |
70e076631c | ||
![]() |
39b01b2794 | ||
![]() |
36f7099354 |
@ -24,6 +24,7 @@ message SS_CMKcpHandshake
|
||||
optional string account_id = 2; //账号id
|
||||
optional string session_id = 3; //session id
|
||||
optional string team_uuid = 4; //保留
|
||||
optional int32 secret_key_place = 5; //私钥存放位置 0:存在用户协议前(老) 1:存在kcp底层协议头之后(新)
|
||||
}
|
||||
|
||||
message SS_SMKcpHandshake
|
||||
|
@ -25,7 +25,6 @@
|
||||
#include "upstreammgr.h"
|
||||
#include "master.h"
|
||||
#include "mastermgr.h"
|
||||
#include "mmcache.h"
|
||||
|
||||
#include "longsessionmgr.h"
|
||||
|
||||
@ -116,10 +115,10 @@ bool App::Init(int argc, char* argv[])
|
||||
});
|
||||
|
||||
uuid_ = std::make_shared<a8::uuid::SnowFlake>();
|
||||
loop_mutex_ = std::make_shared<std::mutex>();
|
||||
loop_cond_ = std::make_shared<std::condition_variable>();
|
||||
msg_mutex_ = std::make_shared<std::mutex>();
|
||||
udp_msg_mutex_ = std::make_shared<std::mutex>();
|
||||
loop_mutex_ = new std::mutex();
|
||||
loop_cond_ = new std::condition_variable();
|
||||
msg_mutex_ = new std::mutex();
|
||||
udp_msg_mutex_ = new std::mutex();
|
||||
|
||||
srand(time(nullptr));
|
||||
InitLog();
|
||||
@ -128,7 +127,6 @@ bool App::Init(int argc, char* argv[])
|
||||
f8::Timer::Instance()->Init();
|
||||
JsonDataMgr::Instance()->Init();
|
||||
uuid_->SetMachineId((node_id_ - 1) * MAX_NODE_ID + instance_id_);
|
||||
MMCache::Instance()->Init();
|
||||
DownStreamMgr::Instance()->Init();
|
||||
MasterMgr::Instance()->Init();
|
||||
UpStreamMgr::Instance()->Init();
|
||||
@ -191,7 +189,6 @@ void App::UnInit()
|
||||
MasterMgr::Instance()->UnInit();
|
||||
UpStreamMgr::Instance()->UnInit();
|
||||
DownStreamMgr::Instance()->UnInit();
|
||||
MMCache::Instance()->UnInit();
|
||||
JsonDataMgr::Instance()->UnInit();
|
||||
f8::Timer::Instance()->UnInit();
|
||||
f8::MsgQueue::Instance()->UnInit();
|
||||
@ -200,6 +197,14 @@ void App::UnInit()
|
||||
|
||||
FreeSocketMsgQueue();
|
||||
FreeUdpMsgQueue();
|
||||
delete msg_mutex_;
|
||||
msg_mutex_ = nullptr;
|
||||
delete udp_msg_mutex_;
|
||||
udp_msg_mutex_ = nullptr;
|
||||
delete loop_cond_;
|
||||
loop_cond_ = nullptr;
|
||||
delete loop_mutex_;
|
||||
loop_mutex_ = nullptr;
|
||||
a8::XPrintf("wsproxy terminated instance_id:%d pid:%d\n", {instance_id_, getpid()});
|
||||
}
|
||||
|
||||
@ -229,8 +234,7 @@ void App::AddSocketMsg(SocketFrom_e sockfrom,
|
||||
int bodylen,
|
||||
int tag)
|
||||
{
|
||||
char* buf = MMCache::Instance()->AllocSocketMsg(sizeof(MsgNode) + bodylen);
|
||||
MsgNode *p = (MsgNode*)buf;
|
||||
MsgNode *p = (MsgNode*)malloc(sizeof(MsgNode));
|
||||
memset(p, 0, sizeof(MsgNode));
|
||||
p->sockfrom = sockfrom;
|
||||
p->ip_saddr = ip_saddr;
|
||||
@ -241,7 +245,7 @@ void App::AddSocketMsg(SocketFrom_e sockfrom,
|
||||
p->buflen = bodylen;
|
||||
p->tag = tag;
|
||||
if (bodylen > 0) {
|
||||
p->buf = buf + sizeof(MsgNode);
|
||||
p->buf = (char*)malloc(bodylen);
|
||||
memmove(p->buf, msgbody, bodylen);
|
||||
}
|
||||
msg_mutex_->lock();
|
||||
@ -366,7 +370,10 @@ void App::DispatchMsg()
|
||||
}
|
||||
break;
|
||||
}
|
||||
MMCache::Instance()->FreeSocketMsg((char*)pdelnode);
|
||||
if (pdelnode->buf) {
|
||||
free(pdelnode->buf);
|
||||
}
|
||||
free(pdelnode);
|
||||
working_msgnode_size_--;
|
||||
if (a8::XGetTickCount() - starttick > 200) {
|
||||
break;
|
||||
@ -540,7 +547,10 @@ void App::FreeSocketMsgQueue()
|
||||
while (work_node_) {
|
||||
MsgNode* pdelnode = work_node_;
|
||||
work_node_ = work_node_->next;
|
||||
MMCache::Instance()->FreeSocketMsg((char*)pdelnode);
|
||||
if (pdelnode->buf) {
|
||||
free(pdelnode->buf);
|
||||
}
|
||||
free(pdelnode);
|
||||
if (!work_node_) {
|
||||
work_node_ = top_node_;
|
||||
top_node_ = nullptr;
|
||||
@ -566,7 +576,7 @@ void App::FreeUdpMsgQueue()
|
||||
free((void*)pdelnode->pkt->buf);
|
||||
}
|
||||
delete pdelnode->pkt;
|
||||
MMCache::Instance()->FreeUdpMsg((char*)pdelnode);
|
||||
free(pdelnode);
|
||||
}
|
||||
if (!udp_work_node_) {
|
||||
udp_work_node_ = udp_top_node_;
|
||||
@ -579,8 +589,7 @@ void App::FreeUdpMsgQueue()
|
||||
|
||||
void App::AddUdpMsg(a8::UdpPacket* pkt)
|
||||
{
|
||||
char* buf = MMCache::Instance()->AllocUdpMsg(sizeof(UdpMsgNode));
|
||||
UdpMsgNode *p = (UdpMsgNode*)buf;
|
||||
UdpMsgNode *p = (UdpMsgNode*) malloc(sizeof(UdpMsgNode));
|
||||
memset(p, 0, sizeof(UdpMsgNode));
|
||||
p->pkt = pkt;
|
||||
udp_msg_mutex_->lock();
|
||||
@ -617,7 +626,7 @@ void App::DispatchUdpMsg()
|
||||
free((void*)pdelnode->pkt->buf);
|
||||
}
|
||||
delete pdelnode->pkt;
|
||||
MMCache::Instance()->FreeUdpMsg((char*)pdelnode);
|
||||
free(pdelnode);
|
||||
}
|
||||
udp_working_msgnode_size_--;
|
||||
if (a8::XGetTickCount() - starttick > 200) {
|
||||
|
@ -78,15 +78,15 @@ private:
|
||||
std::set<int> flags_;
|
||||
|
||||
std::shared_ptr<a8::uuid::SnowFlake> uuid_;
|
||||
std::shared_ptr<std::mutex> loop_mutex_;
|
||||
std::shared_ptr<std::condition_variable> loop_cond_;
|
||||
std::mutex *loop_mutex_ = nullptr;
|
||||
std::condition_variable *loop_cond_ = nullptr;
|
||||
|
||||
std::shared_ptr<std::mutex> msg_mutex_;
|
||||
std::mutex* msg_mutex_ = nullptr;
|
||||
MsgNode* top_node_ = nullptr;
|
||||
MsgNode* bot_node_ = nullptr;
|
||||
MsgNode* work_node_ = nullptr;
|
||||
|
||||
std::shared_ptr<std::mutex> udp_msg_mutex_;
|
||||
std::mutex* udp_msg_mutex_ = nullptr;
|
||||
UdpMsgNode* udp_top_node_ = nullptr;
|
||||
UdpMsgNode* udp_bot_node_ = nullptr;
|
||||
UdpMsgNode* udp_work_node_ = nullptr;
|
||||
|
@ -37,5 +37,3 @@ const int MAX_NODE_ID = 8;
|
||||
const int MAX_INSTANCE_ID = 500;
|
||||
|
||||
const int ALLOC_TARGET_SERVER_SUCCESS_TIMER_EVENT = a8::TIMER_USER_EVENT + 1;
|
||||
|
||||
const int PACK_MAX = 1024 * 64;
|
||||
|
@ -25,7 +25,7 @@ void DownStream::ReBindUpStream(std::weak_ptr<UpStream> up)
|
||||
|
||||
void DownStream::ForwardUpStreamMsg(f8::MsgHdr& hdr)
|
||||
{
|
||||
char* buff = forward_buff_.data();
|
||||
char* buff = (char*)malloc(sizeof(f8::PackHead) + hdr.buflen);
|
||||
f8::PackHead* head = (f8::PackHead*)buff;
|
||||
head->packlen = hdr.buflen;
|
||||
head->msgid = hdr.msgid;
|
||||
@ -41,6 +41,10 @@ void DownStream::ForwardUpStreamMsg(f8::MsgHdr& hdr)
|
||||
ss::SS_SMPing msg;
|
||||
msg.set_source(1);
|
||||
{
|
||||
free(buff);
|
||||
|
||||
buff = (char*)malloc(sizeof(f8::PackHead) + msg.ByteSize());
|
||||
f8::PackHead* head = (f8::PackHead*)buff;
|
||||
head->packlen = msg.ByteSize();
|
||||
head->msgid = hdr.msgid;
|
||||
head->seqid = hdr.seqid;
|
||||
@ -55,6 +59,7 @@ void DownStream::ForwardUpStreamMsg(f8::MsgHdr& hdr)
|
||||
} else {
|
||||
GCListener::Instance()->SendBuf(hdr.socket_handle, buff, sizeof(f8::PackHead) + head->packlen);
|
||||
}
|
||||
free(buff);
|
||||
}
|
||||
|
||||
void DownStream::OnClose()
|
||||
|
@ -18,7 +18,6 @@ class DownStream
|
||||
void OnClose();
|
||||
|
||||
private:
|
||||
std::array<char, PACK_MAX> forward_buff_;
|
||||
int socket_handle_ = a8::INVALID_SOCKET_HANDLE;
|
||||
std::weak_ptr<UpStream> up_;
|
||||
bool is_long_session_ = false;
|
||||
|
@ -1,6 +1,7 @@
|
||||
#include "precompile.h"
|
||||
|
||||
#include <memory.h>
|
||||
#include <string.h>
|
||||
|
||||
#include <f8/netmsghandler.h>
|
||||
#include <f8/udplog.h>
|
||||
@ -33,23 +34,22 @@ KcpSession::~KcpSession()
|
||||
|
||||
}
|
||||
|
||||
void KcpSession::Init(int socket_handle)
|
||||
void KcpSession::Init(int socket_handle, int secret_key_place)
|
||||
{
|
||||
socket_handle_ = socket_handle;
|
||||
secret_key_ = App::Instance()->NewUuid();
|
||||
kcp_ = ikcp_create(socket_handle_, (void*)this);
|
||||
#if 1
|
||||
const KcpConf& kcp_conf = JsonDataMgr::Instance()->GetKcpConf();
|
||||
ikcp_wndsize(kcp_, kcp_conf.sndwnd, kcp_conf.rcvwnd);
|
||||
ikcp_nodelay(kcp_, kcp_conf.nodelay, kcp_conf.interval, kcp_conf.resend, kcp_conf.nc);
|
||||
kcp_->rx_minrto = kcp_conf.rx_minrto;
|
||||
kcp_->fastresend = kcp_conf.fastresend;
|
||||
#else
|
||||
ikcp_wndsize(kcp_, 128, 128);
|
||||
ikcp_nodelay(kcp_, 2, 10, 2, 1);
|
||||
kcp_->rx_minrto = 10;
|
||||
kcp_->fastresend = 1;
|
||||
#endif
|
||||
secret_key_place_ = secret_key_place;
|
||||
if (secret_key_place_ > 0) {
|
||||
secret_key_place_ = 1;
|
||||
} else if (secret_key_place_ < 0) {
|
||||
secret_key_place_ = 0;
|
||||
}
|
||||
kcp_->output = UdpOutput;
|
||||
init_tick_ = a8::XGetTickCount();
|
||||
}
|
||||
@ -75,8 +75,21 @@ void KcpSession::SendClientMsg(char* buf, int buf_len)
|
||||
|
||||
void KcpSession::OnRecvPacket(a8::UdpPacket* pkt)
|
||||
{
|
||||
const int IKCP_OVERHEAD = 24;
|
||||
|
||||
remote_addr_ = pkt->remote_addr;
|
||||
if (GetSecretKeyPlace()) {
|
||||
char* buf = (char*)malloc(pkt->buf_len - GetSecretKeyLen());
|
||||
int buflen = pkt->buf_len - GetSecretKeyLen();
|
||||
memmove(buf, pkt->buf, IKCP_OVERHEAD);
|
||||
if (pkt->buf_len > IKCP_OVERHEAD + GetSecretKeyLen()) {
|
||||
memmove(buf + IKCP_OVERHEAD, pkt->buf + IKCP_OVERHEAD + GetSecretKeyLen(), buflen - IKCP_OVERHEAD);
|
||||
}
|
||||
ikcp_input(kcp_, buf, buflen);
|
||||
free(buf);
|
||||
} else {
|
||||
ikcp_input(kcp_, pkt->buf, pkt->buf_len);
|
||||
}
|
||||
}
|
||||
|
||||
void KcpSession::UpdateInput()
|
||||
@ -91,8 +104,20 @@ void KcpSession::UpdateInput()
|
||||
|
||||
void KcpSession::DecodeUserPacket(char* buf, int& offset, unsigned int buflen)
|
||||
{
|
||||
//packagelen + msgid + magiccode + msgbody
|
||||
//2 + 2 + 4+ xx + \0 + xx
|
||||
if (GetSecretKeyPlace()) {
|
||||
DecodeUserPacketNew(buf, offset, buflen);
|
||||
} else {
|
||||
DecodeUserPacketOld(buf, offset, buflen);
|
||||
}
|
||||
}
|
||||
|
||||
int KcpSession::GetSecretKeyPlace()
|
||||
{
|
||||
return secret_key_place_;
|
||||
}
|
||||
|
||||
void KcpSession::DecodeUserPacketOld(char* buf, int& offset, unsigned int buflen)
|
||||
{
|
||||
bool warning = false;
|
||||
while (buflen - offset >= sizeof(f8::PackHead) + GetSecretKeyLen()) {
|
||||
long long secret_key = KcpSession::ReadSecretKey(&buf[offset], buflen);
|
||||
@ -124,6 +149,37 @@ void KcpSession::DecodeUserPacket(char* buf, int& offset, unsigned int buflen)
|
||||
}
|
||||
|
||||
if (warning) {
|
||||
f8::UdpLog::Instance()->Warning("收到kcp client非法数据包", {});
|
||||
f8::UdpLog::Instance()->Warning("收到kcp client非法数据包1", {});
|
||||
}
|
||||
}
|
||||
|
||||
void KcpSession::DecodeUserPacketNew(char* buf, int& offset, unsigned int buflen)
|
||||
{
|
||||
bool warning = false;
|
||||
while (buflen - offset >= sizeof(f8::PackHead)) {
|
||||
f8::PackHead* p = (f8::PackHead*)&buf[offset];
|
||||
if (p->magic_code == f8::MAGIC_CODE) {
|
||||
if (buflen - offset < sizeof(f8::PackHead) + p->packlen) {
|
||||
break;
|
||||
}
|
||||
App::Instance()->AddSocketMsg(SF_Client,
|
||||
socket_handle_,
|
||||
0,
|
||||
//saddr,
|
||||
p->msgid,
|
||||
p->seqid,
|
||||
&buf[offset + sizeof(f8::PackHead)],
|
||||
p->packlen,
|
||||
ST_Udp);
|
||||
offset += sizeof(f8::PackHead) + p->packlen;
|
||||
} else {
|
||||
warning = true;
|
||||
offset++;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if (warning) {
|
||||
f8::UdpLog::Instance()->Warning("收到kcp client非法数据包2", {});
|
||||
}
|
||||
}
|
||||
|
@ -11,7 +11,7 @@ public:
|
||||
KcpSession();
|
||||
~KcpSession();
|
||||
|
||||
void Init(int socket_handle);
|
||||
void Init(int socket_handle, int secret_key_place);
|
||||
void UnInit();
|
||||
void Update(long long tick);
|
||||
|
||||
@ -19,6 +19,7 @@ public:
|
||||
int GetSocketHandle() { return socket_handle_; }
|
||||
long long GetSecretKey() { return secret_key_; }
|
||||
void* GetSecretKeyDataPtr() { return &secret_key_; }
|
||||
int GetSecretKeyPlace();
|
||||
|
||||
void SendClientMsg(char* buf, int buf_len);
|
||||
virtual void OnRecvPacket(a8::UdpPacket* pkt) override;
|
||||
@ -31,11 +32,14 @@ public:
|
||||
|
||||
protected:
|
||||
virtual void DecodeUserPacket(char* buf, int& offset, unsigned int buflen) override;
|
||||
void DecodeUserPacketOld(char* buf, int& offset, unsigned int buflen);
|
||||
void DecodeUserPacketNew(char* buf, int& offset, unsigned int buflen);
|
||||
|
||||
private:
|
||||
void UpdateInput();
|
||||
|
||||
private:
|
||||
int secret_key_place_ = 0;
|
||||
int socket_handle_ = 0;
|
||||
long long secret_key_ = 0;
|
||||
ikcpcb* kcp_ = nullptr;
|
||||
|
@ -5,10 +5,13 @@
|
||||
#include "longsession.h"
|
||||
#include "kcpsession.h"
|
||||
|
||||
#include "ss_msgid.pb.h"
|
||||
#include "ss_proto.pb.h"
|
||||
|
||||
void LongSession::Init(f8::MsgHdr& hdr, const ss::SS_CMKcpHandshake& msg)
|
||||
{
|
||||
kcp_session_ = std::make_shared<KcpSession>();
|
||||
kcp_session_->Init(hdr.socket_handle);
|
||||
kcp_session_->Init(hdr.socket_handle, msg.secret_key_place());
|
||||
}
|
||||
|
||||
void LongSession::UnInit()
|
||||
@ -25,3 +28,8 @@ void LongSession::UpdatePing()
|
||||
{
|
||||
last_ping_tick_ = a8::XGetTickCount();
|
||||
}
|
||||
|
||||
int LongSession::GetSecretKeyPlace()
|
||||
{
|
||||
return kcp_session_->GetSecretKeyPlace();
|
||||
}
|
||||
|
@ -17,6 +17,7 @@ class LongSession
|
||||
|
||||
std::shared_ptr<KcpSession> GetKcpSession() { return kcp_session_; }
|
||||
void UpdatePing();
|
||||
int GetSecretKeyPlace();
|
||||
|
||||
private:
|
||||
long long last_ping_tick_ = 0;
|
||||
|
@ -133,9 +133,8 @@ void LongSessionMgr::ProcUdpPacket(a8::UdpPacket* pkt)
|
||||
});
|
||||
return;
|
||||
}
|
||||
session->GetKcpSession()->OnRecvPacket(pkt);
|
||||
#if 0
|
||||
if (pkt->buf_len > IKCP_OVERHEAD + KcpSession::GetSecretKeyLen()) {
|
||||
if (session->GetSecretKeyPlace()) {
|
||||
if (pkt->buf_len >= IKCP_OVERHEAD + KcpSession::GetSecretKeyLen()) {
|
||||
long long secret_key = KcpSession::ReadSecretKey(pkt->buf + IKCP_OVERHEAD, pkt->buf_len);
|
||||
if (secret_key == session->GetKcpSession()->GetSecretKey()) {
|
||||
session->GetKcpSession()->OnRecvPacket(pkt);
|
||||
@ -156,7 +155,9 @@ void LongSessionMgr::ProcUdpPacket(a8::UdpPacket* pkt)
|
||||
pkt->buf_len
|
||||
});
|
||||
}
|
||||
#endif
|
||||
} else {
|
||||
session->GetKcpSession()->OnRecvPacket(pkt);
|
||||
}
|
||||
}
|
||||
|
||||
void LongSessionMgr::DelSession(int socket_handle)
|
||||
|
@ -13,6 +13,8 @@
|
||||
|
||||
#include "app.h"
|
||||
|
||||
const int PACK_MAX = 1024 * 64;
|
||||
|
||||
void Master::Init(int instance_id, const std::string& remote_ip, int remote_port)
|
||||
{
|
||||
this->instance_id = instance_id;
|
||||
@ -21,7 +23,10 @@ void Master::Init(int instance_id, const std::string& remote_ip, int remote_port
|
||||
|
||||
recv_bufflen_ = 0;
|
||||
last_pong_tick = a8::XGetTickCount();
|
||||
tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
|
||||
recv_buff_ = (char*) malloc(PACK_MAX * 2);
|
||||
tcp_client_ = std::make_shared<a8::TcpClient>();
|
||||
tcp_client_->remote_address = remote_ip;
|
||||
tcp_client_->remote_port = remote_port;
|
||||
tcp_client_->on_error = std::bind(&Master::on_error, this, std::placeholders::_1, std::placeholders::_2);
|
||||
tcp_client_->on_connect = std::bind(&Master::on_connect, this, std::placeholders::_1);
|
||||
tcp_client_->on_disconnect = std::bind(&Master::on_disconnect, this, std::placeholders::_1);
|
||||
@ -48,6 +53,8 @@ void Master::UnInit()
|
||||
}
|
||||
tcp_client_ = nullptr;
|
||||
recv_bufflen_ = 0;
|
||||
free(recv_buff_);
|
||||
recv_buff_ = nullptr;
|
||||
}
|
||||
|
||||
void Master::Open()
|
||||
@ -70,8 +77,8 @@ void Master::on_error(a8::TcpClient* sender, int errorId)
|
||||
f8::UdpLog::Instance()->Error("Master errorid=%d remote_ip:%s remote_port:%d",
|
||||
{
|
||||
errorId,
|
||||
sender->GetRemoteAddress(),
|
||||
sender->GetRemotePort()
|
||||
sender->remote_address,
|
||||
sender->remote_port
|
||||
});
|
||||
}
|
||||
|
||||
@ -128,7 +135,7 @@ void Master::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
|
||||
f8::UdpLog::Instance()->Debug("recvied bad package", {});
|
||||
}
|
||||
if (offset > 0 && offset < recv_bufflen_) {
|
||||
memmove(recv_buff_.data(), recv_buff_.data() + offset, recv_bufflen_ - offset);
|
||||
memmove(recv_buff_, recv_buff_ + offset, recv_bufflen_ - offset);
|
||||
}
|
||||
recv_bufflen_ -= offset;
|
||||
#if 1
|
||||
|
@ -39,7 +39,7 @@ class Master
|
||||
void CheckAlive();
|
||||
|
||||
private:
|
||||
std::array<char, PACK_MAX> recv_buff_;
|
||||
char *recv_buff_ = nullptr;
|
||||
unsigned int recv_bufflen_ = 0;
|
||||
std::shared_ptr<a8::TcpClient> tcp_client_;
|
||||
f8::TimerWp timer_wp_;
|
||||
|
@ -1,33 +0,0 @@
|
||||
#include "precompile.h"
|
||||
|
||||
#include "mmcache.h"
|
||||
|
||||
void MMCache::Init()
|
||||
{
|
||||
socket_msg_buffer_ = std::make_shared<a8::FifoBuffer>(1024 * 1024 * 10);
|
||||
udp_msg_buffer_ = std::make_shared<a8::FifoBuffer>(1024 * 1024 * 10);
|
||||
}
|
||||
|
||||
void MMCache::UnInit()
|
||||
{
|
||||
}
|
||||
|
||||
char* MMCache::AllocSocketMsg(int len)
|
||||
{
|
||||
return socket_msg_buffer_->Alloc(len);
|
||||
}
|
||||
|
||||
void MMCache::FreeSocketMsg(char* p)
|
||||
{
|
||||
socket_msg_buffer_->Free(p);
|
||||
}
|
||||
|
||||
char* MMCache::AllocUdpMsg(int len)
|
||||
{
|
||||
return udp_msg_buffer_->Alloc(len);
|
||||
}
|
||||
|
||||
void MMCache::FreeUdpMsg(char* p)
|
||||
{
|
||||
udp_msg_buffer_->Free(p);
|
||||
}
|
@ -1,28 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include <a8/singleton.h>
|
||||
#include <a8/fifobuffer.h>
|
||||
|
||||
class MMCache : public a8::Singleton<MMCache>
|
||||
{
|
||||
private:
|
||||
MMCache() {};
|
||||
friend class a8::Singleton<MMCache>;
|
||||
|
||||
public:
|
||||
|
||||
void Init();
|
||||
void UnInit();
|
||||
|
||||
char* AllocSocketMsg(int len);
|
||||
void FreeSocketMsg(char* p);
|
||||
|
||||
char* AllocUdpMsg(int len);
|
||||
void FreeUdpMsg(char* p);
|
||||
|
||||
private:
|
||||
|
||||
std::shared_ptr<a8::FifoBuffer> socket_msg_buffer_;
|
||||
std::shared_ptr<a8::FifoBuffer> udp_msg_buffer_;
|
||||
|
||||
};
|
@ -14,6 +14,8 @@
|
||||
|
||||
#include "app.h"
|
||||
|
||||
const int PACK_MAX = 1024 * 64 * 2;
|
||||
|
||||
void UpStream::Init(int instance_id, const std::string& remote_ip, int remote_port)
|
||||
{
|
||||
if (remote_ip.empty()) {
|
||||
@ -25,7 +27,10 @@ void UpStream::Init(int instance_id, const std::string& remote_ip, int remote_po
|
||||
|
||||
recv_bufflen_ = 0;
|
||||
last_pong_tick = a8::XGetTickCount();
|
||||
tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
|
||||
recv_buff_ = (char*) malloc(PACK_MAX * 2);
|
||||
tcp_client_ = std::make_shared<a8::TcpClient>();
|
||||
tcp_client_->remote_address = remote_ip;
|
||||
tcp_client_->remote_port = remote_port;
|
||||
tcp_client_->on_error = std::bind(&UpStream::on_error, this, std::placeholders::_1, std::placeholders::_2);
|
||||
tcp_client_->on_connect = std::bind(&UpStream::on_connect, this, std::placeholders::_1);
|
||||
tcp_client_->on_disconnect = std::bind(&UpStream::on_disconnect, this, std::placeholders::_1);
|
||||
@ -62,6 +67,8 @@ void UpStream::UnInit()
|
||||
abort();
|
||||
}
|
||||
recv_bufflen_ = 0;
|
||||
free(recv_buff_);
|
||||
recv_buff_ = nullptr;
|
||||
}
|
||||
|
||||
void UpStream::Open()
|
||||
@ -104,7 +111,7 @@ void UpStream::SendStockMsg()
|
||||
|
||||
void UpStream::ForwardClientMsg(f8::MsgHdr& hdr)
|
||||
{
|
||||
char* buff = forward_buff_.data();
|
||||
char* buff = (char*)malloc(sizeof(f8::WSProxyPackHead_C) + hdr.buflen);
|
||||
memset(buff, 0, sizeof(f8::WSProxyPackHead_C));
|
||||
f8::WSProxyPackHead_C* head = (f8::WSProxyPackHead_C*)buff;
|
||||
head->packlen = hdr.buflen;
|
||||
@ -122,6 +129,7 @@ void UpStream::ForwardClientMsg(f8::MsgHdr& hdr)
|
||||
}
|
||||
|
||||
tcp_client_->SendBuff(buff, sizeof(f8::WSProxyPackHead_C) + head->packlen);
|
||||
free(buff);
|
||||
}
|
||||
|
||||
void UpStream::ForwardClientMsgEx(f8::MsgHdr* hdr)
|
||||
@ -145,8 +153,8 @@ void UpStream::on_error(a8::TcpClient* sender, int errorId)
|
||||
f8::UdpLog::Instance()->Error("target server errorid=%d remote_ip:%s remote_port:%d",
|
||||
{
|
||||
errorId,
|
||||
sender->GetRemoteAddress(),
|
||||
sender->GetRemotePort()
|
||||
sender->remote_address,
|
||||
sender->remote_port
|
||||
});
|
||||
}
|
||||
|
||||
@ -155,8 +163,8 @@ void UpStream::on_connect(a8::TcpClient* sender)
|
||||
recv_bufflen_ = 0;
|
||||
f8::UdpLog::Instance()->Info("target server connected remote_ip:%s remote_port:%d",
|
||||
{
|
||||
sender->GetRemoteAddress(),
|
||||
sender->GetRemotePort()
|
||||
sender->remote_address,
|
||||
sender->remote_port
|
||||
});
|
||||
f8::MsgQueue::Instance()->PostMsg
|
||||
(
|
||||
@ -177,8 +185,8 @@ void UpStream::on_disconnect(a8::TcpClient* sender)
|
||||
"remote_ip:%s remote_port:%d",
|
||||
{
|
||||
instance_id,
|
||||
sender->GetRemoteAddress(),
|
||||
sender->GetRemotePort()
|
||||
sender->remote_address,
|
||||
sender->remote_port
|
||||
});
|
||||
f8::MsgQueue::Instance()->PostMsg
|
||||
(
|
||||
@ -234,7 +242,7 @@ void UpStream::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
|
||||
f8::UdpLog::Instance()->Debug("recvied bad package", {});
|
||||
}
|
||||
if (offset > 0 && offset < recv_bufflen_) {
|
||||
memmove(recv_buff_.data(), recv_buff_.data() + offset, recv_bufflen_ - offset);
|
||||
memmove(recv_buff_, recv_buff_ + offset, recv_bufflen_ - offset);
|
||||
}
|
||||
recv_bufflen_ -= offset;
|
||||
#if 1
|
||||
|
@ -64,8 +64,7 @@ class UpStream
|
||||
f8::MsgHdr* hdr);
|
||||
|
||||
private:
|
||||
std::array<char, PACK_MAX> recv_buff_;
|
||||
std::array<char, PACK_MAX> forward_buff_;
|
||||
char *recv_buff_ = nullptr;
|
||||
unsigned int recv_bufflen_ = 0;
|
||||
std::shared_ptr<a8::TcpClient> tcp_client_;
|
||||
f8::TimerWp timer_wp_;
|
||||
|
2
third_party/a8
vendored
2
third_party/a8
vendored
@ -1 +1 @@
|
||||
Subproject commit 424f7083751ca1d2bf289ac9be46b658437b7d33
|
||||
Subproject commit 1e577389c8a2870db9ddbf18577bfca24def049b
|
2
third_party/f8
vendored
2
third_party/f8
vendored
@ -1 +1 @@
|
||||
Subproject commit 874ff7853e30ff22c556a2c4cbf4046423c7be86
|
||||
Subproject commit 243bbe515ef4a01089f9a6cf608c93d4097018de
|
Loading…
x
Reference in New Issue
Block a user