private Future<StreamState> streamRanges(Map<String,Multimap<Range<Token>,InetAddress>> rangesToStreamByKeyspace){
  Map<String,Map<InetAddress,List<Range<Token>>>> sessionsToStreamByKeyspace=new HashMap<>();
  for (  Map.Entry<String,Multimap<Range<Token>,InetAddress>> entry : rangesToStreamByKeyspace.entrySet()) {
    String keyspace=entry.getKey();
    Multimap<Range<Token>,InetAddress> rangesWithEndpoints=entry.getValue();
    if (rangesWithEndpoints.isEmpty())     continue;
    Map<InetAddress,List<Range<Token>>> rangesPerEndpoint=new HashMap<>();
    for (    Map.Entry<Range<Token>,InetAddress> endPointEntry : rangesWithEndpoints.entries()) {
      Range<Token> range=endPointEntry.getKey();
      InetAddress endpoint=endPointEntry.getValue();
      List<Range<Token>> curRanges=rangesPerEndpoint.get(endpoint);
      if (curRanges == null) {
        curRanges=new LinkedList<>();
        rangesPerEndpoint.put(endpoint,curRanges);
      }
      curRanges.add(range);
    }
    sessionsToStreamByKeyspace.put(keyspace,rangesPerEndpoint);
  }
  StreamPlan streamPlan=new StreamPlan("Unbootstrap");
  for (  Map.Entry<String,Map<InetAddress,List<Range<Token>>>> entry : sessionsToStreamByKeyspace.entrySet()) {
    String keyspaceName=entry.getKey();
    Map<InetAddress,List<Range<Token>>> rangesPerEndpoint=entry.getValue();
    for (    Map.Entry<InetAddress,List<Range<Token>>> rangesEntry : rangesPerEndpoint.entrySet()) {
      List<Range<Token>> ranges=rangesEntry.getValue();
      InetAddress newEndpoint=rangesEntry.getKey();
      InetAddress preferred=SystemKeyspace.getPreferredIP(newEndpoint);
      streamPlan.transferRanges(newEndpoint,preferred,keyspaceName,ranges);
    }
  }
  return streamPlan.execute();
}

private Future < StreamState > streamRanges ( Map < String , Multimap < Range < Token > , InetAddress > > rangesToStreamByKeyspace ) { Map < String , Map < InetAddress , List < Range < Token > > > > sessionsToStreamByKeyspace = new HashMap < > ( ) ; for ( Map . Entry < String , Multimap < Range < Token > , InetAddress > > entry : rangesToStreamByKeyspace . entrySet ( ) ) { String keyspace = entry . getKey ( ) ; Multimap < Range < Token > , InetAddress > rangesWithEndpoints = entry . getValue ( ) ; if ( rangesWithEndpoints . isEmpty ( ) ) continue ; Map < InetAddress , List < Range < Token > > > rangesPerEndpoint = new HashMap < > ( ) ; for ( Map . Entry < Range < Token > , InetAddress > endPointEntry : rangesWithEndpoints . entries ( ) ) { Range < Token > range = endPointEntry . getKey ( ) ; InetAddress endpoint = endPointEntry . getValue ( ) ; List < Range < Token > > curRanges = rangesPerEndpoint . get ( endpoint ) ; if ( curRanges == null ) { curRanges = new LinkedList < > ( ) ; rangesPerEndpoint . put ( endpoint , curRanges ) ; } curRanges . add ( range ) ; } sessionsToStreamByKeyspace . put ( keyspace , rangesPerEndpoint ) ; } StreamPlan streamPlan = new StreamPlan ( "Unbootstrap" ) ; for ( Map . Entry < String , Map < InetAddress , List < Range < Token > > > > entry : sessionsToStreamByKeyspace . entrySet ( ) ) { String keyspaceName = entry . getKey ( ) ; Map < InetAddress , List < Range < Token > > > rangesPerEndpoint = entry . getValue ( ) ; for ( Map . Entry < InetAddress , List < Range < Token > > > rangesEntry : rangesPerEndpoint . entrySet ( ) ) { List < Range < Token > > ranges = rangesEntry . getValue ( ) ; InetAddress newEndpoint = rangesEntry . getKey ( ) ; InetAddress preferred = SystemKeyspace . getPreferredIP ( newEndpoint ) ; streamPlan . transferRanges ( newEndpoint , preferred , keyspaceName , ranges ) ; } } return streamPlan . execute ( ) ; }
seed data to the endpoints that will be responsible for it at the future
