### Question:
ProcessingStage implements Stage { @Override public void run() { while (inputPipe.hasMore()) { processor.processNextInput(inputPipe, outputPipe); } if (inputPipe.isAborted()) { processor.abort(); } while (!processor.attemptFinalization(outputPipe)) { if (inputPipe.isAborted()) { processor.abort(); break; } } outputPipe.close(); } ProcessingStage(
      final String name,
      final ReadPipe<I> inputPipe,
      final WritePipe<O> outputPipe,
      final Processor<I, O> processor); @Override void run(); @Override String getName();  }### Answer:
@Test public void shouldCallSingleStepStageForEachInput() { when(singleStep.attemptFinalization(outputPipe)).thenReturn(true); inputPipe.put("A"); inputPipe.put("B"); inputPipe.put("C"); inputPipe.close(); stage.run(); assertThat(outputPipe.poll()).isEqualTo("a"); assertThat(outputPipe.poll()).isEqualTo("b"); assertThat(outputPipe.poll()).isEqualTo("c"); assertThat(outputPipe.poll()).isNull(); verify(singleStep, times(3)).processNextInput(inputPipe, outputPipe); }
@Test public void shouldFinalizeSingleStepStageAndCloseOutputPipeWhenInputCloses() { when(singleStep.attemptFinalization(outputPipe)).thenReturn(true); inputPipe.close(); stage.run(); verify(singleStep).attemptFinalization(outputPipe); verifyNoMoreInteractions(singleStep); assertThat(outputPipe.isOpen()).isFalse(); }
@Test public void shouldAbortIfPipeIsCancelledWhileAttemptingToFinalise() { when(singleStep.attemptFinalization(outputPipe)) .thenAnswer( invocation -> { inputPipe.abort(); return false; }); inputPipe.close(); stage.run(); verify(singleStep).attemptFinalization(outputPipe); verify(singleStep).abort(); verifyNoMoreInteractions(singleStep); assertThat(outputPipe.isOpen()).isFalse(); }
@Test public void shouldAbortProcessorIfReadPipeIsAborted() { when(singleStep.attemptFinalization(outputPipe)).thenReturn(true); inputPipe.abort(); stage.run(); verify(singleStep).abort(); verify(singleStep).attemptFinalization(outputPipe); assertThat(outputPipe.isOpen()).isFalse(); }