Compare commits

..

20 Commits
master ... asio

Author SHA1 Message Date
azw
be09f290d0 1 2023-05-07 01:57:22 +00:00
azw
e41ffbd58f 1 2023-05-07 01:55:19 +00:00
azw
9dc08638a0 1 2023-05-07 01:33:14 +00:00
azw
8569af06ea 1 2023-05-07 01:26:46 +00:00
azw
f71a4ef0fd 1 2023-05-07 01:21:54 +00:00
azw
101ab36a7a 1 2023-05-07 01:19:08 +00:00
azw
9394d487c0 1 2023-05-07 01:17:03 +00:00
azw
48295756bf 1 2023-05-07 01:07:17 +00:00
azw
997a1de9b8 1 2023-05-06 15:32:12 +00:00
azw
c5e5a8b615 1 2023-05-06 15:00:54 +00:00
azw
4a00d82cd6 1 2023-05-06 14:31:17 +00:00
azw
a19a9dd588 1 2023-05-06 12:19:12 +00:00
azw
e0ebd4624a 1 2023-05-05 23:42:01 +00:00
azw
b064aed353 1 2023-05-05 23:22:34 +00:00
azw
5d69655f43 1 2023-05-04 15:51:44 +00:00
azw
9dad68d86f 1 2023-05-04 15:29:41 +00:00
azw
cc60352018 1 2023-05-04 15:23:46 +00:00
aozhiwei
cb70b1f901 1 2023-05-04 14:53:31 +08:00
aozhiwei
ec7c4797e9 1 2023-05-04 13:40:08 +08:00
aozhiwei
17ea527dba 1 2023-05-04 13:37:51 +08:00
20 changed files with 243 additions and 120 deletions

16
server/bin/udpclient.js Normal file
View File

@ -0,0 +1,16 @@
const dgram = require('dgram');
const client = dgram.createSocket('udp4');
client.on('close', () => {
console.log('close');
});
client.on('error', () => {
console.log('error');
});
client.on('message', (msg, rinfo) => {
console.log(`接收到来自:${rinfo.address}:${rinfo.port} 的消息: ${msg}`);
});
client.send('hello', 17601, '108.137.177.19');

17
server/bin/udpserver.js Normal file
View File

@ -0,0 +1,17 @@
const dgram = require('dgram');
const server = dgram.createSocket('udp4');
server.on('message', (msg, rinfo) => {
console.log('rinfo.address = ' + rinfo.address);
console.log('rinfo.port = ' + rinfo.port);
console.log(msg.toString());
server.send(msg, rinfo.port, rinfo.address);
});
server.on('listening', () => {
console.log('address:' + server.address().address);
console.log('port:' + server.address().port);
});
server.bind(17601);

View File

@ -24,7 +24,6 @@ message SS_CMKcpHandshake
optional string account_id = 2; //id optional string account_id = 2; //id
optional string session_id = 3; //session id optional string session_id = 3; //session id
optional string team_uuid = 4; // optional string team_uuid = 4; //
optional int32 secret_key_place = 5; // 0() 1kcp底层协议头之后()
} }
message SS_SMKcpHandshake message SS_SMKcpHandshake

View File

