diff --git a/multiplex/multiplex_manager.cpp b/multiplex/multiplex_manager.cpp index 065ff3a..267934e 100644 --- a/multiplex/multiplex_manager.cpp +++ b/multiplex/multiplex_manager.cpp @@ -1,10 +1,11 @@ #include "multiplex_manager.h" +#include "nanoid/nanoid.h" #include #include 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 socket) { +std::string MultiplexManager::addClient(std::shared_ptr socket) { std::lock_guard 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 lock(mapMutex_); auto it = clientMap_.find(id); if (it != clientMap_.end()) { @@ -32,7 +33,7 @@ void MultiplexManager::removeClient(uint32_t id) { } } -std::shared_ptr MultiplexManager::getClient(uint32_t id) { +std::shared_ptr MultiplexManager::getClient(const std::string& id) { std::lock_guard lock(mapMutex_); auto it = clientMap_.find(id); if (it != clientMap_.end()) { @@ -41,31 +42,32 @@ std::shared_ptr 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 packet(packetSize); - uint32_t* pId = reinterpret_cast(&packet[0]); - uint32_t* pType = reinterpret_cast(&packet[sizeof(uint32_t)]); - *pId = id; + std::memcpy(&packet[0], id.c_str(), idLen); + uint32_t* pType = reinterpret_cast(&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(data); - uint32_t type = *reinterpret_cast(data + sizeof(uint32_t)); + std::string id(data, 6); + uint32_t type = *reinterpret_cast(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，创建一个连接到本地端口 diff --git a/multiplex/multiplex_manager.h b/multiplex/multiplex_manager.h index d38f496..76976c5 100644 --- a/multiplex/multiplex_manager.h +++ b/multiplex/multiplex_manager.h @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -17,20 +18,19 @@ public: boost::asio::io_context& io_context, bool& isHost, int& localPort); ~MultiplexManager(); - uint32_t addClient(std::shared_ptr socket); - void removeClient(uint32_t id); - std::shared_ptr getClient(uint32_t id); + std::string addClient(std::shared_ptr socket); + void removeClient(const std::string& id); + std::shared_ptr 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> clientMap_; + std::unordered_map> clientMap_; std::mutex mapMutex_; - uint32_t nextId_; boost::asio::io_context& io_context_; bool& isHost_; int& localPort_; diff --git a/steam/steam_message_handler.cpp b/steam/steam_message_handler.cpp index 1cdbdb4..784e86d 100644 --- a/steam/steam_message_handler.cpp +++ b/steam/steam_message_handler.cpp @@ -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; diff --git a/tcp/tcp_client.cpp b/tcp/tcp_client.cpp index 5cd39b3..adcb989 100644 --- a/tcp/tcp_client.cpp +++ b/tcp/tcp_client.cpp @@ -1,7 +1,8 @@ #include "tcp_client.h" +#include "../steam/steam_networking_manager.h" #include -TCPClient::TCPClient(const std::string& host, int port) : host_(host), port_(port), connected_(false), disconnected_(false), socket_(std::make_shared(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(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(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 call receiveCallbackBytes_ = callback; } -void TCPClient::setDisconnectCallback(std::function 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_(); } diff --git a/tcp/tcp_client.h b/tcp/tcp_client.h index 8e8406e..7019ba9 100644 --- a/tcp/tcp_client.h +++ b/tcp/tcp_client.h @@ -6,13 +6,16 @@ #include #include #include +#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 callback); void setReceiveCallback(std::function callback); void setDisconnectCallback(std::function 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 receiveCallbackBytes_; std::function disconnectCallback_; std::vector buffer_; + SteamNetworkingManager* manager_; + std::unique_ptr multiplexManager_; }; \ No newline at end of file diff --git a/tcp/tcp_server.cpp b/tcp/tcp_server.cpp index 744e951..689849e 100644 --- a/tcp/tcp_server.cpp +++ b/tcp/tcp_server.cpp @@ -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 lock(clientsMutex_); clients_.push_back(socket); @@ -79,7 +79,7 @@ void TCPServer::start_accept() { }); } -void TCPServer::start_read(std::shared_ptr socket, uint32_t id) { +void TCPServer::start_read(std::shared_ptr socket, std::string id) { auto buffer = std::make_shared>(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) { diff --git a/tcp/tcp_server.h b/tcp/tcp_server.h index 059e17f..e111290 100644 --- a/tcp/tcp_server.h +++ b/tcp/tcp_server.h @@ -31,7 +31,7 @@ public: private: void start_accept(); - void start_read(std::shared_ptr socket, uint32_t id); + void start_read(std::shared_ptr socket, std::string id); int port_; bool running_;