#ifdef NETWORK #include "WebSocketClient.h" #include #include // Вспомогательный split std::vector split(const std::string& s, char delimiter) { std::vector tokens; std::string token; std::istringstream tokenStream(s); while (std::getline(tokenStream, token, delimiter)) { tokens.push_back(token); } return tokens; } namespace ZL { void WebSocketClient::Connect(const std::string& host, uint16_t port) { try { boost::asio::ip::tcp::resolver resolver(ioc_); auto const results = resolver.resolve(host, std::to_string(port)); ws_ = std::make_unique>(ioc_); // Выполняем синхронный коннект и handshake для простоты старта boost::beast::get_lowest_layer(*ws_).connect(results); ws_->handshake(host, "/"); connected = true; // Запускаем асинхронное чтение в пуле потоков TaskManager startAsyncRead(); } catch (std::exception& e) { std::cerr << "Network Error: " << e.what() << std::endl; } } void WebSocketClient::startAsyncRead() { ws_->async_read(buffer_, [this](boost::beast::error_code ec, std::size_t bytes) { if (!ec) { std::string msg = boost::beast::buffers_to_string(buffer_.data()); buffer_.consume(bytes); processIncomingMessage(msg); startAsyncRead(); } else { connected = false; } }); } void WebSocketClient::processIncomingMessage(const std::string& msg) { // Логика парсинга... if (msg.rfind("ID:", 0) == 0) { clientId = std::stoi(msg.substr(3)); } // Безопасно кладем в очередь для главного потока std::lock_guard lock(queueMutex); messageQueue.push(msg); } void WebSocketClient::Poll() { std::lock_guard lock(queueMutex); while (!messageQueue.empty()) { auto nowTime = std::chrono::system_clock::now(); //Apply server delay: nowTime -= std::chrono::milliseconds(CLIENT_DELAY); auto now_ms = std::chrono::duration_cast( nowTime.time_since_epoch() ).count(); std::string msg = messageQueue.front(); messageQueue.pop(); if (msg.rfind("EVENT:", 0) == 0) { auto parts = split(msg, ':'); if (parts.size() < 5) continue; // EVENT:ID:TYPE:TIME:DATA... int remoteId = std::stoi(parts[1]); std::string subType = parts[2]; uint64_t sentTime = std::stoull(parts[3]); ClientState remoteState; remoteState.id = remoteId; std::chrono::system_clock::time_point uptime_timepoint{ std::chrono::duration_cast(std::chrono::milliseconds(sentTime)) }; remoteState.lastUpdateServerTime = uptime_timepoint; if (subType == "UPD") { int startFrom = 4; remoteState.position = { std::stof(parts[startFrom]), std::stof(parts[startFrom + 1]), std::stof(parts[startFrom + 2]) }; Eigen::Quaternionf q( std::stof(parts[startFrom + 3]), std::stof(parts[startFrom + 4]), std::stof(parts[startFrom + 5]), std::stof(parts[startFrom + 6])); remoteState.rotation = q.toRotationMatrix(); remoteState.currentAngularVelocity = Eigen::Vector3f{ std::stof(parts[startFrom + 7]), std::stof(parts[startFrom + 8]), std::stof(parts[startFrom + 9]) }; remoteState.velocity = std::stof(parts[startFrom + 10]); remoteState.selectedVelocity = std::stoi(parts[startFrom + 11]); remoteState.discreteMag = std::stof(parts[startFrom + 12]); remoteState.discreteAngle = std::stoi(parts[startFrom + 13]); } else { throw std::runtime_error("Unknown EVENT subtype: " + subType); } { std::lock_guard pLock(playersMutex); auto& rp = remotePlayers[remoteId]; rp.add_state(remoteState); } } } } void WebSocketClient::Send(const std::string& message) { if (!connected) return; auto ss = std::make_shared(message); std::lock_guard lock(writeMutex_); writeQueue_.push(ss); // Если сейчас ничего не записывается, инициируем первую запись if (!isWriting_) { doWrite(); } } void WebSocketClient::doWrite() { // Эта функция всегда вызывается под мьютексом или из колбэка if (writeQueue_.empty()) { isWriting_ = false; return; } isWriting_ = true; auto message = writeQueue_.front(); // Захватываем self (shared_from_this), чтобы объект не удалился во время записи ws_->async_write( boost::asio::buffer(*message), [this, message](boost::beast::error_code ec, std::size_t) { if (ec) { connected = false; return; } std::lock_guard lock(writeMutex_); writeQueue_.pop(); // Удаляем отправленное сообщение doWrite(); // Проверяем следующее } ); } } #endif