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

重构MultiplexManager和TCPServer，优化客户端管理和连接逻辑，添加异步读取和错误处理

Browse Source
This commit is contained in:
Ayndpa
2025-11-19 20:25:25 +08:00
parent 1991cc9a5d
commit cd7a035ba8
6 changed files with 116 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
net/multiplex_manager.cpp
View File

@@ -8,44 +8,58 @@ MultiplexManager::MultiplexManager(ISteamNetworkingSockets* steamInterface, HSte
: steamInterface_(steamInterface), steamConn_(steamConn),
io_context_(io_context), isHost_(isHost), localPort_(localPort) {}
MultiplexManager::~MultiplexManager() {
MultiplexManager::~MultiplexManager()
{
// Close all sockets
std::lock_guard<std::mutex> lock(mapMutex_);
for (auto& pair : clientMap_) {
for (auto &pair : clientMap_)
{
pair.second->close();
}
clientMap_.clear();
}
std::string MultiplexManager::addClient(std::shared_ptr<tcp::socket> socket) {
std::string MultiplexManager::addClient(std::shared_ptr<tcp::socket> socket)
{
std::string id;
{
std::lock_guard<std::mutex> lock(mapMutex_);
std::string id = nanoid::generate(6);
id = nanoid::generate(6);
clientMap_[id] = socket;
readBuffers_[id].resize(1024);
}
startAsyncRead(id);
std::cout << "Added client with id " << id << std::endl;
return id;
}
void MultiplexManager::removeClient(const std::string& id) {
void MultiplexManager::removeClient(const std::string &id)
{
std::lock_guard<std::mutex> lock(mapMutex_);
auto it = clientMap_.find(id);
if (it != clientMap_.end()) {
if (it != clientMap_.end())
{
it->second->close();
clientMap_.erase(it);
}
readBuffers_.erase(id);
std::cout << "Removed client with id " << id << std::endl;
}
std::shared_ptr<tcp::socket> MultiplexManager::getClient(const std::string& 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()) {
if (it != clientMap_.end())
{
return it->second;
}
return nullptr;
}
void MultiplexManager::sendTunnelPacket(const std::string& id, const char* data, size_t len, int type) {
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);
@@ -53,68 +67,99 @@ void MultiplexManager::sendTunnelPacket(const std::string& id, const char* data,
std::memcpy(&packet[0], id.c_str(), idLen);
uint32_t *pType = reinterpret_cast<uint32_t *>(&packet[idLen]);
*pType = type;
if (type == 0 && data) {
if (type == 0 && data)
{
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) {
void MultiplexManager::handleTunnelPacket(const char *data, size_t len)
{
size_t idLen = 7; // 6 + null
if (len < idLen + sizeof(uint32_t)) {
if (len < idLen + sizeof(uint32_t))
{
std::cerr << "Invalid tunnel packet size" << std::endl;
return;
}
std::string id(data, 6);
uint32_t type = *reinterpret_cast<const uint32_t *>(data + idLen);
if (type == 0) {
if (type == 0)
{
// Data packet
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) {
if (!socket && isHost_ && localPort_ > 0)
{
// 如果是主持且没有对应的 TCP Client，创建一个连接到本地端口
std::cout << "Creating new TCP client for id " << id << " connecting to localhost:" << localPort_ << std::endl;
try {
try
{
auto newSocket = std::make_shared<tcp::socket>(io_context_);
tcp::resolver resolver(io_context_);
auto endpoints = resolver.resolve("127.0.0.1", std::to_string(localPort_));
boost::asio::connect(*newSocket, endpoints);
std::string tempId = id;
{
std::lock_guard<std::mutex> lock(mapMutex_);
clientMap_[id] = newSocket;
readBuffers_[id].resize(1024);
socket = newSocket;
}
std::cout << "Successfully created TCP client for id " << id << std::endl;
startAsyncRead(id);
} catch (const std::exception& e) {
startAsyncRead(tempId);
}
catch (const std::exception &e)
{
std::cerr << "Failed to create TCP client for id " << id << ": " << e.what() << std::endl;
return;
}
}
if (socket) {
if (socket)
{
boost::asio::async_write(*socket, boost::asio::buffer(packetData, dataLen), [](const boost::system::error_code &, std::size_t) {});
} else {
}
else
{
std::cerr << "No client found for id " << id << std::endl;
}
} else if (type == 1) {
}
else if (type == 1)
{
// Disconnect packet
removeClient(id);
std::cout << "Client " << id << " disconnected" << std::endl;
} else {
}
else
{
std::cerr << "Unknown packet type " << type << std::endl;
}
}
void MultiplexManager::startAsyncRead(const std::string& id) {
void MultiplexManager::startAsyncRead(const std::string &id)
{
auto socket = getClient(id);
if (!socket || readBuffers_.find(id) == readBuffers_.end()) return;
if (!socket)
{
std::cout << "Error: Socket is null for id " << id << std::endl;
return;
}
socket->async_read_some(boost::asio::buffer(readBuffers_[id]),
[this, id](const boost::system::error_code& ec, std::size_t bytes_transferred) {
if (!ec && bytes_transferred > 0) {
[this, id](const boost::system::error_code &ec, std::size_t bytes_transferred)
{
if (!ec)
{
if (bytes_transferred > 0)
{
sendTunnelPacket(id, readBuffers_[id].data(), bytes_transferred, 0);
}
startAsyncRead(id);
} else {
}
else
{
std::cout << "Error reading from TCP client " << id << ": " << ec.message() << std::endl;
removeClient(id);
}
});

18
net/tcp_server.cpp
View File

@@ -15,9 +15,6 @@ bool TCPServer::start() {
acceptor_.bind(endpoint);
acceptor_.listen();
multiplexManager_ = std::make_unique<MultiplexManager>(manager_->getInterface(), manager_->getConnection(),
io_context_, manager_->getIsHost(), *manager_->getLocalPort());
running_ = true;
serverThread_ = std::thread([this]() {
std::cout << "Server thread started" << std::endl;
@@ -40,7 +37,6 @@ void TCPServer::stop() {
serverThread_.join();
}
acceptor_.close();
multiplexManager_.reset();
}
void TCPServer::sendToAll(const std::string& message, std::shared_ptr<tcp::socket> excludeSocket) {
@@ -66,7 +62,8 @@ 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;
std::string id = multiplexManager_->addClient(socket);
auto multiplexManager = manager_->getMessageHandler()->getMultiplexManager(manager_->getConnection());
std::string id = multiplexManager->addClient(socket);
{
std::lock_guard<std::mutex> lock(clientsMutex_);
clients_.push_back(socket);
@@ -83,9 +80,9 @@ 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) {
std::cout << "Received " << bytes_transferred << " bytes from TCP client " << id << std::endl;
if (manager_->isConnected()) {
multiplexManager_->sendTunnelPacket(id, buffer->data(), bytes_transferred, 0);
auto multiplexManager = manager_->getMessageHandler()->getMultiplexManager(manager_->getConnection());
multiplexManager->sendTunnelPacket(id, buffer->data(), bytes_transferred, 0);
} else {
std::cout << "Not connected to Steam, skipping forward" << std::endl;
}
@@ -95,10 +92,11 @@ void TCPServer::start_read(std::shared_ptr<tcp::socket> socket, std::string id)
std::cout << "TCP client " << id << " disconnected or error: " << error.message() << std::endl;
// Send disconnect packet
if (manager_->isConnected()) {
multiplexManager_->sendTunnelPacket(id, nullptr, 0, 1);
}
auto multiplexManager = manager_->getMessageHandler()->getMultiplexManager(manager_->getConnection());
multiplexManager->sendTunnelPacket(id, nullptr, 0, 1);
// Remove client
multiplexManager_->removeClient(id);
multiplexManager->removeClient(id);
}
std::lock_guard<std::mutex> lock(clientsMutex_);
clients_.erase(std::remove(clients_.begin(), clients_.end(), socket), clients_.end());
}

2
net/tcp_server.h
View File

@@ -27,7 +27,6 @@ public:
void sendToAll(const std::string& message, std::shared_ptr<tcp::socket> excludeSocket = nullptr);
void sendToAll(const char* data, size_t size, std::shared_ptr<tcp::socket> excludeSocket = nullptr);
int getClientCount();
MultiplexManager* getMultiplexManager() { return multiplexManager_.get(); }
private:
void start_accept();
@@ -42,5 +41,4 @@ private:
std::mutex clientsMutex_;
std::thread serverThread_;
SteamNetworkingManager* manager_;
std::unique_ptr<MultiplexManager> multiplexManager_;
};

17
steam/steam_message_handler.cpp
View File

@@ -17,14 +17,30 @@ void SteamMessageHandler::start() {
if (running_) return;
running_ = true;
thread_ = std::thread([this]() { run(); });
// Start io_context in a separate thread
io_thread_ = std::thread([this]() {
auto work = boost::asio::make_work_guard(io_context_);
io_context_.run();
});
}
void SteamMessageHandler::stop() {
if (!running_) return;
running_ = false;
io_context_.stop();
if (thread_.joinable()) {
thread_.join();
}
if (io_thread_.joinable()) {
io_thread_.join();
}
}
std::shared_ptr<MultiplexManager> SteamMessageHandler::getMultiplexManager(HSteamNetConnection conn) {
if (multiplexManagers_.find(conn) == multiplexManagers_.end()) {
multiplexManagers_[conn] = std::make_shared<MultiplexManager>(m_pInterface_, conn, io_context_, g_isHost_, localPort_);
}
return multiplexManagers_[conn];
}
void SteamMessageHandler::run() {
@@ -50,7 +66,6 @@ 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;

3
steam/steam_message_handler.h
View File

@@ -19,6 +19,8 @@ public:
void start();
void stop();
std::shared_ptr<MultiplexManager> getMultiplexManager(HSteamNetConnection conn);
private:
void run();
void pollMessages();
@@ -33,6 +35,7 @@ private:
std::map<HSteamNetConnection, std::shared_ptr<MultiplexManager>> multiplexManagers_;
std::thread thread_;
std::thread io_thread_;
bool running_;
};

1
steam/steam_networking_manager.h
View File

@@ -58,6 +58,7 @@ public:
// Message handler
void startMessageHandler();
void stopMessageHandler();
SteamMessageHandler* getMessageHandler() { return messageHandler_; }
// Update user info (ping, relay status)
void update();