AstyanaxEventReaderDAO.java 文件源码

void readAll(String channel, SlabFilter filter, EventSink sink, ConsistencyLevel consistency) {
    // PeekingIterator is needed so that we can look ahead and see the next slab Id
    PeekingIterator<Column<ByteBuffer>> manifestColumns = Iterators.peekingIterator(executePaginated(
            _keyspace.prepareQuery(ColumnFamilies.MANIFEST, consistency)
                    .getKey(channel)
                    .withColumnRange(new RangeBuilder().setLimit(50).build())
                    .autoPaginate(true)).iterator());

    while (manifestColumns.hasNext()) {
        Column<ByteBuffer> manifestColumn = manifestColumns.next();
        ByteBuffer slabId = manifestColumn.getName();
        ByteBuffer nextSlabId = manifestColumns.hasNext() ? manifestColumns.peek().getName() : null;
        boolean open = manifestColumn.getBooleanValue();
        if (filter != null && !filter.accept(slabId, open, nextSlabId)) {
            continue;
        }
        if (!readSlab(channel, slabId, new SlabCursor(), open, sink)) {
            break;
        }
    }
}