@ -20,9 +20,9 @@ endif()
set(CMAKE_CXX_FLAGS_RELEASE "-O3 -Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DNDEBUG") set(CMAKE_CXX_FLAGS_RELEASE "-O3 -Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DNDEBUG")
if (${ASAN}) if (${ASAN})
set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -fsanitize=address -fno-omit-frame-pointer") set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -fsanitize=address -fno-omit-frame-pointer -DUSE_ASIO=1")
else() else()
set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG") set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -DUSE_ASIO=1")
endif() endif()
include_directories( include_directories(

View File

@ -25,6 +25,7 @@
#include "upstreammgr.h" #include "upstreammgr.h"
#include "master.h" #include "master.h"
#include "mastermgr.h" #include "mastermgr.h"
#include "iomgr.h"
#include "longsessionmgr.h" #include "longsessionmgr.h"
@ -127,6 +128,9 @@ bool App::Init(int argc, char* argv[])
f8::Timer::Instance()->Init(); f8::Timer::Instance()->Init();
JsonDataMgr::Instance()->Init(); JsonDataMgr::Instance()->Init();
uuid_->SetMachineId((node_id_ - 1) * MAX_NODE_ID + instance_id_); uuid_->SetMachineId((node_id_ - 1) * MAX_NODE_ID + instance_id_);
#ifdef USE_ASIO
IoMgr::Instance()->Init();
#endif
DownStreamMgr::Instance()->Init(); DownStreamMgr::Instance()->Init();
MasterMgr::Instance()->Init(); MasterMgr::Instance()->Init();
UpStreamMgr::Instance()->Init(); UpStreamMgr::Instance()->Init();
@ -189,6 +193,9 @@ void App::UnInit()
MasterMgr::Instance()->UnInit(); MasterMgr::Instance()->UnInit();
UpStreamMgr::Instance()->UnInit(); UpStreamMgr::Instance()->UnInit();
DownStreamMgr::Instance()->UnInit(); DownStreamMgr::Instance()->UnInit();
#ifdef USE_ASIO
IoMgr::Instance()->UnInit();
#endif
JsonDataMgr::Instance()->UnInit(); JsonDataMgr::Instance()->UnInit();
f8::Timer::Instance()->UnInit(); f8::Timer::Instance()->UnInit();
f8::MsgQueue::Instance()->UnInit(); f8::MsgQueue::Instance()->UnInit();

View File

@ -23,6 +23,13 @@ enum InnerMesssage_e
IM_HandleRedirect IM_HandleRedirect
}; };
enum IoContext_e
{
IC_Master,
IC_UpStream,
IC_Max
};
//网络处理对象 //网络处理对象
enum NetHandler_e enum NetHandler_e
{ {

41
server/wsproxy/iomgr.cc Normal file
View File

@ -0,0 +1,41 @@
#include "precompile.h"
#include "iomgr.h"
#ifdef USE_ASIO
void IoMgr::Init()
{
for (int i = 0; i < IC_Max; ++i) {
io_contexts_.push_back(std::vector<std::shared_ptr<asio::io_context>>());
auto& contexts = io_contexts_.at(i);
contexts.push_back(std::make_shared<asio::io_context>());
new std::thread(&IoMgr::WorkerThreadProc, this, contexts.at(0));
}
}
void IoMgr::UnInit()
{
}
std::shared_ptr<asio::io_context> IoMgr::GetIoContext(int type)
{
if (type >= 0 && type < io_contexts_.size()) {
auto& contexts = io_contexts_.at(type);
if (!contexts.empty()) {
return contexts.at(rand() % contexts.size());
}
}
return nullptr;
}
void IoMgr::WorkerThreadProc(std::shared_ptr<asio::io_context> io_context)
{
while (true) {
io_context->run();
io_context->reset();
}
}
#endif

28
server/wsproxy/iomgr.h Normal file
View File

@ -0,0 +1,28 @@
#pragma once
#ifdef USE_ASIO
#include <asio.hpp>
class IoMgr : public a8::Singleton<IoMgr>
{
private:
IoMgr() {};
friend class a8::Singleton<IoMgr>;
public:
void Init();
void UnInit();
std::shared_ptr<asio::io_context> GetIoContext(int type);
private:
void WorkerThreadProc(std::shared_ptr<asio::io_context> io_context);
private:
std::vector<std::vector<std::shared_ptr<asio::io_context>>> io_contexts_;
};
#endif

View File

@ -1,7 +1,6 @@
#include "precompile.h" #include "precompile.h"
#include <memory.h> #include <memory.h>
#include <string.h>
#include <f8/netmsghandler.h> #include <f8/netmsghandler.h>
#include <f8/udplog.h> #include <f8/udplog.h>
@ -34,22 +33,23 @@ KcpSession::~KcpSession()
} }
void KcpSession::Init(int socket_handle, int secret_key_place) void KcpSession::Init(int socket_handle)
{ {
socket_handle_ = socket_handle; socket_handle_ = socket_handle;
secret_key_ = App::Instance()->NewUuid(); secret_key_ = App::Instance()->NewUuid();
kcp_ = ikcp_create(socket_handle_, (void*)this); kcp_ = ikcp_create(socket_handle_, (void*)this);
#if 1
const KcpConf& kcp_conf = JsonDataMgr::Instance()->GetKcpConf(); const KcpConf& kcp_conf = JsonDataMgr::Instance()->GetKcpConf();
ikcp_wndsize(kcp_, kcp_conf.sndwnd, kcp_conf.rcvwnd); ikcp_wndsize(kcp_, kcp_conf.sndwnd, kcp_conf.rcvwnd);
ikcp_nodelay(kcp_, kcp_conf.nodelay, kcp_conf.interval, kcp_conf.resend, kcp_conf.nc); ikcp_nodelay(kcp_, kcp_conf.nodelay, kcp_conf.interval, kcp_conf.resend, kcp_conf.nc);
kcp_->rx_minrto = kcp_conf.rx_minrto; kcp_->rx_minrto = kcp_conf.rx_minrto;
kcp_->fastresend = kcp_conf.fastresend; kcp_->fastresend = kcp_conf.fastresend;
secret_key_place_ = secret_key_place; #else
if (secret_key_place_ > 0) { ikcp_wndsize(kcp_, 128, 128);
secret_key_place_ = 1; ikcp_nodelay(kcp_, 2, 10, 2, 1);
} else if (secret_key_place_ < 0) { kcp_->rx_minrto = 10;
secret_key_place_ = 0; kcp_->fastresend = 1;
} #endif
kcp_->output = UdpOutput; kcp_->output = UdpOutput;
init_tick_ = a8::XGetTickCount(); init_tick_ = a8::XGetTickCount();
} }
@ -75,21 +75,8 @@ void KcpSession::SendClientMsg(char* buf, int buf_len)
void KcpSession::OnRecvPacket(a8::UdpPacket* pkt) void KcpSession::OnRecvPacket(a8::UdpPacket* pkt)
{ {
const int IKCP_OVERHEAD = 24;
remote_addr_ = pkt->remote_addr; 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); ikcp_input(kcp_, pkt->buf, pkt->buf_len);
}
} }
void KcpSession::UpdateInput() void KcpSession::UpdateInput()
@ -104,20 +91,8 @@ void KcpSession::UpdateInput()
void KcpSession::DecodeUserPacket(char* buf, int& offset, unsigned int buflen) void KcpSession::DecodeUserPacket(char* buf, int& offset, unsigned int buflen)
{ {
if (GetSecretKeyPlace()) { //packagelen + msgid + magiccode + msgbody
DecodeUserPacketNew(buf, offset, buflen); //2 + 2 + 4+ xx + \0 + xx
} 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; bool warning = false;
while (buflen - offset >= sizeof(f8::PackHead) + GetSecretKeyLen()) { while (buflen - offset >= sizeof(f8::PackHead) + GetSecretKeyLen()) {
long long secret_key = KcpSession::ReadSecretKey(&buf[offset], buflen); long long secret_key = KcpSession::ReadSecretKey(&buf[offset], buflen);
@ -149,37 +124,6 @@ void KcpSession::DecodeUserPacketOld(char* buf, int& offset, unsigned int buflen
} }
if (warning) { if (warning) {
f8::UdpLog::Instance()->Warning("收到kcp client非法数据包1", {}); f8::UdpLog::Instance()->Warning("收到kcp client非法数据包", {});
}
}
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", {});
} }
} }

