import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public Iterable<Row> resolve()
{
    ArrayList<RowIterator> iters = new ArrayList<RowIterator>(responses.size());
    int n = 0;
    for (MessageIn<RangeSliceReply> response : responses)
    {
        RangeSliceReply reply = response.payload;
        n = Math.max(n, reply.rows.size());
        iters.add(new RowIterator(reply.rows.iterator(), response.from));
    }
    // for each row, compute the combination of all different versions seen, and repair incomplete versions
    // TODO do we need to call close?
    CloseableIterator<Row> iter = MergeIterator.get(iters, pairComparator, new Reducer());

    List<Row> resolvedRows = new ArrayList<Row>(n);
    while (iter.hasNext())
        resolvedRows.add(iter.next());

    return resolvedRows;
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void doVerb(MessageIn<GossipDigestAck2> message, int id)
{
    if (logger.isTraceEnabled())
    {
        InetAddress from = message.from;
        logger.trace("Received a GossipDigestAck2Message from {}", from);
    }
    if (!Gossiper.instance.isEnabled())
    {
        if (logger.isTraceEnabled())
            logger.trace("Ignoring GossipDigestAck2Message because gossip is disabled");
        return;
    }
    Map<InetAddress, EndpointState> remoteEpStateMap = message.payload.getEndpointStateMap();
    /* Notify the Failure Detector */
    Gossiper.instance.notifyFailureDetector(remoteEpStateMap);
    Gossiper.instance.applyStateLocally(remoteEpStateMap);
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void response(MessageIn message)
{
    String dataCenter = message == null
                        ? DatabaseDescriptor.getLocalDataCenter()
                        : snitch.getDatacenter(message.from);

    responses.get(dataCenter).getAndDecrement();
    acks.incrementAndGet();

    for (AtomicInteger i : responses.values())
    {
        if (i.get() > 0)
            return;
    }

    // all the quorum conditions are met
    signal();
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void doVerb(MessageIn<Truncation> message, int id)
{
    Truncation t = message.payload;
    Tracing.trace("Applying truncation of {}.{}", t.keyspace, t.columnFamily);
    try
    {
        ColumnFamilyStore cfs = Keyspace.open(t.keyspace).getColumnFamilyStore(t.columnFamily);
        cfs.truncateBlocking();
    }
    catch (Exception e)
    {
        logger.error("Error in truncation", e);
        respondError(t, message);

        if (FSError.findNested(e) != null)
            throw FSError.findNested(e);
    }
    Tracing.trace("Enqueuing response to truncate operation to {}", message.from);

    TruncateResponse response = new TruncateResponse(t.keyspace, t.columnFamily, true);
    logger.trace("{} applied.  Enqueuing response to {}@{} ", new Object[]{ t, id, message.from });
    MessagingService.instance().sendReply(response.createMessage(), id, message.from);
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void doVerb(MessageIn<ReadCommand> message, int id)
{
    if (StorageService.instance.isBootstrapMode())
    {
        throw new RuntimeException("Cannot service reads while bootstrapping!");
    }

    ReadCommand command = message.payload;
    Keyspace keyspace = Keyspace.open(command.ksName);
    Row row;
    try
    {
        row = command.getRow(keyspace);
    }
    catch (TombstoneOverwhelmingException e)
    {
        // error already logged.  Drop the request
        return;
    }

    MessageOut<ReadResponse> reply = new MessageOut<ReadResponse>(MessagingService.Verb.REQUEST_RESPONSE,
                                                                  getResponse(command, row),
                                                                  ReadResponse.serializer);
    Tracing.trace("Enqueuing response to {}", message.from);
    MessagingService.instance().sendReply(reply, id, message.from);
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
/**
 * Determines the tracing context from a message.  Does NOT set the threadlocal state.
 * 
 * @param message The internode message
 */
public TraceState initializeFromMessage(final MessageIn<?> message)
{
    final byte[] sessionBytes = message.parameters.get(Tracing.TRACE_HEADER);

    if (sessionBytes == null)
        return null;

    assert sessionBytes.length == 16;
    UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes));
    TraceState ts = sessions.get(sessionId);
    if (ts != null && ts.acquireReference())
        return ts;

    if (message.verb == MessagingService.Verb.REQUEST_RESPONSE)
    {
        // received a message for a session we've already closed out.  see CASSANDRA-5668
        return new ExpiredTraceState(sessionId);
    }
    else
    {
        ts = new TraceState(message.from, sessionId);
        sessions.put(sessionId, ts);
        return ts;
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void doVerb(MessageIn<AbstractRangeCommand> message, int id)
{
    try
    {
        if (StorageService.instance.isBootstrapMode())
        {
            /* Don't service reads! */
            throw new RuntimeException("Cannot service reads while bootstrapping!");
        }
        RangeSliceReply reply = new RangeSliceReply(message.payload.executeLocally());
        Tracing.trace("Enqueuing response to {}", message.from);
        MessagingService.instance().sendReply(reply.createMessage(), id, message.from);
    }
    catch (TombstoneOverwhelmingException e)
    {
        // error already logged.  Drop the request
    }
    catch (Exception ex)
    {
        throw new RuntimeException(ex);
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public boolean preprocess(MessageIn<ReadResponse> message)
{
    MessageIn<ReadResponse> toReplace = null;
    for (MessageIn<ReadResponse> reply : replies)
    {
        if (reply.from.equals(message.from))
        {
            if (!message.payload.isDigestQuery())
                toReplace = reply;
            break;
        }
    }
    // replace old message
    if (toReplace != null)
    {
        replies.remove(toReplace);
        replies.add(message);
        return false;
    }
    replies.add(message);
    return true;
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void doVerb(final MessageIn<CounterMutation> message, final int id)
{
    final CounterMutation cm = message.payload;
    logger.trace("Applying forwarded {}", cm);

    String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());
    // We should not wait for the result of the write in this thread,
    // otherwise we could have a distributed deadlock between replicas
    // running this VerbHandler (see #4578).
    // Instead, we use a callback to send the response. Note that the callback
    // will not be called if the request timeout, but this is ok
    // because the coordinator of the counter mutation will timeout on
    // it's own in that case.
    StorageProxy.applyCounterMutationOnLeader(cm, localDataCenter, new Runnable()
    {
        public void run()
        {
            MessagingService.instance().sendReply(WriteResponse.createMessage(), id, message.from);
        }
    });
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void doVerb(MessageIn<RepairMessage> message, int id)
{
    // TODO add cancel/interrupt message
    RepairJobDesc desc = message.payload.desc;
    switch (message.payload.messageType)
    {
        case VALIDATION_REQUEST:
            ValidationRequest validationRequest = (ValidationRequest) message.payload;
            // trigger read-only compaction
            ColumnFamilyStore store = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily);
            Validator validator = new Validator(desc, message.from, validationRequest.gcBefore);
            CompactionManager.instance.submitValidation(store, validator);
            break;

        case SYNC_REQUEST:
            // forwarded sync request
            SyncRequest request = (SyncRequest) message.payload;
            StreamingRepairTask task = new StreamingRepairTask(desc, request);
            task.run();
            break;

        default:
            ActiveRepairService.instance.handleMessage(message.from, message.payload);
            break;
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
/**
 * Determines the tracing context from a message.  Does NOT set the threadlocal state.
 * 
 * @param message The internode message
 */
public TraceState initializeFromMessage(final MessageIn<?> message)
{
    final byte[] sessionBytes = message.parameters.get(Tracing.TRACE_HEADER);

    if (sessionBytes == null)
        return null;

    assert sessionBytes.length == 16;
    UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes));
    TraceState ts = sessions.get(sessionId);
    if (ts != null)
        return ts;

    if (message.verb == MessagingService.Verb.REQUEST_RESPONSE)
    {
        // received a message for a session we've already closed out.  see CASSANDRA-5668
        return new ExpiredTraceState(sessionId);
    }
    else
    {
        ts = new TraceState(message.from, sessionId);
        sessions.put(sessionId, ts);
        return ts;
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void response(MessageIn<T> message)
{
    String dataCenter = message == null
                        ? DatabaseDescriptor.getLocalDataCenter()
                        : snitch.getDatacenter(message.from);

    responses.get(dataCenter).getAndDecrement();
    acks.incrementAndGet();

    for (AtomicInteger i : responses.values())
    {
        if (i.get() > 0)
            return;
    }

    // all the quorum conditions are met
    signal();
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void response(MessageIn<ReadResponse> message)
{
    repairResolver.preprocess(message);
    if (received.incrementAndGet() == blockfor)
    {
        StageManager.getStage(Stage.READ_REPAIR).execute(new WrappedRunnable()
        {
            protected void runMayThrow() throws DigestMismatchException, IOException
            {
                repairResolver.resolve();
            }
        });
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void response(MessageIn<Boolean> msg)
{
    logger.debug("Propose response {} from {}", msg.payload, msg.from);

    if (msg.payload)
        accepts.incrementAndGet();

    latch.countDown();

    if (isSuccessful() || (failFast && (latch.getCount() + accepts.get() < requiredAccepts)))
    {
        while (latch.getCount() > 0)
            latch.countDown();
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void doVerb(MessageIn<SnapshotCommand> message, int id)
{
    SnapshotCommand command = message.payload;
    if (command.clear_snapshot)
    {
        Keyspace.clearSnapshot(command.snapshot_name, command.keyspace);
    }
    else
        Keyspace.open(command.keyspace).getColumnFamilyStore(command.column_family).snapshot(command.snapshot_name);
    logger.debug("Enqueuing response to snapshot request {} to {}", command.snapshot_name, message.from);
    MessagingService.instance().sendReply(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE), id, message.from);
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
/**
 * Determines the tracing context from a message.  Does NOT set the threadlocal state.
 * 
 * @param message The internode message
 */
public TraceState initializeFromMessage(final MessageIn<?> message)
{
    final byte[] sessionBytes = message.parameters.get(TRACE_HEADER);

    if (sessionBytes == null)
        return null;

    assert sessionBytes.length == 16;
    UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes));
    TraceState ts = sessions.get(sessionId);
    if (ts != null && ts.acquireReference())
        return ts;

    byte[] tmpBytes;
    TraceType traceType = TraceType.QUERY;
    if ((tmpBytes = message.parameters.get(TRACE_TYPE)) != null)
        traceType = TraceType.deserialize(tmpBytes[0]);

    if (message.verb == MessagingService.Verb.REQUEST_RESPONSE)
    {
        // received a message for a session we've already closed out.  see CASSANDRA-5668
        return new ExpiredTraceState(sessionId, traceType);
    }
    else
    {
        ts = new TraceState(message.from, sessionId, traceType);
        sessions.put(sessionId, ts);
        return ts;
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void response(MessageIn<TMessage> message)
{
    resolver.preprocess(message);
    int n = waitingFor(message)
          ? recievedUpdater.incrementAndGet(this)
          : received;
    if (n >= blockfor && resolver.isDataPresent())
    {
        condition.signalAll();
        // kick off a background digest comparison if this is a result that (may have) arrived after
        // the original resolve that get() kicks off as soon as the condition is signaled
        if (blockfor < endpoints.size() && n == endpoints.size())
            StageManager.getStage(Stage.READ_REPAIR).execute(new AsyncRepairRunner());
    }
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
/**
 * @return true if the message counts towards the blockfor threshold
 */
private boolean waitingFor(MessageIn message)
{
    return consistencyLevel.isDatacenterLocal()
         ? DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(message.from))
         : true;
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
public void response(ReadResponse result)
{
    MessageIn<ReadResponse> message = MessageIn.create(FBUtilities.getBroadcastAddress(),
                                                       result,
                                                       Collections.<String, byte[]>emptyMap(),
                                                       MessagingService.Verb.INTERNAL_RESPONSE,
                                                       MessagingService.current_version);
    response(message);
}
 

import org.apache.cassandra.net.MessageIn; //导入依赖的package包/类
/**
 * Special case of resolve() so that CL.ONE reads never throw DigestMismatchException in the foreground
 */
public Row getData()
{
    for (MessageIn<ReadResponse> message : replies)
    {
        ReadResponse result = message.payload;
        if (!result.isDigestQuery())
            return result.row();
    }
    return null;
}