### Question:
FlatMapProcessor implements Processor<I, O> { @Override public void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> outputPipe) { final I value = inputPipe.get(); if (value != null) { mapper.apply(value).forEach(outputPipe::put); } } FlatMapProcessor(final Function<I, Stream<O>> mapper); @Override void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> outputPipe);  }### Answer:
@Test public void shouldOutputItemsFromReturnedStreamIndividually() { when(mapper.apply("A")).thenReturn(Stream.of("a", "b", "c")); input.put("A"); stage.processNextInput(input, output); assertThat(output.poll()).isEqualTo("a"); assertThat(output.poll()).isEqualTo("b"); assertThat(output.poll()).isEqualTo("c"); assertThat(output.poll()).isNull(); verify(mapper).apply("A"); }
@Test public void shouldSkipProcessingWhenInputIsClosed() { input.close(); stage.processNextInput(input, output); verifyZeroInteractions(mapper); }