feat(server): 添加数据处理模块并增强服务器功能

- 新增 DataProcesser 类,用于处理数据包和管理会话
- 在 GateServer 中添加客户端管理、广播消息和服务器停止功能
- 优化 GateServer 的连接处理逻辑,增加客户端 ID 和信号处理
This commit is contained in:
2025-04-10 07:23:02 +08:00
parent d1643f054e
commit 002e7214ce
2 changed files with 110 additions and 3 deletions

70
GateServer/DataProcess.js Normal file
View File

@@ -0,0 +1,70 @@
class DataProcesser {
constructor(maxSessionCount) {
this.maxSessionCount = maxSessionCount;
this.activeUser = 0;
this.sessions = new Array(maxSessionCount).fill(null);
this.sendThreads = [];
this.recvQueue = [];
this.serverBuf = { buffer: null, size: 0, offset: 0 };
this.userVerify = 1;
this.processRecvSize = 0;
this.waitSendUserSize = 0;
this.waitSendQueueSize = 0;
this.sendUserSize = 0;
this.recvSeverSize = 0;
this.lastProcUsrMsgTime = 0;
this.lastProcSrvMsgTime = 0;
this.lastRecvSrvMsgTime = 0;
this.procSrvThreadSleep = 0;
this.sendQueueSize = 0;
this.ignoreDataPacket = 0;
this.sndThreadSleepTime = 20;
}
initSessions() {
for (let i = 0; i < this.maxSessionCount; i++) {
this.sessions[i] = {
socket: null,
serverIdx: 0,
packetIdx: 0,
packetError: 0,
recvPacketCount: 0,
sendPacketCount: 0,
markToClose: false,
remoteClosed: false,
sendAvaliable: true,
sendTimeout: 0,
closeTick: 0,
connectTick: Date.now(),
clientMsgTick: Date.now(),
serverMsgTick: Date.now(),
verifyIdx: this.userVerify++,
recvBuf: { buffer: null, size: 0, offset: 0 },
sendBuf: { buffer: null, size: 0, offset: 0 }
};
}
}
processUserRecvPacket(session, buffer, bufferSize) {
// 处理用户接收到的数据包
// 这里需要实现具体的数据处理逻辑
}
sendServerMessage(ident, sessionIdx, socket, serverIdx, buffer, bufferSize) {
// 发送消息到服务器
// 这里需要实现具体的消息发送逻辑
}
startup() {
this.initSessions();
// 启动数据处理线程
// 这里需要实现具体的线程启动逻辑
}
stop() {
// 停止数据处理线程
// 这里需要实现具体的线程停止逻辑
}
}
module.exports = DataProcesser;

View File

@@ -7,16 +7,32 @@ class GateServer extends EventEmitter {
this.port = port;
this.host = host;
this.server = net.createServer(this.handleConnection.bind(this));
this.clients = new Map(); // 用于存储连接的客户端
this.gateEngineRunning = true;
}
start() {
this.server.listen(this.port, this.host, () => {
console.log(`GateServer is listening on ${this.host}:${this.port}`);
});
// 处理信号
process.on('SIGINT', () => this.signalHandler('SIGINT'));
process.on('SIGTERM', () => this.signalHandler('SIGTERM'));
}
signalHandler(signal) {
if (signal === 'SIGINT' || signal === 'SIGTERM') {
this.gateEngineRunning = false;
console.log(`[SIGNAL] ${signal} received, shutting down...`);
this.stop();
}
}
handleConnection(socket) {
console.log('New client connected');
const clientId = `${socket.remoteAddress}:${socket.remotePort}`;
console.log(`New client connected: ${clientId}`);
this.clients.set(clientId, socket); // 存储客户端连接
this.emit('connection', socket);
socket.on('data', (data) => {
@@ -24,15 +40,36 @@ class GateServer extends EventEmitter {
});
socket.on('end', () => {
console.log('Client disconnected');
console.log(`Client disconnected: ${clientId}`);
this.clients.delete(clientId); // 移除断开的客户端
this.emit('end', socket);
});
socket.on('error', (err) => {
console.error('Socket error:', err);
console.error(`Socket error for client ${clientId}:`, err);
this.clients.delete(clientId); // 移除出错的客户端
this.emit('error', err);
});
}
// 新增方法:获取当前连接的客户端数量
getClientCount() {
return this.clients.size;
}
// 新增方法:向所有客户端广播消息
broadcast(message) {
this.clients.forEach((client) => {
client.write(message);
});
}
// 新增方法:停止服务器
stop() {
this.server.close(() => {
console.log('GateServer has been stopped.');
});
}
}
module.exports = GateServer;