public/ConnectTools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构MultiplexManager，使用字符串ID替代uint32_t，优化TCPClient和TCPServer的客户端管理

Browse Source
This commit is contained in:
Ayndpa
2025-11-19 18:19:14 +08:00
parent 2e0febc3a3
commit dc44cf4df1
7 changed files with 62 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
multiplex/multiplex_manager.cpp
View File

@@ -1,10 +1,11 @@
#include "multiplex_manager.h"
#include "nanoid/nanoid.h"
#include <iostream>
#include <cstring>
MultiplexManager::MultiplexManager(ISteamNetworkingSockets* steamInterface, HSteamNetConnection steamConn,
boost::asio::io_context& io_context, bool& isHost, int& localPort)
: steamInterface_(steamInterface), steamConn_(steamConn), nextId_(1),
: steamInterface_(steamInterface), steamConn_(steamConn),
io_context_(io_context), isHost_(isHost), localPort_(localPort) {}
MultiplexManager::~MultiplexManager() {
@@ -16,14 +17,14 @@ MultiplexManager::~MultiplexManager() {
clientMap_.clear();
}
uint32_t MultiplexManager::addClient(std::shared_ptr<tcp::socket> socket) {
std::string MultiplexManager::addClient(std::shared_ptr<tcp::socket> socket) {
std::lock_guard<std::mutex> lock(mapMutex_);
uint32_t id = nextId_++;
std::string id = nanoid::generate(6);
clientMap_[id] = socket;
return id;
}
void MultiplexManager::removeClient(uint32_t id) {
void MultiplexManager::removeClient(const std::string& id) {
std::lock_guard<std::mutex> lock(mapMutex_);
auto it = clientMap_.find(id);
if (it != clientMap_.end()) {
@@ -32,7 +33,7 @@ void MultiplexManager::removeClient(uint32_t id) {
}
}
std::shared_ptr<tcp::socket> MultiplexManager::getClient(uint32_t id) {
std::shared_ptr<tcp::socket> MultiplexManager::getClient(const std::string& id) {
std::lock_guard<std::mutex> lock(mapMutex_);
auto it = clientMap_.find(id);
if (it != clientMap_.end()) {
@@ -41,31 +42,32 @@ std::shared_ptr<tcp::socket> MultiplexManager::getClient(uint32_t id) {
return nullptr;
}
void MultiplexManager::sendTunnelPacket(uint32_t id, const char* data, size_t len, int type) {
// Packet format: uint32_t id, uint32_t type, then data if type==0
size_t packetSize = sizeof(uint32_t) * 2 + (type == 0 ? len : 0);
void MultiplexManager::sendTunnelPacket(const std::string& id, const char* data, size_t len, int type) {
// Packet format: string id (6 chars + null), uint32_t type, then data if type==0
size_t idLen = id.size() + 1; // include null terminator
size_t packetSize = idLen + sizeof(uint32_t) + (type == 0 ? len : 0);
std::vector<char> packet(packetSize);
uint32_t* pId = reinterpret_cast<uint32_t*>(&packet[0]);
uint32_t* pType = reinterpret_cast<uint32_t*>(&packet[sizeof(uint32_t)]);
*pId = id;
std::memcpy(&packet[0], id.c_str(), idLen);
uint32_t* pType = reinterpret_cast<uint32_t*>(&packet[idLen]);
*pType = type;
if (type == 0 && data) {
std::memcpy(&packet[sizeof(uint32_t) * 2], data, len);
std::memcpy(&packet[idLen + sizeof(uint32_t)], data, len);
}
steamInterface_->SendMessageToConnection(steamConn_, packet.data(), packet.size(), k_nSteamNetworkingSend_Reliable, nullptr);
}
void MultiplexManager::handleTunnelPacket(const char* data, size_t len) {
if (len < sizeof(uint32_t) * 2) {
size_t idLen = 7; // 6 + null
if (len < idLen + sizeof(uint32_t)) {
std::cerr << "Invalid tunnel packet size" << std::endl;
return;
}
uint32_t id = *reinterpret_cast<const uint32_t*>(data);
uint32_t type = *reinterpret_cast<const uint32_t*>(data + sizeof(uint32_t));
std::string id(data, 6);
uint32_t type = *reinterpret_cast<const uint32_t*>(data + idLen);
if (type == 0) {
// Data packet
size_t dataLen = len - sizeof(uint32_t) * 2;
const char* packetData = data + sizeof(uint32_t) * 2;
size_t dataLen = len - idLen - sizeof(uint32_t);
const char* packetData = data + idLen + sizeof(uint32_t);
auto socket = getClient(id);
if (!socket && isHost_ && localPort_ > 0) {
// 如果是主持且没有对应的 TCP Client，创建一个连接到本地端口

12
multiplex/multiplex_manager.h
View File

@@ -4,6 +4,7 @@
#include <memory>
#include <mutex>
#include <vector>
#include <string>
#include <boost/asio.hpp>
#include <steam_api.h>
#include <isteamnetworkingsockets.h>
@@ -17,20 +18,19 @@ public:
boost::asio::io_context& io_context, bool& isHost, int& localPort);
~MultiplexManager();
uint32_t addClient(std::shared_ptr<tcp::socket> socket);
void removeClient(uint32_t id);
std::shared_ptr<tcp::socket> getClient(uint32_t id);
std::string addClient(std::shared_ptr<tcp::socket> socket);
void removeClient(const std::string& id);
std::shared_ptr<tcp::socket> getClient(const std::string& id);
void sendTunnelPacket(uint32_t id, const char* data, size_t len, int type);
void sendTunnelPacket(const std::string& id, const char* data, size_t len, int type);
void handleTunnelPacket(const char* data, size_t len);
private:
ISteamNetworkingSockets* steamInterface_;
HSteamNetConnection steamConn_;
std::unordered_map<uint32_t, std::shared_ptr<tcp::socket>> clientMap_;
std::unordered_map<std::string, std::shared_ptr<tcp::socket>> clientMap_;
std::mutex mapMutex_;
uint32_t nextId_;
boost::asio::io_context& io_context_;
bool& isHost_;
int& localPort_;

1
steam/steam_message_handler.cpp
View File

@@ -51,6 +51,7 @@ void SteamMessageHandler::pollMessages() {
ISteamNetworkingMessage* pIncomingMsgs[10];
int numMsgs = m_pInterface_->ReceiveMessagesOnConnection(conn, pIncomingMsgs, 10);
for (int i = 0; i < numMsgs; ++i) {
std::cout << "Received message on connection " << conn << std::endl;
ISteamNetworkingMessage* pIncomingMsg = pIncomingMsgs[i];
const char* data = (const char*)pIncomingMsg->m_pData;
size_t size = pIncomingMsg->m_cbSize;

36
tcp/tcp_client.cpp
View File

@@ -1,7 +1,8 @@
#include "tcp_client.h"
#include "../steam/steam_networking_manager.h"
#include <iostream>
TCPClient::TCPClient(const std::string& host, int port) : host_(host), port_(port), connected_(false), disconnected_(false), socket_(std::make_shared<tcp::socket>(io_context_)), work_(boost::asio::make_work_guard(io_context_)), buffer_(1024) {}
TCPClient::TCPClient(const std::string& host, int port, SteamNetworkingManager* manager) : host_(host), port_(port), connected_(false), disconnected_(false), socket_(std::make_shared<tcp::socket>(io_context_)), work_(boost::asio::make_work_guard(io_context_)), buffer_(1024), manager_(manager) {}
TCPClient::~TCPClient() { disconnect(); }
@@ -11,12 +12,15 @@ bool TCPClient::connect() {
auto endpoints = resolver.resolve(host_, std::to_string(port_));
boost::asio::connect(*socket_, endpoints);
connected_ = true;
multiplexManager_ = std::make_unique<MultiplexManager>(manager_->getInterface(), manager_->getConnection(),
io_context_, manager_->getIsHost(), *manager_->getLocalPort());
std::string id = multiplexManager_->addClient(socket_);
clientThread_ = std::thread([this]() {
std::cout << "Client thread started" << std::endl;
io_context_.run();
std::cout << "Client thread stopped" << std::endl;
});
start_read();
start_read(id);
std::cout << "Connected to " << host_ << ":" << port_ << std::endl;
return true;
} catch (const std::exception& e) {
@@ -44,6 +48,7 @@ void TCPClient::disconnect() {
} catch (const std::exception& e) {
std::cerr << "Error closing socket: " << e.what() << std::endl;
}
multiplexManager_.reset();
}
void TCPClient::send(const std::string& message) {
@@ -69,32 +74,39 @@ void TCPClient::setReceiveCallback(std::function<void(const char*, size_t)> call
receiveCallbackBytes_ = callback;
}
void TCPClient::setDisconnectCallback(std::function<void()> callback) {
disconnectCallback_ = callback;
}
void TCPClient::start_read() {
socket_->async_read_some(boost::asio::buffer(buffer_), [this](const boost::system::error_code& error, std::size_t bytes_transferred) {
handle_read(error, bytes_transferred);
void TCPClient::start_read(std::string id) {
socket_->async_read_some(boost::asio::buffer(buffer_), [this, id](const boost::system::error_code& error, std::size_t bytes_transferred) {
handle_read(id, error, bytes_transferred);
});
}
void TCPClient::handle_read(const boost::system::error_code& error, std::size_t bytes_transferred) {
void TCPClient::handle_read(std::string id, const boost::system::error_code& error, std::size_t bytes_transferred) {
if (!error) {
// std::cout << "Received " << bytes_transferred << " bytes" << std::endl;
std::cout << "Received " << bytes_transferred << " bytes from TCP server" << std::endl;
if (manager_->isConnected()) {
multiplexManager_->sendTunnelPacket(id, buffer_.data(), bytes_transferred, 0);
} else {
std::cout << "Not connected to Steam, skipping forward" << std::endl;
}
if (receiveCallbackBytes_) {
receiveCallbackBytes_(buffer_.data(), bytes_transferred);
} else if (receiveCallback_) {
std::string message(buffer_.data(), bytes_transferred);
receiveCallback_(message);
}
start_read();
start_read(id);
} else {
if (error == boost::asio::error::eof) {
std::cout << "Connection closed by peer" << std::endl;
} else {
std::cerr << "Read failed: " << error.message() << std::endl;
}
// Send disconnect packet
if (manager_->isConnected()) {
multiplexManager_->sendTunnelPacket(id, nullptr, 0, 1);
}
// Remove client
multiplexManager_->removeClient(id);
if (disconnectCallback_) {
disconnectCallback_();
}

12
tcp/tcp_client.h
View File

@@ -6,13 +6,16 @@
#include <thread>
#include <mutex>
#include <functional>
#include "../multiplex/multiplex_manager.h"
using boost::asio::ip::tcp;
class SteamNetworkingManager;
// TCP Client class
class TCPClient {
public:
TCPClient(const std::string& host, int port);
TCPClient(const std::string& host, int port, SteamNetworkingManager* manager);
~TCPClient();
bool connect();
@@ -22,10 +25,11 @@ public:
void setReceiveCallback(std::function<void(const std::string&)> callback);
void setReceiveCallback(std::function<void(const char*, size_t)> callback);
void setDisconnectCallback(std::function<void()> callback);
MultiplexManager* getMultiplexManager() { return multiplexManager_.get(); }
private:
void start_read();
void handle_read(const boost::system::error_code& error, std::size_t bytes_transferred);
void start_read(std::string id);
void handle_read(std::string id, const boost::system::error_code& error, std::size_t bytes_transferred);
std::string host_;
int port_;
@@ -40,4 +44,6 @@ private:
std::function<void(const char*, size_t)> receiveCallbackBytes_;
std::function<void()> disconnectCallback_;
std::vector<char> buffer_;
SteamNetworkingManager* manager_;
std::unique_ptr<MultiplexManager> multiplexManager_;
};

4
tcp/tcp_server.cpp
View File

@@ -66,7 +66,7 @@ void TCPServer::start_accept() {
acceptor_.async_accept(*socket, [this, socket](const boost::system::error_code& error) {
if (!error) {
std::cout << "New client connected" << std::endl;
uint32_t id = multiplexManager_->addClient(socket);
std::string id = multiplexManager_->addClient(socket);
{
std::lock_guard<std::mutex> lock(clientsMutex_);
clients_.push_back(socket);
@@ -79,7 +79,7 @@ void TCPServer::start_accept() {
});
}
void TCPServer::start_read(std::shared_ptr<tcp::socket> socket, uint32_t id) {
void TCPServer::start_read(std::shared_ptr<tcp::socket> socket, std::string id) {
auto buffer = std::make_shared<std::vector<char>>(1024);
socket->async_read_some(boost::asio::buffer(*buffer), [this, socket, buffer, id](const boost::system::error_code& error, std::size_t bytes_transferred) {
if (!error) {

2
tcp/tcp_server.h
View File

@@ -31,7 +31,7 @@ public:
private:
void start_accept();
void start_read(std::shared_ptr<tcp::socket> socket, uint32_t id);
void start_read(std::shared_ptr<tcp::socket> socket, std::string id);
int port_;
bool running_;