View File

@ -11,7 +11,7 @@ public:
KcpSession(); KcpSession();
~KcpSession(); ~KcpSession();
void Init(int socket_handle, int secret_key_place); void Init(int socket_handle);
void UnInit(); void UnInit();
void Update(long long tick); void Update(long long tick);
@ -19,7 +19,6 @@ public:
int GetSocketHandle() { return socket_handle_; } int GetSocketHandle() { return socket_handle_; }
long long GetSecretKey() { return secret_key_; } long long GetSecretKey() { return secret_key_; }
void* GetSecretKeyDataPtr() { return &secret_key_; } void* GetSecretKeyDataPtr() { return &secret_key_; }
int GetSecretKeyPlace();
void SendClientMsg(char* buf, int buf_len); void SendClientMsg(char* buf, int buf_len);
virtual void OnRecvPacket(a8::UdpPacket* pkt) override; virtual void OnRecvPacket(a8::UdpPacket* pkt) override;
@ -32,14 +31,11 @@ public:
protected: protected:
virtual void DecodeUserPacket(char* buf, int& offset, unsigned int buflen) override; 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: private:
void UpdateInput(); void UpdateInput();
private: private:
int secret_key_place_ = 0;
int socket_handle_ = 0; int socket_handle_ = 0;
long long secret_key_ = 0; long long secret_key_ = 0;
ikcpcb* kcp_ = nullptr; ikcpcb* kcp_ = nullptr;

