### Question:
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); final Thread stageThread = Thread.currentThread(); inProgress.add(future); updateNextOutputAvailableFuture(); future.whenComplete((result, error) -> stageThread.interrupt()); } outputCompletedTasks(outputPipe); } else { outputNextCompletedTask(outputPipe); } } AsyncOperationProcessor(
      final Function<I, CompletableFuture<O>> processor,
      final int maxConcurrency,
      final boolean preserveOrder); @Override void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> outputPipe); @Override boolean attemptFinalization(final WritePipe<O> outputPipe); @Override void abort();  }### Answer:
@Test public void shouldImmediatelyOutputTasksThatAreAlreadyCompleteEvenIfOutputPipeIsFull() { final AsyncOperationProcessor<CompletableFuture<String>, String> processor = createProcessor(false); when(writePipe.hasRemainingCapacity()).thenReturn(false); when(readPipe.get()).thenReturn(completedFuture("a")); processor.processNextInput(readPipe, writePipe); verify(writePipe).put("a"); }
@Test public void shouldNotExceedConcurrentJobLimit() { final AsyncOperationProcessor<CompletableFuture<String>, String> processor = createProcessor(false); final CompletableFuture<String> task1 = new CompletableFuture<>(); final CompletableFuture<String> task2 = new CompletableFuture<>(); final CompletableFuture<String> task3 = new CompletableFuture<>(); final CompletableFuture<String> task4 = new CompletableFuture<>(); when(readPipe.get()).thenReturn(task1).thenReturn(task2).thenReturn(task3).thenReturn(task4); processor.processNextInput(readPipe, writePipe); processor.processNextInput(readPipe, writePipe); processor.processNextInput(readPipe, writePipe); verify(readPipe, times(3)).get(); processor.processNextInput(readPipe, writePipe); verify(readPipe, times(3)).get(); task1.complete("a"); processor.processNextInput(readPipe, writePipe); verify(writePipe).put("a"); processor.processNextInput(readPipe, writePipe); verify(readPipe, times(4)).get(); }
@Test public void shouldInterruptThreadWhenFutureCompletes() { final AsyncOperationProcessor<CompletableFuture<String>, String> processor = createProcessor(false); final CompletableFuture<String> task1 = new CompletableFuture<>(); when(readPipe.get()).thenReturn(task1); processor.processNextInput(readPipe, writePipe); task1.complete("a"); assertThat(Thread.currentThread().isInterrupted()).isTrue(); }
@Test public void shouldThrowExceptionWhenFutureCompletesExceptionally() { final AsyncOperationProcessor<CompletableFuture<String>, String> processor = createProcessor(false); final CompletableFuture<String> task1 = new CompletableFuture<>(); when(readPipe.get()).thenReturn(task1); processor.processNextInput(readPipe, writePipe); final Exception exception = new IndexOutOfBoundsException("Oh dear"); task1.completeExceptionally(exception); assertThatThrownBy(() -> processor.processNextInput(readPipe, writePipe)) .hasRootCause(exception); }