class IteratorSourceStage implements Stage {

  @Override
  public void run() {
    while (pipe.isOpen() && source.hasNext()) {
      final T value = source.next();
      if (value != null) {
        pipe.put(value);
      }
    }
    pipe.close();
  }

    IteratorSourceStage(final String name, final Iterator<T> source, final Pipe<T> pipe);

}