shadow-over-bishkek001/src/network/WebSocketClient.cpp

#include "WebSocketClient.h"
#include <iostream>
#include <SDL2/SDL.h>
#include "RemotePlayer.h"

// Âñïîìîãàòåëüíûé split
std::vector<std::string> split(const std::string& s, char delimiter) {
    std::vector<std::string> 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<boost::beast::websocket::stream<boost::beast::tcp_stream>>(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<std::mutex> lock(queueMutex);
        messageQueue.push(msg);
    }

    void WebSocketClient::Poll() {
        std::lock_guard<std::mutex> 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<std::chrono::milliseconds>(
                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::system_clock::time_point::duration>(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<std::mutex> 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<std::string>(message);

        std::lock_guard<std::mutex> 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<std::mutex> lock(writeMutex_);
                writeQueue_.pop(); // Óäàëÿåì îòïðàâëåííîå ñîîáùåíèå
                doWrite();         // Ïðîâåðÿåì ñëåäóþùåå
            }
        );
    }
}