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

}