Java 类org.apache.hadoop.fs.BatchedRemoteIterator 实例源码

List<CacheDirectiveEntry> getAllCacheDirectives(UpstreamManager.Upstream upstream) throws IOException {
    CacheDirectiveInfo filter = new CacheDirectiveInfo.Builder().build();
    List<CacheDirectiveEntry> directives = new ArrayList<>();
    long prevId = -1;
    while (true) {
        BatchedRemoteIterator.BatchedEntries<CacheDirectiveEntry> it =
                upstream.protocol.listCacheDirectives(prevId, filter);
        if (it.size() == 0) {
            break;
        }
        for (int i = 0; i < it.size(); i++) {
            CacheDirectiveEntry entry = it.get(i);
            prevId = entry.getInfo().getId();
            directives.add(entry);
        }
    }
    return directives;
}

List<CachePoolEntry> getAllCachePools(UpstreamManager.Upstream upstream) throws IOException {
    String prevPool = "";
    List<CachePoolEntry> pools = new ArrayList<>();

    while (true) {
        BatchedRemoteIterator.BatchedEntries<CachePoolEntry> it = upstream.protocol.listCachePools(prevPool);
        if (it.size() == 0) {
            break;
        }
        for (int i = 0; i < it.size(); i++) {
            CachePoolEntry entry = it.get(i);
            prevPool = entry.getInfo().getPoolName();
            pools.add(entry);
        }
    }
    return pools;
}

public BatchedRemoteIterator.BatchedListEntries<CachePoolEntry> listCachePools(String prevKey) {
    final int NUM_PRE_ALLOCATED_ENTRIES = 16;
    ArrayList<CachePoolEntry> results =
            new ArrayList<CachePoolEntry>(NUM_PRE_ALLOCATED_ENTRIES);
    SortedMap<String, CachePoolEntry> tailMap = cachePools.tailMap(prevKey, false);
    int numListed = 0;
    for (Map.Entry<String, CachePoolEntry> cur : tailMap.entrySet()) {
        if (numListed++ >= maxListCachePoolsResponses) {
            return new BatchedRemoteIterator.BatchedListEntries<>(results, true);
        }
        results.add(cur.getValue());
    }
    return new BatchedRemoteIterator.BatchedListEntries<>(results, false);
}

@Override
public BatchedRemoteIterator.BatchedEntries<CacheDirectiveEntry> 
    listCacheDirectives(long prevId, CacheDirectiveInfo filter) 
    throws IOException {
  try {
    AuthorizationProvider.beginClientOp();
    return server.listCacheDirectives(prevId, filter);
  } finally {
    AuthorizationProvider.endClientOp();
  }
}

@Override
public BatchedRemoteIterator.BatchedEntries<CachePoolEntry> listCachePools(
    String prevPool) throws IOException {
  try {
    AuthorizationProvider.beginClientOp();
    return server.listCachePools(prevPool);
  } finally {
    AuthorizationProvider.endClientOp();
  }
}

@Override
public BatchedRemoteIterator.BatchedEntries<EncryptionZone> 
    listEncryptionZones(long prevId) throws IOException {
  try {
    AuthorizationProvider.beginClientOp();
    return server.listEncryptionZones(prevId);
  } finally {
    AuthorizationProvider.endClientOp();
  }
}

public BatchedRemoteIterator.BatchedListEntries<CacheDirectiveEntry> listCacheDirectives(long prevId,
                                                                                         CacheDirectiveInfo filter) throws InvalidRequestException {
    final int NUM_PRE_ALLOCATED_ENTRIES = 16;
    String filterPath = null;
    if (filter.getPath() != null) {
        filterPath = validatePath(filter);
    }
    if (filter.getReplication() != null) {
        throw new InvalidRequestException(
                "Filtering by replication is unsupported.");
    }

    // Querying for a single ID
    final Long id = filter.getId();
    if (id != null) {
        if (!directivesById.containsKey(id)) {
            throw new InvalidRequestException("Did not find requested id " + id);
        }
        // Since we use a tailMap on directivesById, setting prev to id-1 gets
        // us the directive with the id (if present)
        prevId = id - 1;
    }

    ArrayList<CacheDirectiveEntry> replies =
            new ArrayList<CacheDirectiveEntry>(NUM_PRE_ALLOCATED_ENTRIES);
    int numReplies = 0;
    SortedMap<Long, CacheDirectiveEntry> tailMap =
            directivesById.tailMap(prevId + 1);
    for (Map.Entry<Long, CacheDirectiveEntry> cur : tailMap.entrySet()) {
        if (numReplies >= maxListCacheDirectivesNumResponses) {
            return new BatchedRemoteIterator.BatchedListEntries<>(replies, true);
        }
        CacheDirectiveInfo info = cur.getValue().getInfo();

        // If the requested ID is present, it should be the first item.
        // Hitting this case means the ID is not present, or we're on the second
        // item and should break out.
        if (id != null &&
                !(info.getId().equals(id))) {
            break;
        }
        if (filter.getPool() != null &&
                !info.getPool().equals(filter.getPool())) {
            continue;
        }
        if (filterPath != null &&
                !info.getPath().toUri().getPath().equals(filterPath)) {
            continue;
        }
        replies.add(cur.getValue());
        numReplies++;
    }
    return new BatchedRemoteIterator.BatchedListEntries<>(replies, false);
}