import org.apache.cassandra.repair.messages.SyncComplete; //导入依赖的package包/类
public void handleMessage(InetAddress endpoint, RepairMessage message)
{
    RepairJobDesc desc = message.desc;
    RepairSession session = sessions.get(desc.sessionId);
    if (session == null)
        return;
    switch (message.messageType)
    {
        case VALIDATION_COMPLETE:
            ValidationComplete validation = (ValidationComplete) message;
            session.validationComplete(desc, endpoint, validation.tree);
            break;
        case SYNC_COMPLETE:
            // one of replica is synced.
            SyncComplete sync = (SyncComplete) message;
            session.syncComplete(desc, sync.nodes, sync.success);
            break;
        default:
            break;
    }
}
 

import org.apache.cassandra.repair.messages.SyncComplete; //导入依赖的package包/类
/**
 * Compares our trees, and triggers repairs for any ranges that mismatch.
 */
public void run()
{
    // compare trees, and collect differences
    differences.addAll(MerkleTree.difference(r1.tree, r2.tree));

    // choose a repair method based on the significance of the difference
    String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
    if (differences.isEmpty())
    {
        logger.info(String.format(format, "are consistent"));
        // send back sync complete message
        MessagingService.instance().sendOneWay(new SyncComplete(desc, r1.endpoint, r2.endpoint, true).createMessage(), FBUtilities.getLocalAddress());
        return;
    }

    // non-0 difference: perform streaming repair
    logger.info(String.format(format, "have " + differences.size() + " range(s) out of sync"));
    performStreamingRepair();
}
 

import org.apache.cassandra.repair.messages.SyncComplete; //导入依赖的package包/类
/**
 * Compares our trees, and triggers repairs for any ranges that mismatch.
 */
public void run()
{
    // compare trees, and collect differences
    differences.addAll(MerkleTree.difference(r1.tree, r2.tree));

    // choose a repair method based on the significance of the difference
    String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
    if (differences.isEmpty())
    {
        logger.info(String.format(format, "are consistent"));
        // send back sync complete message
        MessagingService.instance().sendOneWay(new SyncComplete(desc, r1.endpoint, r2.endpoint, true).createMessage(), FBUtilities.getLocalAddress());
        return;
    }

    // non-0 difference: perform streaming repair
    logger.info(format, "have {} range(s) out of sync", differences.size());
    performStreamingRepair();
}
 

import org.apache.cassandra.repair.messages.SyncComplete; //导入依赖的package包/类
/**
 * If we succeeded on both stream in and out, reply back to the initiator.
 */
public void onSuccess(StreamState state)
{
    logger.info(String.format("[repair #%s] streaming task succeed, returning response to %s", desc.sessionId, request.initiator));
    MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, true).createMessage(), request.initiator);
}
 

import org.apache.cassandra.repair.messages.SyncComplete; //导入依赖的package包/类
/**
 * If we failed on either stream in or out, reply fail to the initiator.
 */
public void onFailure(Throwable t)
{
    MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, false).createMessage(), request.initiator);
}
 

import org.apache.cassandra.repair.messages.SyncComplete; //导入依赖的package包/类
/**
 * If we succeeded on both stream in and out, reply back to coordinator
 */
public void onSuccess(StreamState state)
{
    logger.info(String.format("[repair #%s] streaming task succeed, returning response to %s", desc.sessionId, request.initiator));
    MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, true).createMessage(), request.initiator);
}
 

import org.apache.cassandra.repair.messages.SyncComplete; //导入依赖的package包/类
/**
 * If we failed on either stream in or out, reply fail to coordinator
 */
public void onFailure(Throwable t)
{
    MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, false).createMessage(), request.initiator);
}