Method parseChunkedMessages

src/fl/net/http/stream_transport.cpp.hpp:76–127

}

void HttpStreamTransport::parseChunkedMessages() {
    for (;;) {
        size_t chunkSize = mReader.nextChunkSize();
        if (chunkSize == 0) {
            break;
        }

        u8 stackBuf[512];
        fl::vector<u8> heapBuf;
        fl::span<u8> outSpan;
        if (chunkSize <= sizeof(stackBuf)) {
            outSpan = fl::span<u8>(stackBuf, sizeof(stackBuf));
        } else {
            heapBuf.resize(chunkSize);
            outSpan = heapBuf;
        }

        ChunkedReadResult result = mReader.readChunk(outSpan);
        if (!result.hasData()) {
            break;
        }

        fl::string jsonStr(reinterpret_cast<const char*>(result.mData.data()), result.mData.size()); // ok reinterpret cast
        fl::json json = fl::json::parse(jsonStr.c_str());
        if (json.is_null()) {
            continue;
        }

        // Filter heartbeats
        if (json["method"].as_string() == "rpc.ping") {
            mLastHeartbeatReceived = getCurrentTimeMs();
            continue;
        }

        mLastHeartbeatReceived = getCurrentTimeMs();

        // Try to dispatch to pending calls/streams by id
        if (json.contains("id") && !json["id"].is_null()) {
            fl::string idKey = idToString(json["id"]);

            if (resolveRpc(json, idKey)) {
                continue;  // Consumed by pending call
            }
            if (resolveRpcStream(json, idKey)) {
                continue;  // Consumed by pending stream
            }
        }

        // Not matched - add to incoming queue for readRequest()
        mIncomingQueue.push_back(fl::move(json));
    }
}

bool HttpStreamTransport::resolveRpc(const fl::json& msg, const fl::string& idKey) {
    auto it = mPendingCalls.find(idKey);