### Question:
CompleterStage implements Stage { @Override public void run() { while (input.hasMore()) { final T value = input.get(); if (value != null) { completer.accept(value); } } future.complete(null); } CompleterStage(final String name, final ReadPipe<T> input, final Consumer<T> completer); @Override void run(); CompletableFuture<?> getFuture(); @Override String getName();  }### Answer:
@Test public void shouldAddItemsToOutputUntilPipeHasNoMore() { pipe.put("a"); pipe.put("b"); pipe.put("c"); pipe.close(); stage.run(); assertThat(output).containsExactly("a", "b", "c"); }