import org.apache.cassandra.streaming.StreamPlan; //导入依赖的package包/类
public RangeStreamer(TokenMetadata metadata,
                     Collection<Token> tokens,
                     InetAddress address,
                     String description,
                     boolean useStrictConsistency,
                     IEndpointSnitch snitch,
                     StreamStateStore stateStore)
{
    this.metadata = metadata;
    this.tokens = tokens;
    this.address = address;
    this.description = description;
    this.streamPlan = new StreamPlan(description, true);
    this.useStrictConsistency = useStrictConsistency;
    this.snitch = snitch;
    this.stateStore = stateStore;
    streamPlan.listeners(this.stateStore);
}
 

import org.apache.cassandra.streaming.StreamPlan; //导入依赖的package包/类
/**
 * Starts sending/receiving our list of differences to/from the remote endpoint: creates a callback
 * that will be called out of band once the streams complete.
 */
protected void startSync(List<Range<Token>> differences)
{
    InetAddress local = FBUtilities.getBroadcastAddress();
    // We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
    InetAddress dst = r2.endpoint.equals(local) ? r1.endpoint : r2.endpoint;
    InetAddress preferred = SystemKeyspace.getPreferredIP(dst);

    String message = String.format("Performing streaming repair of %d ranges with %s", differences.size(), dst);
    logger.info("[repair #{}] {}", desc.sessionId, message);
    boolean isIncremental = false;
    if (desc.parentSessionId != null)
    {
        ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId);
        isIncremental = prs.isIncremental;
    }
    Tracing.traceRepair(message);
    new StreamPlan("Repair", repairedAt, 1, false, isIncremental).listeners(this)
                                        .flushBeforeTransfer(true)
                                        // request ranges from the remote node
                                        .requestRanges(dst, preferred, desc.keyspace, differences, desc.columnFamily)
                                        // send ranges to the remote node
                                        .transferRanges(dst, preferred, desc.keyspace, differences, desc.columnFamily)
                                        .execute();
}
 

import org.apache.cassandra.streaming.StreamPlan; //导入依赖的package包/类
public void run()
{
    InetAddress dest = request.dst;
    InetAddress preferred = SystemKeyspace.getPreferredIP(dest);
    logger.info(String.format("[streaming task #%s] Performing streaming repair of %d ranges with %s", desc.sessionId, request.ranges.size(), request.dst));
    boolean isIncremental = false;
    if (desc.parentSessionId != null)
    {
        ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId);
        isIncremental = prs.isIncremental;
    }
    new StreamPlan("Repair", repairedAt, 1, false, isIncremental).listeners(this)
                                        .flushBeforeTransfer(true)
                                        // request ranges from the remote node
                                        .requestRanges(dest, preferred, desc.keyspace, request.ranges, desc.columnFamily)
                                        // send ranges to the remote node
                                        .transferRanges(dest, preferred, desc.keyspace, request.ranges, desc.columnFamily)
                                        .execute();
}
 

import org.apache.cassandra.streaming.StreamPlan; //导入依赖的package包/类
public RangeStreamer(TokenMetadata metadata, Collection<Token> tokens, InetAddress address, String description)
{
    this.metadata = metadata;
    this.tokens = tokens;
    this.address = address;
    this.description = description;
    this.streamPlan = new StreamPlan(description);
}
 

import org.apache.cassandra.streaming.StreamPlan; //导入依赖的package包/类
private Future<StreamState> streamHints()
{
    // StreamPlan will not fail if there are zero files to transfer, so flush anyway (need to get any in-memory hints, as well)
    ColumnFamilyStore hintsCF = Keyspace.open(Keyspace.SYSTEM_KS).getColumnFamilyStore(SystemKeyspace.HINTS_CF);
    FBUtilities.waitOnFuture(hintsCF.forceFlush());

    // gather all live nodes in the cluster that aren't also leaving
    List<InetAddress> candidates = new ArrayList<>(StorageService.instance.getTokenMetadata().cloneAfterAllLeft().getAllEndpoints());
    candidates.remove(FBUtilities.getBroadcastAddress());
    for (Iterator<InetAddress> iter = candidates.iterator(); iter.hasNext(); )
    {
        InetAddress address = iter.next();
        if (!FailureDetector.instance.isAlive(address))
            iter.remove();
    }

    if (candidates.isEmpty())
    {
        logger.warn("Unable to stream hints since no live endpoints seen");
        return Futures.immediateFuture(null);
    }
    else
    {
        // stream to the closest peer as chosen by the snitch
        DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getBroadcastAddress(), candidates);
        InetAddress hintsDestinationHost = candidates.get(0);
        InetAddress preferred = SystemKeyspace.getPreferredIP(hintsDestinationHost);

        // stream all hints -- range list will be a singleton of "the entire ring"
        Token token = StorageService.getPartitioner().getMinimumToken();
        List<Range<Token>> ranges = Collections.singletonList(new Range<>(token, token));

        return new StreamPlan("Hints").transferRanges(hintsDestinationHost,
                                                      preferred,
                                                                  Keyspace.SYSTEM_KS,
                                                                  ranges,
                                                                  SystemKeyspace.HINTS_CF)
                                                  .execute();
    }
}
 

import org.apache.cassandra.streaming.StreamPlan; //导入依赖的package包/类
public RangeStreamer(TokenMetadata metadata, InetAddress address, String description)
{
    this.metadata = metadata;
    this.address = address;
    this.description = description;
    this.streamPlan = new StreamPlan(description);
}
 

import org.apache.cassandra.streaming.StreamPlan; //导入依赖的package包/类
private void streamLegacyTable(String tablePattern, String legacyVersion, String compactNameSuffix) throws Exception
{
    String table = String.format(tablePattern, legacyVersion, compactNameSuffix);
    SSTableReader sstable = SSTableReader.open(getDescriptor(legacyVersion, table));
    IPartitioner p = sstable.getPartitioner();
    List<Range<Token>> ranges = new ArrayList<>();
    ranges.add(new Range<>(p.getMinimumToken(), p.getToken(ByteBufferUtil.bytes("100"))));
    ranges.add(new Range<>(p.getToken(ByteBufferUtil.bytes("100")), p.getMinimumToken()));
    ArrayList<StreamSession.SSTableStreamingSections> details = new ArrayList<>();
    details.add(new StreamSession.SSTableStreamingSections(sstable.ref(),
                                                           sstable.getPositionsForRanges(ranges),
                                                           sstable.estimatedKeysForRanges(ranges), sstable.getSSTableMetadata().repairedAt));
    new StreamPlan("LegacyStreamingTest").transferFiles(FBUtilities.getBroadcastAddress(), details)
                                         .execute().get();
}