class AsyncOperationProcessor implements Processor<I, O> {

  @Override
  public void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> outputPipe) {
    if (inProgress.size() < maxConcurrency) {
      final I value = inputPipe.get();
      if (value != null) {
        final CompletableFuture<O> future = processor.apply(value);
        // When the future completes, interrupt so if we're waiting for new input we wake up and
        // schedule the output.
        final Thread stageThread = Thread.currentThread();
        inProgress.add(future);
        updateNextOutputAvailableFuture();
        future.whenComplete((result, error) -> stageThread.interrupt());
      }
      outputCompletedTasks(outputPipe);
    } else {
      outputNextCompletedTask(outputPipe);
    }
  }

  public  AsyncOperationProcessor(
      final Function<I, CompletableFuture<O>> processor,
      final int maxConcurrency,
      final boolean preserveOrder);

}