class BatchingReadPipe implements ReadPipe<List<T>> {

  @Override
  public List<T> get() {
    final T firstItem = input.get();
    if (firstItem == null) {
      // Contract of get is to explicitly return null when no more items are available.
      // An empty list is not a suitable thing to return here.
      return null;
    }
    final List<T> batch = new ArrayList<>();
    batch.add(firstItem);
    input.drainTo(batch, maximumBatchSize - 1);
    batchCounter.inc();
    return batch;
  }

  public  BatchingReadPipe(
      final ReadPipe<T> input, final int maximumBatchSize, final Counter batchCounter);

}