### Question:
Pipe implements ReadPipe<T>, WritePipe<T> { @Override public void put(final T value) { while (isOpen()) { try { if (queue.offer(value, 1, TimeUnit.SECONDS)) { inputCounter.inc(); return; } } catch (final InterruptedException e) { LOG.trace("Interrupted while waiting to add to output", e); } } } Pipe(
      final int capacity,
      final Counter inputCounter,
      final Counter outputCounter,
      final Counter abortedItemCounter); @Override boolean isOpen(); @Override boolean isAborted(); @Override boolean hasRemainingCapacity(); @Override void close(); @Override void abort(); @Override boolean hasMore(); @Override T get(); @Override T poll(); @Override void drainTo(final Collection<T> output, final int maxElements); @Override void put(final T value);  }### Answer:
@Test public void shouldIncrementInputCounterWhenItemAddedToPipe() { pipe.put("A"); verify(inputCounter).inc(); pipe.put("B"); verify(inputCounter, times(2)).inc(); }