import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
public synchronized void finishParentSession(UUID parentSession, Set<InetAddress> neighbors, boolean doAntiCompaction) throws InterruptedException, ExecutionException, IOException
{
    try
    {
        if (doAntiCompaction)
        {
            for (InetAddress neighbor : neighbors)
            {
                AnticompactionRequest acr = new AnticompactionRequest(parentSession);
                MessageOut<RepairMessage> req = acr.createMessage();
                MessagingService.instance().sendOneWay(req, neighbor);
            }
            List<Future<?>> futures = doAntiCompaction(parentSession);
            FBUtilities.waitOnFutures(futures);
        }
    }
    finally
    {
        parentRepairSessions.remove(parentSession);
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
/**
 * Sends a notification to a node indicating we have finished replicating data.
 *
 * @param remote node to send notification to
 */
private void sendReplicationNotification(InetAddress remote)
{
    // notify the remote token
    MessageOut msg = new MessageOut(MessagingService.Verb.REPLICATION_FINISHED);
    IFailureDetector failureDetector = FailureDetector.instance;
    if (logger.isDebugEnabled())
        logger.debug("Notifying {} of replication completion\n", remote);
    while (failureDetector.isAlive(remote))
    {
        AsyncOneResponse iar = MessagingService.instance().sendRR(msg, remote);
        try
        {
            iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
            return; // done
        }
        catch(TimeoutException e)
        {
            // try again
        }
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
private void makeRequests(ReadCommand readCommand, Iterable<InetAddress> endpoints)
{
    MessageOut<ReadCommand> message = null;
    boolean hasLocalEndpoint = false;

    for (InetAddress endpoint : endpoints)
    {
        if (isLocalRequest(endpoint))
        {
            hasLocalEndpoint = true;
            continue;
        }

        logger.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint);
        if (message == null)
            message = readCommand.createMessage();
        MessagingService.instance().sendRR(message, endpoint, handler);
    }

    // We delay the local (potentially blocking) read till the end to avoid stalling remote requests.
    if (hasLocalEndpoint)
    {
        logger.trace("reading {} locally", readCommand.isDigestQuery() ? "digest" : "data");
        StageManager.getStage(Stage.READ).maybeExecuteImmediately(new LocalReadRunnable(command, handler));
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
/**
 * Returns true if the chosen target was also a seed. False otherwise
 *
 * @param message
 * @param epSet   a set of endpoint from which a random endpoint is chosen.
 * @return true if the chosen endpoint is also a seed.
 */
private boolean sendGossip(MessageOut<GossipDigestSyn> message, Set<InetAddress> epSet)
{
    List<InetAddress> liveEndpoints = ImmutableList.copyOf(epSet);
    
    int size = liveEndpoints.size();
    if (size < 1)
        return false;
    /* Generate a random number from 0 -> size */
    int index = (size == 1) ? 0 : random.nextInt(size);
    InetAddress to = liveEndpoints.get(index);
    if (logger.isTraceEnabled())
        logger.trace("Sending a GossipDigestSyn to {} ...", to);
    MessagingService.instance().sendOneWay(message, to);
    return seeds.contains(to);
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
private void doGossipToSeed(MessageOut<GossipDigestSyn> prod)
{
    int size = seeds.size();
    if (size > 0)
    {
        if (size == 1 && seeds.contains(FBUtilities.getBroadcastAddress()))
        {
            return;
        }

        if (liveEndpoints.size() == 0)
        {
            sendGossip(prod, seeds);
        }
        else
        {
            /* Gossip with the seed with some probability. */
            double probability = seeds.size() / (double) (liveEndpoints.size() + unreachableEndpoints.size());
            double randDbl = random.nextDouble();
            if (randDbl <= probability)
                sendGossip(prod, seeds);
        }
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的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.MessageOut; //导入依赖的package包/类
/**
 * Sends a notification to a node indicating we have finished replicating data.
 *
 * @param remote node to send notification to
 */
private void sendReplicationNotification(InetAddress remote)
{
    // notify the remote token
    MessageOut msg = new MessageOut(MessagingService.Verb.REPLICATION_FINISHED);
    IFailureDetector failureDetector = FailureDetector.instance;
    if (logger.isDebugEnabled())
        logger.debug("Notifying " + remote.toString() + " of replication completion\n");
    while (failureDetector.isAlive(remote))
    {
        AsyncOneResponse iar = MessagingService.instance().sendRR(msg, remote);
        try
        {
            iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
            return; // done
        }
        catch(TimeoutException e)
        {
            // try again
        }
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
protected void makeDigestRequests(Iterable<InetAddress> endpoints)
{
    ReadCommand digestCommand = command.copy();
    digestCommand.setDigestQuery(true);
    MessageOut<?> message = digestCommand.createMessage();
    for (InetAddress endpoint : endpoints)
    {
        if (isLocalRequest(endpoint))
        {
            logger.trace("reading digest locally");
            StageManager.getStage(Stage.READ).execute(new LocalReadRunnable(digestCommand, handler));
        }
        else
        {
            logger.trace("reading digest from {}", endpoint);
            MessagingService.instance().sendRR(message, endpoint, handler);
        }
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
private void makeRequests(ReadCommand readCommand, Iterable<InetAddress> endpoints)
{
    boolean hasLocalEndpoint = false;

    for (InetAddress endpoint : endpoints)
    {
        if (StorageProxy.canDoLocalRequest(endpoint))
        {
            hasLocalEndpoint = true;
            continue;
        }

        if (traceState != null)
            traceState.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint);
        logger.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint);
        MessageOut<ReadCommand> message = readCommand.createMessage(MessagingService.instance().getVersion(endpoint));
        MessagingService.instance().sendRRWithFailure(message, endpoint, handler);
    }

    // We delay the local (potentially blocking) read till the end to avoid stalling remote requests.
    if (hasLocalEndpoint)
    {
        logger.trace("reading {} locally", readCommand.isDigestQuery() ? "digest" : "data");
        StageManager.getStage(Stage.READ).maybeExecuteImmediately(new LocalReadRunnable(command, handler));
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
private void maybeGossipToSeed(MessageOut<GossipDigestSyn> prod)
{
    int size = seeds.size();
    if (size > 0)
    {
        if (size == 1 && seeds.contains(FBUtilities.getBroadcastAddress()))
        {
            return;
        }

        if (liveEndpoints.size() == 0)
        {
            sendGossip(prod, seeds);
        }
        else
        {
            /* Gossip with the seed with some probability. */
            double probability = seeds.size() / (double) (liveEndpoints.size() + unreachableEndpoints.size());
            double randDbl = random.nextDouble();
            if (randDbl <= probability)
                sendGossip(prod, seeds);
        }
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
public void stop()
{
    EndpointState mystate = endpointStateMap.get(FBUtilities.getBroadcastAddress());
    if (mystate != null && !isSilentShutdownState(mystate))
    {
        logger.info("Announcing shutdown");
        addLocalApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.shutdown(true));
        MessageOut message = new MessageOut(MessagingService.Verb.GOSSIP_SHUTDOWN);
        for (InetAddress ep : liveEndpoints)
            MessagingService.instance().sendOneWay(message, ep);
        Uninterruptibles.sleepUninterruptibly(Integer.getInteger("cassandra.shutdown_announce_in_ms", 2000), TimeUnit.MILLISECONDS);
    }
    else
        logger.warn("No local state or state is in silent shutdown, not announcing shutdown");
    if (scheduledGossipTask != null)
        scheduledGossipTask.cancel(false);
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
public synchronized void enqueue(MessageOut<?> message, int id) throws IOException {
    long timestamp = System.currentTimeMillis();
    out.writeInt(MessagingService.PROTOCOL_MAGIC);

    if (targetVersion < MessagingService.VERSION_20)
        out.writeUTF(String.valueOf(id));
    else
        out.writeInt(id);
    out.writeInt((int) timestamp);
    message.serialize(out, targetVersion);
    out.flush();
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的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.MessageOut; //导入依赖的package包/类
private static void pushSchemaMutation(InetAddress endpoint, Collection<Mutation> schema)
{
    MessageOut<Collection<Mutation>> msg = new MessageOut<>(MessagingService.Verb.DEFINITIONS_UPDATE,
                                                            schema,
                                                            MigrationsSerializer.instance);
    MessagingService.instance().sendOneWay(msg, endpoint);
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
public static MessageOut processOutboundMessage(MessageOut message, int id, InetAddress to)
{
    if (messageSinks.isEmpty())
        return message;

    for (IMessageSink ms : messageSinks)
    {
        message = ms.handleMessage(message, id, to);
        if (message == null)
            return null;
    }
    return message;
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
private boolean doGossipToLiveMember(MessageOut<GossipDigestSyn> message)
{
    int size = liveEndpoints.size();
    if (size == 0)
        return false;
    return sendGossip(message, liveEndpoints);
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
private void doGossipToUnreachableMember(MessageOut<GossipDigestSyn> message)
{
    double liveEndpointCount = liveEndpoints.size();
    double unreachableEndpointCount = unreachableEndpoints.size();
    if (unreachableEndpointCount > 0)
    {
        /* based on some probability */
        double prob = unreachableEndpointCount / (liveEndpointCount + 1);
        double randDbl = random.nextDouble();
        if (randDbl < prob)
            sendGossip(message, unreachableEndpoints.keySet());
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
private void markAlive(final InetAddress addr, final EndpointState localState)
{
    if (MessagingService.instance().getVersion(addr) < MessagingService.VERSION_20)
    {
        realMarkAlive(addr, localState);
        return;
    }

    localState.markDead();

    MessageOut<EchoMessage> echoMessage = new MessageOut<EchoMessage>(MessagingService.Verb.ECHO, new EchoMessage(), EchoMessage.serializer);
    logger.trace("Sending a EchoMessage to {}", addr);
    IAsyncCallback echoHandler = new IAsyncCallback()
    {
        public boolean isLatencyForSnitch()
        {
            return false;
        }

        public void response(MessageIn msg)
        {
            realMarkAlive(addr, localState);
        }
    };

    MessagingService.instance().sendRR(echoMessage, addr, echoHandler);
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
/**
 *  Do a single 'shadow' round of gossip, where we do not modify any state
 *  Only used when replacing a node, to get and assume its states
 */
public void doShadowRound()
{
    buildSeedsList();
    // send a completely empty syn
    List<GossipDigest> gDigests = new ArrayList<GossipDigest>();
    GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(),
            DatabaseDescriptor.getPartitionerName(),
            gDigests);
    MessageOut<GossipDigestSyn> message = new MessageOut<GossipDigestSyn>(MessagingService.Verb.GOSSIP_DIGEST_SYN,
            digestSynMessage,
            GossipDigestSyn.serializer);
    inShadowRound = true;
    for (InetAddress seed : seeds)
        MessagingService.instance().sendOneWay(message, seed);
    int slept = 0;
    try
    {
        while (true)
        {
            Thread.sleep(1000);
            if (!inShadowRound)
                break;
            slept += 1000;
            if (slept > StorageService.RING_DELAY)
                throw new RuntimeException("Unable to gossip with any seeds");
        }
    }
    catch (InterruptedException wtf)
    {
        throw new RuntimeException(wtf);
    }
}
 

import org.apache.cassandra.net.MessageOut; //导入依赖的package包/类
@Test
public void testRemoveHostId() throws InterruptedException
{
    // start removal in background and send replication confirmations
    final AtomicBoolean success = new AtomicBoolean(false);
    Thread remover = new Thread()
    {
        public void run()
        {
            try
            {
                ss.removeNode(removalId.toString());
            }
            catch (Exception e)
            {
                System.err.println(e);
                e.printStackTrace();
                return;
            }
            success.set(true);
        }
    };
    remover.start();

    Thread.sleep(1000); // make sure removal is waiting for confirmation

    assertTrue(tmd.isLeaving(removalhost));
    assertEquals(1, tmd.getLeavingEndpoints().size());

    for (InetAddress host : hosts)
    {
        MessageOut msg = new MessageOut(host, MessagingService.Verb.REPLICATION_FINISHED, null, null, Collections.<String, byte[]>emptyMap());
        MessagingService.instance().sendRR(msg, FBUtilities.getBroadcastAddress());
    }

    remover.join();

    assertTrue(success.get());
    assertTrue(tmd.getLeavingEndpoints().isEmpty());
}