### Question:
MapProcessor implements Processor<I, O> { @Override public void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> outputPipe) { final I value = inputPipe.get(); if (value != null) { outputPipe.put(processor.apply(value)); } } MapProcessor(final Function<I, O> processor); @Override void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> outputPipe);  }### Answer:
@Test public void shouldApplyFunctionToItems() { when(processor.apply("A")).thenReturn("a"); input.put("A"); stage.processNextInput(input, output); assertThat(output.hasMore()).isTrue(); assertThat(output.get()).isEqualTo("a"); verify(processor).apply("A"); }
@Test public void shouldSkipProcessingWhenInputIsClosed() { input.close(); stage.processNextInput(input, output); verifyZeroInteractions(processor); }