### Question:
BatchingReadPipe implements ReadPipe<List<T>> { @Override public List<T> get() { final T firstItem = input.get(); if (firstItem == null) { return null; } final List<T> batch = new ArrayList<>(); batch.add(firstItem); input.drainTo(batch, maximumBatchSize - 1); batchCounter.inc(); return batch; } BatchingReadPipe(
      final ReadPipe<T> input, final int maximumBatchSize, final Counter batchCounter); @Override boolean hasMore(); @Override boolean isAborted(); @Override List<T> get(); @Override List<T> poll(); @Override void drainTo(final Collection<List<T>> output, final int maxElements);  }### Answer:
@Test public void shouldGetABatchOfAvailableItems() { source.put("a"); source.put("b"); source.put("c"); source.put("d"); assertThat(batchingPipe.get()).containsExactly("a", "b", "c"); }
@Test public void shouldNotWaitToFillBatch() { source.put("a"); assertThat(batchingPipe.get()).containsExactly("a"); }
@Test public void shouldCountBatchesReturnedFromGet() { source.put("a"); source.close(); batchingPipe.get(); assertThat(batchingPipe.get()).isNull(); verify(batchCounter, times(1)).inc(); }