Method parseRequest

src/Coordination/KeeperStateMachine.cpp:409–539

}

std::shared_ptr<KeeperRequestForSession> IKeeperStateMachine::parseRequest(
    nuraft::buffer & data, bool final, ZooKeeperLogSerializationVersion * serialization_version, size_t * request_end_position)
{
    ReadBufferFromNuraftBuffer buffer(data);
    auto request_for_session = std::make_shared<KeeperRequestForSession>();
    readIntBinary(request_for_session->session_id, buffer);

    int32_t length = 0;
    Coordination::read(length, buffer);
    /// Request should not exceed max_request_size (this is verified in KeeperTCPHandler)
    if (length < 0)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Invalid request length: {}", length);

    /// because of backwards compatibility, only 32bit xid could be written
    /// for that reason we serialize XID in 2 parts:
    /// - lower: 32 least significant bits of 64bit XID OR 32bit XID
    /// - upper: 32 most significant bits of 64bit XID
    XidHelper xid_helper{};
    Coordination::read(xid_helper.parts.lower, buffer);

    /// go to end of the buffer and read extra information including second part of XID
    auto buffer_position = buffer.getPosition();
    buffer.seek(length - sizeof(uint32_t), SEEK_CUR);

    if (request_end_position)
        *request_end_position = buffer.getPosition();

    using enum ZooKeeperLogSerializationVersion;
    ZooKeeperLogSerializationVersion version = INITIAL;

    if (!buffer.eof())
    {
        version = WITH_TIME;
        readIntBinary(request_for_session->time, buffer);
    }
    else
        request_for_session->time
            = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();

    if (!buffer.eof())
    {
        version = WITH_ZXID_DIGEST;

        readIntBinary(request_for_session->zxid, buffer);

        chassert(!buffer.eof());

        request_for_session->digest.emplace();
        readIntBinary(request_for_session->digest->version, buffer);
        if (request_for_session->digest->version != KeeperDigestVersion::NO_DIGEST || !buffer.eof())
            readIntBinary(request_for_session->digest->value, buffer);
    }

    if (!buffer.eof())
    {
        version = WITH_XID_64;
        Coordination::read(xid_helper.parts.upper, buffer);
    }