class 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();
  }

  public  ProcessingStage(
      final String name,
      final ReadPipe<I> inputPipe,
      final WritePipe<O> outputPipe,
      final Processor<I, O> processor);

}