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

重构网络管理，移除TCPClient和TCPServer类，整合多路复用功能，优化连接管理和数据传输逻辑

Browse Source
This commit is contained in:
Ayndpa
2025-11-19 18:30:52 +08:00
parent dc44cf4df1
commit dc3535f8c2
11 changed files with 16 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
multiplex/multiplex_manager.cpp → net/multiplex_manager.cpp
View File

0
multiplex/multiplex_manager.h → net/multiplex_manager.h
View File

0
tcp/tcp_server.cpp → net/tcp_server.cpp
View File

2
tcp/tcp_server.h → net/tcp_server.h
View File

@@ -10,7 +10,7 @@
#include <isteamnetworkingsockets.h>
#include <isteamnetworkingutils.h>
#include <steamnetworkingtypes.h>
#include "../multiplex/multiplex_manager.h"
#include "multiplex_manager.h"
class SteamNetworkingManager;

16
online_game_tool.cpp
View File

@@ -13,7 +13,6 @@
#include <boost/asio.hpp>
#include <memory>
#include "tcp_server.h"
#include "tcp/tcp_client.h"
#include "steam/steam_networking_manager.h"
#include "steam/steam_room_manager.h"
#include "steam/steam_utils.h"
@@ -22,8 +21,6 @@ using boost::asio::ip::tcp;
// New variables for multiple connections and TCP clients
std::vector<HSteamNetConnection> connections;
std::map<HSteamNetConnection, std::shared_ptr<TCPClient>> clientMap;
std::mutex clientMutex;
std::mutex connectionsMutex; // Add mutex for connections
int localPort = 0;
std::unique_ptr<TCPServer> server;
@@ -73,7 +70,7 @@ int main() {
ImGui_ImplOpenGL3_Init("#version 130");
// Set message handler dependencies
steamManager.setMessageHandlerDependencies(io_context, clientMap, clientMutex, server, localPort);
steamManager.setMessageHandlerDependencies(io_context, server, localPort);
steamManager.startMessageHandler();
// Steam Networking variables
@@ -185,10 +182,8 @@ int main() {
ImGui::Text("房间内玩家: %d", server->getClientCount() + 1); // +1 for host
}
{
std::lock_guard<std::mutex> lock(clientMutex);
std::lock_guard<std::mutex> lockConn(connectionsMutex);
ImGui::Text("连接的好友: %d", (int)steamManager.getConnections().size());
ImGui::Text("活跃的TCP客户端: %d", (int)clientMap.size());
}
ImGui::Separator();
ImGui::Text("用户列表:");
@@ -198,7 +193,6 @@ int main() {
ImGui::TableSetupColumn("连接类型");
ImGui::TableHeadersRow();
{
std::lock_guard<std::mutex> lock(clientMutex);
for (const auto& pair : steamManager.getUserMap()) {
ImGui::TableNextRow();
ImGui::TableNextColumn();
@@ -231,14 +225,6 @@ int main() {
steamManager.stopMessageHandler();
// Cleanup
// Cleanup TCP Clients
{
std::lock_guard<std::mutex> lock(clientMutex);
for (auto& pair : clientMap) {
pair.second->disconnect();
}
clientMap.clear();
}
if (server) {
server->stop();
}

10
steam/steam_message_handler.cpp
View File

@@ -6,8 +6,8 @@
#include <steam_api.h>
#include <isteamnetworkingsockets.h>
SteamMessageHandler::SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::map<HSteamNetConnection, std::shared_ptr<TCPClient>>& clientMap, std::mutex& clientMutex, std::mutex& connectionsMutex, std::unique_ptr<TCPServer>& server, bool& g_isHost, int& localPort)
: io_context_(io_context), m_pInterface_(interface), connections_(connections), clientMap_(clientMap), clientMutex_(clientMutex), connectionsMutex_(connectionsMutex), server_(server), g_isHost_(g_isHost), localPort_(localPort), running_(false) {}
SteamMessageHandler::SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::mutex& connectionsMutex, bool& g_isHost, int& localPort)
: io_context_(io_context), m_pInterface_(interface), connections_(connections), connectionsMutex_(connectionsMutex), g_isHost_(g_isHost), localPort_(localPort), running_(false) {}
SteamMessageHandler::~SteamMessageHandler() {
stop();
@@ -46,7 +46,6 @@ void SteamMessageHandler::pollMessages() {
std::lock_guard<std::mutex> lockConn(connectionsMutex_);
currentConnections = connections_;
}
std::lock_guard<std::mutex> lock(clientMutex_);
for (auto conn : currentConnections) {
ISteamNetworkingMessage* pIncomingMsgs[10];
int numMsgs = m_pInterface_->ReceiveMessagesOnConnection(conn, pIncomingMsgs, 10);
@@ -56,9 +55,10 @@ void SteamMessageHandler::pollMessages() {
const char* data = (const char*)pIncomingMsg->m_pData;
size_t size = pIncomingMsg->m_cbSize;
// Handle tunnel packets with multiplexing
if (server_ && server_->getMultiplexManager()) {
server_->getMultiplexManager()->handleTunnelPacket(data, size);
if (multiplexManagers_.find(conn) == multiplexManagers_.end()) {
multiplexManagers_[conn] = std::make_shared<MultiplexManager>(m_pInterface_, conn, io_context_, g_isHost_, localPort_);
}
multiplexManagers_[conn]->handleTunnelPacket(data, size);
pIncomingMsg->Release();
}
}

11
steam/steam_message_handler.h
View File

@@ -8,12 +8,12 @@
#include <memory>
#include <boost/asio.hpp>
#include <steamnetworkingtypes.h>
#include "tcp_server.h"
#include "tcp/tcp_client.h"
#include "../net/tcp_server.h"
#include "../net/multiplex_manager.h"
class SteamMessageHandler {
public:
SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::map<HSteamNetConnection, std::shared_ptr<TCPClient>>& clientMap, std::mutex& clientMutex, std::mutex& connectionsMutex, std::unique_ptr<TCPServer>& server, bool& g_isHost, int& localPort);
SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::mutex& connectionsMutex, bool& g_isHost, int& localPort);
~SteamMessageHandler();
void start();
@@ -26,13 +26,12 @@ private:
boost::asio::io_context& io_context_;
ISteamNetworkingSockets* m_pInterface_;
std::vector<HSteamNetConnection>& connections_;
std::map<HSteamNetConnection, std::shared_ptr<TCPClient>>& clientMap_;
std::mutex& clientMutex_;
std::mutex& connectionsMutex_;
std::unique_ptr<TCPServer>& server_;
bool& g_isHost_;
int& localPort_;
std::map<HSteamNetConnection, std::shared_ptr<MultiplexManager>> multiplexManagers_;
std::thread thread_;
bool running_;
};

8
steam/steam_networking_manager.cpp
View File

@@ -16,7 +16,7 @@ void SteamNetworkingManager::OnSteamNetConnectionStatusChanged(SteamNetConnectio
SteamNetworkingManager::SteamNetworkingManager()
: m_pInterface(nullptr), hListenSock(k_HSteamListenSocket_Invalid), g_isHost(false), g_isClient(false), g_isConnected(false),
g_hConnection(k_HSteamNetConnection_Invalid),
io_context_(nullptr), clientMap_(nullptr), clientMutex_(nullptr), server_(nullptr), localPort_(nullptr), messageHandler_(nullptr)
io_context_(nullptr), server_(nullptr), localPort_(nullptr), messageHandler_(nullptr)
{
std::cout << "Initialized SteamNetworkingManager" << std::endl;
}
@@ -129,14 +129,12 @@ bool SteamNetworkingManager::joinHost(uint64 hostID)
}
}
void SteamNetworkingManager::setMessageHandlerDependencies(boost::asio::io_context &io_context, std::map<HSteamNetConnection, std::shared_ptr<TCPClient>> &clientMap, std::mutex &clientMutex, std::unique_ptr<TCPServer> &server, int &localPort)
void SteamNetworkingManager::setMessageHandlerDependencies(boost::asio::io_context &io_context, std::unique_ptr<TCPServer> &server, int &localPort)
{
io_context_ = &io_context;
clientMap_ = &clientMap;
clientMutex_ = &clientMutex;
server_ = &server;
localPort_ = &localPort;
messageHandler_ = new SteamMessageHandler(io_context, m_pInterface, connections, clientMap, clientMutex, connectionsMutex, server, g_isHost, localPort);
messageHandler_ = new SteamMessageHandler(io_context, m_pInterface, connections, connectionsMutex, g_isHost, localPort);
}
void SteamNetworkingManager::startMessageHandler()

7
steam/steam_networking_manager.h
View File

@@ -12,7 +12,6 @@
#include "steam_message_handler.h"
// Forward declarations
class TCPClient;
class TCPServer;
class SteamNetworkingManager;
@@ -50,13 +49,11 @@ public:
std::unique_ptr<TCPServer>*& getServer() { return server_; }
int*& getLocalPort() { return localPort_; }
boost::asio::io_context*& getIOContext() { return io_context_; }
std::map<HSteamNetConnection, std::shared_ptr<TCPClient>>*& getClientMap() { return clientMap_; }
std::mutex*& getClientMutex() { return clientMutex_; }
HSteamListenSocket& getListenSock() { return hListenSock; }
ISteamNetworkingSockets* getInterface() { return m_pInterface; }
bool& getIsHost() { return g_isHost; }
void setMessageHandlerDependencies(boost::asio::io_context& io_context, std::map<HSteamNetConnection, std::shared_ptr<TCPClient>>& clientMap, std::mutex& clientMutex, std::unique_ptr<TCPServer>& server, int& localPort);
void setMessageHandlerDependencies(boost::asio::io_context& io_context, std::unique_ptr<TCPServer>& server, int& localPort);
// Message handler
void startMessageHandler();
@@ -93,8 +90,6 @@ private:
// Message handler dependencies
boost::asio::io_context* io_context_;
std::map<HSteamNetConnection, std::shared_ptr<TCPClient>>* clientMap_;
std::mutex* clientMutex_;
std::unique_ptr<TCPServer>* server_;
int* localPort_;
SteamMessageHandler* messageHandler_;

115
tcp/tcp_client.cpp
View File

@@ -1,115 +0,0 @@
#include "tcp_client.h"
#include "../steam/steam_networking_manager.h"
#include <iostream>
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(); }
bool TCPClient::connect() {
try {
tcp::resolver resolver(io_context_);
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(id);
std::cout << "Connected to " << host_ << ":" << port_ << std::endl;
return true;
} catch (const std::exception& e) {
std::cerr << "Failed to connect: " << e.what() << std::endl;
return false;
}
}
void TCPClient::disconnect() {
if (disconnected_) return;
disconnected_ = true;
connected_ = false;
io_context_.stop();
if (clientThread_.joinable()) {
if (clientThread_.get_id() == std::this_thread::get_id()) {
clientThread_.detach();
} else {
clientThread_.join();
}
}
try {
if (socket_->is_open()) {
socket_->close();
}
} catch (const std::exception& e) {
std::cerr << "Error closing socket: " << e.what() << std::endl;
}
multiplexManager_.reset();
}
void TCPClient::send(const std::string& message) {
send(message.c_str(), message.size());
}
void TCPClient::send(const char* data, size_t size) {
if (!connected_) return;
// std::cout << "Sending " << size << " bytes" << std::endl;
boost::asio::async_write(*socket_, boost::asio::buffer(data, size), [this](const boost::system::error_code& error, std::size_t) {
if (error) {
std::cerr << "Send failed: " << error.message() << std::endl;
disconnect();
}
});
}
void TCPClient::setReceiveCallback(std::function<void(const std::string&)> callback) {
receiveCallback_ = callback;
}
void TCPClient::setReceiveCallback(std::function<void(const char*, size_t)> callback) {
receiveCallbackBytes_ = callback;
}
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(std::string id, const boost::system::error_code& error, std::size_t bytes_transferred) {
if (!error) {
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(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_();
}
disconnect();
}
}

49
tcp/tcp_client.h
View File

@@ -1,49 +0,0 @@
#pragma once
#include <boost/asio.hpp>
#include <memory>
#include <string>
#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, SteamNetworkingManager* manager);
~TCPClient();
bool connect();
void disconnect();
void send(const std::string& message);
void send(const char* data, size_t size);
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(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_;
bool connected_;
bool disconnected_;
boost::asio::io_context io_context_;
boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_;
std::shared_ptr<tcp::socket> socket_;
std::thread clientThread_;
std::mutex socketMutex_;
std::function<void(const std::string&)> receiveCallback_;
std::function<void(const char*, size_t)> receiveCallbackBytes_;
std::function<void()> disconnectCallback_;
std::vector<char> buffer_;
SteamNetworkingManager* manager_;
std::unique_ptr<MultiplexManager> multiplexManager_;
};