View File

@ -5,13 +5,10 @@
#include "longsession.h" #include "longsession.h"
#include "kcpsession.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) void LongSession::Init(f8::MsgHdr& hdr, const ss::SS_CMKcpHandshake& msg)
{ {
kcp_session_ = std::make_shared<KcpSession>(); kcp_session_ = std::make_shared<KcpSession>();
kcp_session_->Init(hdr.socket_handle, msg.secret_key_place()); kcp_session_->Init(hdr.socket_handle);
} }
void LongSession::UnInit() void LongSession::UnInit()
@ -28,8 +25,3 @@ void LongSession::UpdatePing()
{ {
last_ping_tick_ = a8::XGetTickCount(); last_ping_tick_ = a8::XGetTickCount();
} }
int LongSession::GetSecretKeyPlace()
{
return kcp_session_->GetSecretKeyPlace();
}

View File

@ -17,7 +17,6 @@ class LongSession
std::shared_ptr<KcpSession> GetKcpSession() { return kcp_session_; } std::shared_ptr<KcpSession> GetKcpSession() { return kcp_session_; }
void UpdatePing(); void UpdatePing();
int GetSecretKeyPlace();
private: private:
long long last_ping_tick_ = 0; long long last_ping_tick_ = 0;

View File

@ -133,8 +133,9 @@ void LongSessionMgr::ProcUdpPacket(a8::UdpPacket* pkt)
}); });
return; return;
} }
if (session->GetSecretKeyPlace()) { session->GetKcpSession()->OnRecvPacket(pkt);
if (pkt->buf_len >= IKCP_OVERHEAD + KcpSession::GetSecretKeyLen()) { #if 0
if (pkt->buf_len > IKCP_OVERHEAD + KcpSession::GetSecretKeyLen()) {
long long secret_key = KcpSession::ReadSecretKey(pkt->buf + IKCP_OVERHEAD, pkt->buf_len); long long secret_key = KcpSession::ReadSecretKey(pkt->buf + IKCP_OVERHEAD, pkt->buf_len);
if (secret_key == session->GetKcpSession()->GetSecretKey()) { if (secret_key == session->GetKcpSession()->GetSecretKey()) {
session->GetKcpSession()->OnRecvPacket(pkt); session->GetKcpSession()->OnRecvPacket(pkt);
@ -155,9 +156,7 @@ void LongSessionMgr::ProcUdpPacket(a8::UdpPacket* pkt)
pkt->buf_len pkt->buf_len
}); });
} }
} else { #endif
session->GetKcpSession()->OnRecvPacket(pkt);
}
} }
void LongSessionMgr::DelSession(int socket_handle) void LongSessionMgr::DelSession(int socket_handle)

View File

