import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public static PagingState deserialize(ByteBuffer bytes)
{
    if (bytes == null)
        return null;

    try
    {
        DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(bytes));
        ByteBuffer pk = ByteBufferUtil.readWithShortLength(in);
        ByteBuffer cn = ByteBufferUtil.readWithShortLength(in);
        int remaining = in.readInt();
        return new PagingState(pk, cn, remaining);
    }
    catch (IOException e)
    {
        throw new ProtocolException("Invalid value for the paging state");
    }
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public static PagingState deserialize(ByteBuffer bytes, ProtocolVersion protocolVersion)
{
    if (bytes == null)
        return null;

    try (DataInputBuffer in = new DataInputBuffer(bytes, true))
    {
        ByteBuffer pk;
        RowMark mark;
        int remaining, remainingInPartition;
        if (protocolVersion.isSmallerOrEqualTo(ProtocolVersion.V3))
        {
            pk = ByteBufferUtil.readWithShortLength(in);
            mark = new RowMark(ByteBufferUtil.readWithShortLength(in), protocolVersion);
            remaining = in.readInt();
            // Note that while 'in.available()' is theoretically an estimate of how many bytes are available
            // without blocking, we know that since we're reading a ByteBuffer it will be exactly how many
            // bytes remain to be read. And the reason we want to condition this is for backward compatility
            // as we used to not set this.
            remainingInPartition = in.available() > 0 ? in.readInt() : Integer.MAX_VALUE;
        }
        else
        {
            pk = ByteBufferUtil.readWithVIntLength(in);
            mark = new RowMark(ByteBufferUtil.readWithVIntLength(in), protocolVersion);
            remaining = (int)in.readUnsignedVInt();
            remainingInPartition = (int)in.readUnsignedVInt();
        }
        return new PagingState(pk.hasRemaining() ? pk : null,
                               mark.mark.hasRemaining() ? mark : null,
                               remaining,
                               remainingInPartition);
    }
    catch (IOException e)
    {
        throw new ProtocolException("Invalid value for the paging state");
    }
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public AuthResponse decode(ByteBuf body, int version)
{
    if (version == 1)
        throw new ProtocolException("SASL Authentication is not supported in version 1 of the protocol");

    ByteBuffer b = CBUtil.readValue(body);
    byte[] token = new byte[b.remaining()];
    b.get(token);
    return new AuthResponse(token);
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public CredentialsMessage decode(ByteBuf body, int version)
{
    if (version > 1)
        throw new ProtocolException("Legacy credentials authentication is not supported in " +
                "protocol versions > 1. Please use SASL authentication via a SaslResponse message");

    Map<String, String> credentials = CBUtil.readStringMap(body);
    return new CredentialsMessage(credentials);
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public AuthResponse decode(ChannelBuffer body, int version)
{
    if (version == 1)
        throw new ProtocolException("SASL Authentication is not supported in version 1 of the protocol");

    ByteBuffer b = CBUtil.readValue(body);
    byte[] token = new byte[b.remaining()];
    b.get(token);
    return new AuthResponse(token);
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public CredentialsMessage decode(ChannelBuffer body, int version)
{
    if (version > 1)
        throw new ProtocolException("Legacy credentials authentication is not supported in " +
                "protocol versions > 1. Please use SASL authentication via a SaslResponse message");

    Map<String, String> credentials = CBUtil.readStringMap(body);
    return new CredentialsMessage(credentials);
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public static PagingState deserialize(ByteBuffer bytes, int protocolVersion)
{
    if (bytes == null)
        return null;

    try (DataInputBuffer in = new DataInputBuffer(bytes, true))
    {
        ByteBuffer pk;
        RowMark mark;
        int remaining, remainingInPartition;
        if (protocolVersion <= Server.VERSION_3)
        {
            pk = ByteBufferUtil.readWithShortLength(in);
            mark = new RowMark(ByteBufferUtil.readWithShortLength(in), protocolVersion);
            remaining = in.readInt();
            // Note that while 'in.available()' is theoretically an estimate of how many bytes are available
            // without blocking, we know that since we're reading a ByteBuffer it will be exactly how many
            // bytes remain to be read. And the reason we want to condition this is for backward compatility
            // as we used to not set this.
            remainingInPartition = in.available() > 0 ? in.readInt() : Integer.MAX_VALUE;
        }
        else
        {
            pk = ByteBufferUtil.readWithVIntLength(in);
            mark = new RowMark(ByteBufferUtil.readWithVIntLength(in), protocolVersion);
            remaining = (int)in.readUnsignedVInt();
            remainingInPartition = (int)in.readUnsignedVInt();
        }
        return new PagingState(pk.hasRemaining() ? pk : null,
                               mark.mark.hasRemaining() ? mark : null,
                               remaining,
                               remainingInPartition);
    }
    catch (IOException e)
    {
        throw new ProtocolException("Invalid value for the paging state");
    }
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public QueryOptions decode(ByteBuf body, ProtocolVersion version)
{
    ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
    EnumSet<Flag> flags = Flag.deserialize(version.isGreaterOrEqualTo(ProtocolVersion.V5)
                                           ? (int)body.readUnsignedInt()
                                           : (int)body.readByte());

    List<ByteBuffer> values = Collections.<ByteBuffer>emptyList();
    List<String> names = null;
    if (flags.contains(Flag.VALUES))
    {
        if (flags.contains(Flag.NAMES_FOR_VALUES))
        {
            Pair<List<String>, List<ByteBuffer>> namesAndValues = CBUtil.readNameAndValueList(body, version);
            names = namesAndValues.left;
            values = namesAndValues.right;
        }
        else
        {
            values = CBUtil.readValueList(body, version);
        }
    }

    boolean skipMetadata = flags.contains(Flag.SKIP_METADATA);
    flags.remove(Flag.VALUES);
    flags.remove(Flag.SKIP_METADATA);

    SpecificOptions options = SpecificOptions.DEFAULT;
    if (!flags.isEmpty())
    {
        int pageSize = flags.contains(Flag.PAGE_SIZE) ? body.readInt() : -1;
        PagingState pagingState = flags.contains(Flag.PAGING_STATE) ? PagingState.deserialize(CBUtil.readValue(body), version) : null;
        ConsistencyLevel serialConsistency = flags.contains(Flag.SERIAL_CONSISTENCY) ? CBUtil.readConsistencyLevel(body) : ConsistencyLevel.SERIAL;
        long timestamp = Long.MIN_VALUE;
        if (flags.contains(Flag.TIMESTAMP))
        {
            long ts = body.readLong();
            if (ts == Long.MIN_VALUE)
                throw new ProtocolException(String.format("Out of bound timestamp, must be in [%d, %d] (got %d)", Long.MIN_VALUE + 1, Long.MAX_VALUE, ts));
            timestamp = ts;
        }

        options = new SpecificOptions(pageSize, pagingState, serialConsistency, timestamp);
    }
    DefaultQueryOptions opts = new DefaultQueryOptions(consistency, values, skipMetadata, options, version);
    return names == null ? opts : new OptionsWithNames(opts, names);
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public QueryOptions decode(ByteBuf body, int version)
{
    ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
    EnumSet<Flag> flags = Flag.deserialize((int)body.readByte());

    List<ByteBuffer> values = Collections.<ByteBuffer>emptyList();
    List<String> names = null;
    if (flags.contains(Flag.VALUES))
    {
        if (flags.contains(Flag.NAMES_FOR_VALUES))
        {
            Pair<List<String>, List<ByteBuffer>> namesAndValues = CBUtil.readNameAndValueList(body, version);
            names = namesAndValues.left;
            values = namesAndValues.right;
        }
        else
        {
            values = CBUtil.readValueList(body, version);
        }
    }

    boolean skipMetadata = flags.contains(Flag.SKIP_METADATA);
    flags.remove(Flag.VALUES);
    flags.remove(Flag.SKIP_METADATA);

    SpecificOptions options = SpecificOptions.DEFAULT;
    if (!flags.isEmpty())
    {
        int pageSize = flags.contains(Flag.PAGE_SIZE) ? body.readInt() : -1;
        PagingState pagingState = flags.contains(Flag.PAGING_STATE) ? PagingState.deserialize(CBUtil.readValue(body), version) : null;
        ConsistencyLevel serialConsistency = flags.contains(Flag.SERIAL_CONSISTENCY) ? CBUtil.readConsistencyLevel(body) : ConsistencyLevel.SERIAL;
        long timestamp = Long.MIN_VALUE;
        if (flags.contains(Flag.TIMESTAMP))
        {
            long ts = body.readLong();
            if (ts == Long.MIN_VALUE)
                throw new ProtocolException(String.format("Out of bound timestamp, must be in [%d, %d] (got %d)", Long.MIN_VALUE + 1, Long.MAX_VALUE, ts));
            timestamp = ts;
        }

        options = new SpecificOptions(pageSize, pagingState, serialConsistency, timestamp);
    }
    DefaultQueryOptions opts = new DefaultQueryOptions(consistency, values, skipMetadata, options, version);
    return names == null ? opts : new OptionsWithNames(opts, names);
}
 

import org.apache.cassandra.transport.ProtocolException; //导入依赖的package包/类
public Message.Response execute(QueryState state)
{
    try
    {
        if (options.getPageSize() == 0)
            throw new ProtocolException("The page size cannot be 0");

        UUID tracingId = null;
        if (isTracingRequested())
        {
            tracingId = UUIDGen.getTimeUUID();
            state.prepareTracingSession(tracingId);
        }

        if (state.traceNextQuery())
        {
            state.createTracingSession();

            ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
            builder.put("query", query);
            if (options.getPageSize() > 0)
                builder.put("page_size", Integer.toString(options.getPageSize()));
            if(options.getConsistency() != null)
                builder.put("consistency_level", options.getConsistency().name());
            if(options.getSerialConsistency() != null)
                builder.put("serial_consistency_level", options.getSerialConsistency().name());

            Tracing.instance.begin("Execute CQL3 query", state.getClientAddress(), builder.build());
        }

        Message.Response response = ClientState.getCQLQueryHandler().process(query, state, options, getCustomPayload());
        if (options.skipMetadata() && response instanceof ResultMessage.Rows)
            ((ResultMessage.Rows)response).result.metadata.setSkipMetadata();

        if (tracingId != null)
            response.setTracingId(tracingId);

        return response;
    }
    catch (Exception e)
    {
        JVMStabilityInspector.inspectThrowable(e);
        if (!((e instanceof RequestValidationException) || (e instanceof RequestExecutionException)))
            logger.error("Unexpected error during query", e);
        return ErrorMessage.fromException(e);
    }
    finally
    {
        Tracing.instance.stopSession();
    }
}