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

添加连接管理的互斥锁，增强多线程环境下的连接安全性；更新TCPClient以支持断开回调

Browse Source
This commit is contained in:
Ayndpa
2025-11-18 19:20:36 +08:00
parent b308a644ff
commit 676d39d6a2
5 changed files with 44 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
online_game_tool.cpp
View File

@@ -48,6 +48,7 @@ int g_currentVirtualPort = 0;
std::vector<HSteamNetConnection> connections;
std::map<HSteamNetConnection, TCPClient*> clientMap;
std::mutex clientMutex;
std::mutex connectionsMutex; // Add mutex for connections
int localPort = 0;
bool g_isHost = false;
bool g_isClient = false;
@@ -73,7 +74,10 @@ void OnSteamNetConnectionStatusChanged(SteamNetConnectionStatusChangedCallback_t
{
// Incoming connection, accept it
SteamNetworkingSockets()->AcceptConnection(pInfo->m_hConn);
{
std::lock_guard<std::mutex> lockConn(connectionsMutex);
connections.push_back(pInfo->m_hConn);
}
g_hConnection = pInfo->m_hConn; // Keep for backward compatibility if needed
g_isConnected = true;
std::cout << "Accepted incoming connection from " << pInfo->m_info.m_identityRemote.GetSteamID().ConvertToUint64() << std::endl;
@@ -115,6 +119,8 @@ void OnSteamNetConnectionStatusChanged(SteamNetConnectionStatusChangedCallback_t
g_isConnected = false;
g_hConnection = k_HSteamNetConnection_Invalid;
// Remove from connections
{
std::lock_guard<std::mutex> lockConn(connectionsMutex);
auto it = connections.begin();
while (it != connections.end()) {
if (*it == pInfo->m_hConn) {
@@ -123,6 +129,7 @@ void OnSteamNetConnectionStatusChanged(SteamNetConnectionStatusChangedCallback_t
++it;
}
}
}
// Remove from userMap
userMap.erase(pInfo->m_hConn);
// Cleanup TCP Client
@@ -198,7 +205,7 @@ int main() {
boost::asio::io_context io_context;
// Create Steam Message Handler
SteamMessageHandler messageHandler(io_context, m_pInterface, connections, clientMap, clientMutex, server, g_isHost, localPort);
SteamMessageHandler messageHandler(io_context, m_pInterface, connections, clientMap, clientMutex, connectionsMutex, server, g_isHost, localPort);
// Initialize GLFW
if (!glfwInit()) {
@@ -344,6 +351,7 @@ int main() {
}
{
std::lock_guard<std::mutex> lock(clientMutex);
std::lock_guard<std::mutex> lockConn(connectionsMutex);
ImGui::Text("连接的好友: %d", (int)connections.size());
ImGui::Text("活跃的TCP客户端: %d", (int)clientMap.size());
}

16
steam_message_handler.cpp
View File

@@ -10,8 +10,8 @@
const char* CONTROL_PREFIX = "CONTROL:";
const size_t CONTROL_PREFIX_LEN = 8;
SteamMessageHandler::SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::map<HSteamNetConnection, TCPClient*>& clientMap, std::mutex& clientMutex, std::unique_ptr<TCPServer>& server, bool& g_isHost, int& localPort)
: io_context_(io_context), m_pInterface_(interface), connections_(connections), clientMap_(clientMap), clientMutex_(clientMutex), 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::map<HSteamNetConnection, 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() {
stop();
@@ -48,8 +48,13 @@ void SteamMessageHandler::run() {
}
void SteamMessageHandler::pollMessages() {
std::vector<HSteamNetConnection> currentConnections;
{
std::lock_guard<std::mutex> lockConn(connectionsMutex_);
currentConnections = connections_;
}
std::lock_guard<std::mutex> lock(clientMutex_);
for (auto conn : connections_) {
for (auto conn : currentConnections) {
ISteamNetworkingMessage* pIncomingMsgs[10];
int numMsgs = m_pInterface_->ReceiveMessagesOnConnection(conn, pIncomingMsgs, 10);
for (int i = 0; i < numMsgs; ++i) {
@@ -72,6 +77,11 @@ void SteamMessageHandler::pollMessages() {
std::lock_guard<std::mutex> lock(clientMutex_);
m_pInterface_->SendMessageToConnection(conn, data, size, k_nSteamNetworkingSend_Reliable, nullptr);
});
client->setDisconnectCallback([conn, this]() {
std::lock_guard<std::mutex> lock(clientMutex_);
m_pInterface_->CloseConnection(conn, 0, nullptr, false);
std::cout << "Closed Steam connection due to TCP client disconnect" << std::endl;
});
clientMap_[conn] = client;
std::cout << "Created TCP Client for connection on first message" << std::endl;
} else {

3
steam_message_handler.h
View File

@@ -13,7 +13,7 @@
class SteamMessageHandler {
public:
SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::map<HSteamNetConnection, TCPClient*>& clientMap, std::mutex& clientMutex, std::unique_ptr<TCPServer>& server, bool& g_isHost, int& localPort);
SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::map<HSteamNetConnection, TCPClient*>& clientMap, std::mutex& clientMutex, std::mutex& connectionsMutex, std::unique_ptr<TCPServer>& server, bool& g_isHost, int& localPort);
~SteamMessageHandler();
void start();
@@ -28,6 +28,7 @@ private:
std::vector<HSteamNetConnection>& connections_;
std::map<HSteamNetConnection, TCPClient*>& clientMap_;
std::mutex& clientMutex_;
std::mutex& connectionsMutex_;
std::unique_ptr<TCPServer>& server_;
bool& g_isHost_;
int& localPort_;

11
tcp/tcp_client.cpp
View File

@@ -1,7 +1,7 @@
#include "tcp_client.h"
#include <iostream>
TCPClient::TCPClient(const std::string& host, int port) : host_(host), port_(port), connected_(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) : 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() { disconnect(); }
@@ -26,6 +26,11 @@ bool TCPClient::connect() {
}
void TCPClient::disconnect() {
if (disconnected_) return;
disconnected_ = true;
if (disconnectCallback_) {
disconnectCallback_();
}
connected_ = false;
io_context_.stop();
if (clientThread_.joinable()) {
@@ -59,6 +64,10 @@ 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);

3
tcp/tcp_client.h
View File

@@ -21,6 +21,7 @@ public:
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);
private:
void start_read();
@@ -29,6 +30,7 @@ private:
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_;
@@ -36,5 +38,6 @@ private:
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_;
};