@ -2,7 +2,11 @@
#include <string.h> #include <string.h>
#ifdef USE_ASIO
#include <a8/asiotcpclient.h>
#else
#include <a8/tcpclient.h> #include <a8/tcpclient.h>
#endif
#include <f8/udplog.h> #include <f8/udplog.h>
#include <f8/timer.h> #include <f8/timer.h>
@ -12,6 +16,7 @@
#include "master.h" #include "master.h"
#include "app.h" #include "app.h"
#include "iomgr.h"
const int PACK_MAX = 1024 * 64; const int PACK_MAX = 1024 * 64;
@ -24,9 +29,15 @@ void Master::Init(int instance_id, const std::string& remote_ip, int remote_port
recv_bufflen_ = 0; recv_bufflen_ = 0;
last_pong_tick = a8::XGetTickCount(); last_pong_tick = a8::XGetTickCount();
recv_buff_ = (char*) malloc(PACK_MAX * 2); recv_buff_ = (char*) malloc(PACK_MAX * 2);
tcp_client_ = std::make_shared<a8::TcpClient>(); #ifdef USE_ASIO
tcp_client_->remote_address = remote_ip; auto io_context = IoMgr::Instance()->GetIoContext(IC_Master);
tcp_client_->remote_port = remote_port; if (!io_context) {
abort();
}
tcp_client_ = std::make_shared<a8::AsioTcpClient>(*io_context, remote_ip, remote_port);
#else
tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
#endif
tcp_client_->on_error = std::bind(&Master::on_error, this, std::placeholders::_1, std::placeholders::_2); 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_connect = std::bind(&Master::on_connect, this, std::placeholders::_1);
tcp_client_->on_disconnect = std::bind(&Master::on_disconnect, this, std::placeholders::_1); tcp_client_->on_disconnect = std::bind(&Master::on_disconnect, this, std::placeholders::_1);
@ -72,29 +83,45 @@ bool Master::Connected()
return tcp_client_->Connected(); return tcp_client_->Connected();
} }
#ifdef USE_ASIO
void Master::on_error(a8::AsioTcpClient* sender, int errorId)
#else
void Master::on_error(a8::TcpClient* sender, int errorId) void Master::on_error(a8::TcpClient* sender, int errorId)
#endif
{ {
f8::UdpLog::Instance()->Error("Master errorid=%d remote_ip:%s remote_port:%d", f8::UdpLog::Instance()->Error("Master errorid=%d remote_ip:%s remote_port:%d",
{ {
errorId, errorId,
sender->remote_address, sender->GetRemoteAddress(),
sender->remote_port sender->GetRemotePort()
}); });
} }
#ifdef USE_ASIO
void Master::on_connect(a8::AsioTcpClient* sender)
#else
void Master::on_connect(a8::TcpClient* sender) void Master::on_connect(a8::TcpClient* sender)
#endif
{ {
recv_bufflen_ = 0; recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("masterserver connected", {}); f8::UdpLog::Instance()->Info("masterserver connected", {});
} }
#ifdef USE_ASIO
void Master::on_disconnect(a8::AsioTcpClient* sender)
#else
void Master::on_disconnect(a8::TcpClient* sender) void Master::on_disconnect(a8::TcpClient* sender)
#endif
{ {
recv_bufflen_ = 0; recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("masterserver %d disconnected after 10s later reconnect", {instance_id}); f8::UdpLog::Instance()->Info("masterserver %d disconnected after 10s later reconnect", {instance_id});
} }
#ifdef USE_ASIO
void Master::on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len)
#else
void Master::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len) void Master::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
#endif
{ {
#if 0 #if 0
++App::Instance()->perf.read_count; ++App::Instance()->perf.read_count;

View File

@ -5,6 +5,7 @@
namespace a8 namespace a8
{ {
class TcpClient; class TcpClient;
class AsioTcpClient;
} }
class Master class Master
@ -31,17 +32,28 @@ class Master
} }
private: private:
#ifdef USE_ASIO
void on_error(a8::AsioTcpClient* sender, int errorId);
void on_connect(a8::AsioTcpClient* sender);
void on_disconnect(a8::AsioTcpClient* sender);
void on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len);
#else
void on_error(a8::TcpClient* sender, int errorId); void on_error(a8::TcpClient* sender, int errorId);
void on_connect(a8::TcpClient* sender); void on_connect(a8::TcpClient* sender);
void on_disconnect(a8::TcpClient* sender); void on_disconnect(a8::TcpClient* sender);
void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len); void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len);
#endif
void CheckAlive(); void CheckAlive();
private: private:
char *recv_buff_ = nullptr; char *recv_buff_ = nullptr;
unsigned int recv_bufflen_ = 0; unsigned int recv_bufflen_ = 0;
#ifdef USE_ASIO
std::shared_ptr<a8::AsioTcpClient> tcp_client_;
#else
std::shared_ptr<a8::TcpClient> tcp_client_; std::shared_ptr<a8::TcpClient> tcp_client_;
#endif
f8::TimerWp timer_wp_; f8::TimerWp timer_wp_;
f8::Attacher attacher_; f8::Attacher attacher_;
}; };

