### Question:
BatchingReadPipe implements ReadPipe<List<T>> { @Override public List<T> poll() { final List<T> batch = new ArrayList<>(); input.drainTo(batch, maximumBatchSize); if (batch.isEmpty()) { return null; } 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 shouldPollForNextBatchWhenAvailable() { source.put("a"); assertThat(batchingPipe.poll()).containsExactly("a"); }
@Test public void shouldReturnNullFromPollWhenNoItemsAvailable() { assertThat(batchingPipe.poll()).isNull(); }
@Test public void shouldCountBatchesReturnedFromPoll() { assertThat(batchingPipe.poll()).isNull(); verifyZeroInteractions(batchCounter); source.put("a"); batchingPipe.poll(); verify(batchCounter, times(1)).inc(); }