View File

@ -2,7 +2,11 @@
#include <string.h> #include <string.h>
#ifdef USE_ASIO
#include <a8/asiotcpclient.h>
#else
#include <a8/tcpclient.h> #include <a8/tcpclient.h>
#endif
#include <f8/udplog.h> #include <f8/udplog.h>
#include <f8/timer.h> #include <f8/timer.h>
@ -13,6 +17,7 @@
#include "upstream.h" #include "upstream.h"
#include "app.h" #include "app.h"
#include "iomgr.h"
const int PACK_MAX = 1024 * 64 * 2; const int PACK_MAX = 1024 * 64 * 2;
@ -28,9 +33,15 @@ void UpStream::Init(int instance_id, const std::string& remote_ip, int remote_po
recv_bufflen_ = 0; recv_bufflen_ = 0;
last_pong_tick = a8::XGetTickCount(); last_pong_tick = a8::XGetTickCount();
recv_buff_ = (char*) malloc(PACK_MAX * 2); recv_buff_ = (char*) malloc(PACK_MAX * 2);
tcp_client_ = std::make_shared<a8::TcpClient>(); #ifdef USE_ASIO
tcp_client_->remote_address = remote_ip; auto io_context = IoMgr::Instance()->GetIoContext(IC_UpStream);
tcp_client_->remote_port = remote_port; if (!io_context) {
abort();
}
tcp_client_ = std::make_shared<a8::AsioTcpClient>(*io_context, remote_ip, remote_port);
#else
tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
#endif
tcp_client_->on_error = std::bind(&UpStream::on_error, this, std::placeholders::_1, std::placeholders::_2); 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_connect = std::bind(&UpStream::on_connect, this, std::placeholders::_1);
tcp_client_->on_disconnect = std::bind(&UpStream::on_disconnect, this, std::placeholders::_1); tcp_client_->on_disconnect = std::bind(&UpStream::on_disconnect, this, std::placeholders::_1);
@ -148,23 +159,31 @@ void UpStream::ForwardClientMsgEx(f8::MsgHdr* hdr)
} }
} }
#ifdef USE_ASIO
void UpStream::on_error(a8::AsioTcpClient* sender, int errorId)
#else
void UpStream::on_error(a8::TcpClient* sender, int errorId) void UpStream::on_error(a8::TcpClient* sender, int errorId)
#endif
{ {
f8::UdpLog::Instance()->Error("target server errorid=%d remote_ip:%s remote_port:%d", f8::UdpLog::Instance()->Error("target server errorid=%d remote_ip:%s remote_port:%d",
{ {
errorId, errorId,
sender->remote_address, sender->GetRemoteAddress(),
sender->remote_port sender->GetRemotePort()
}); });
} }
#ifdef USE_ASIO
void UpStream::on_connect(a8::AsioTcpClient* sender)
#else
void UpStream::on_connect(a8::TcpClient* sender) void UpStream::on_connect(a8::TcpClient* sender)
#endif
{ {
recv_bufflen_ = 0; recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("target server connected remote_ip:%s remote_port:%d", f8::UdpLog::Instance()->Info("target server connected remote_ip:%s remote_port:%d",
{ {
sender->remote_address, sender->GetRemoteAddress(),
sender->remote_port sender->GetRemotePort()
}); });
f8::MsgQueue::Instance()->PostMsg f8::MsgQueue::Instance()->PostMsg
( (
@ -178,15 +197,19 @@ void UpStream::on_connect(a8::TcpClient* sender)
); );
} }
#ifdef USE_ASIO
void UpStream::on_disconnect(a8::AsioTcpClient* sender)
#else
void UpStream::on_disconnect(a8::TcpClient* sender) void UpStream::on_disconnect(a8::TcpClient* sender)
#endif
{ {
recv_bufflen_ = 0; recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("target server %d disconnected after 10s later reconnect " f8::UdpLog::Instance()->Info("target server %d disconnected after 10s later reconnect "
"remote_ip:%s remote_port:%d", "remote_ip:%s remote_port:%d",
{ {
instance_id, instance_id,
sender->remote_address, sender->GetRemoteAddress(),
sender->remote_port sender->GetRemotePort()
}); });
f8::MsgQueue::Instance()->PostMsg f8::MsgQueue::Instance()->PostMsg
( (
@ -200,7 +223,11 @@ void UpStream::on_disconnect(a8::TcpClient* sender)
); );
} }
#ifdef USE_ASIO
void UpStream::on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len)
#else
void UpStream::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len) void UpStream::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
#endif
{ {
#if 0 #if 0
++App::Instance()->perf.read_count; ++App::Instance()->perf.read_count;

View File

@ -5,6 +5,7 @@
namespace a8 namespace a8
{ {
class TcpClient; class TcpClient;
class AsioTcpClient;
} }
struct UpStreamMsgNode struct UpStreamMsgNode
@ -26,6 +27,7 @@ class UpStream
a8::tick_t last_pong_tick = 0; a8::tick_t last_pong_tick = 0;
public: public:
void Init(int instance_id, const std::string& remote_ip, int remote_port); void Init(int instance_id, const std::string& remote_ip, int remote_port);
void UnInit(); void UnInit();
@ -54,10 +56,17 @@ class UpStream
void ForwardClientMsgEx(f8::MsgHdr* hdr); void ForwardClientMsgEx(f8::MsgHdr* hdr);
private: private:
#ifdef USE_ASIO
void on_error(a8::AsioTcpClient* sender, int errorId);
void on_connect(a8::AsioTcpClient* sender);
void on_disconnect(a8::AsioTcpClient* sender);
void on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len);
#else
void on_error(a8::TcpClient* sender, int errorId); void on_error(a8::TcpClient* sender, int errorId);
void on_connect(a8::TcpClient* sender); void on_connect(a8::TcpClient* sender);
void on_disconnect(a8::TcpClient* sender); void on_disconnect(a8::TcpClient* sender);
void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len); void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len);
#endif
void CheckAlive(); void CheckAlive();
void AddStockMsg(unsigned short socket_handle, int msgid, ::google::protobuf::Message* msg, void AddStockMsg(unsigned short socket_handle, int msgid, ::google::protobuf::Message* msg,
@ -66,7 +75,11 @@ class UpStream
private: private:
char *recv_buff_ = nullptr; char *recv_buff_ = nullptr;
unsigned int recv_bufflen_ = 0; unsigned int recv_bufflen_ = 0;
#ifdef USE_ASIO
std::shared_ptr<a8::AsioTcpClient> tcp_client_;
#else
std::shared_ptr<a8::TcpClient> tcp_client_; std::shared_ptr<a8::TcpClient> tcp_client_;
#endif
f8::TimerWp timer_wp_; f8::TimerWp timer_wp_;
f8::Attacher attacher_; f8::Attacher attacher_;

View File

@ -55,7 +55,6 @@ std::weak_ptr<UpStream> UpStreamMgr::RecreateUpStream(const std::string& host, i
int instance_id = curr_id_; int instance_id = curr_id_;
std::string remote_ip = host; std::string remote_ip = host;
int remote_port = port; int remote_port = port;
std::shared_ptr<UpStream> conn = std::make_shared<UpStream>(); std::shared_ptr<UpStream> conn = std::make_shared<UpStream>();
conn->Init(instance_id, remote_ip, remote_port); conn->Init(instance_id, remote_ip, remote_port);
id_hash_[conn->instance_id] = conn; id_hash_[conn->instance_id] = conn;

2
third_party/a8 vendored

@ -1 +1 @@
Subproject commit 1e577389c8a2870db9ddbf18577bfca24def049b Subproject commit 466448796dff77a16190519c25879cfe0bc534ec

2
third_party/f8 vendored

@ -1 +1 @@
Subproject commit 243bbe515ef4a01089f9a6cf608c93d4097018de Subproject commit 8a49aa65c8576cabe1672be55fc903a6b17e2c37