code: public final void streamBufferWithoutGroups(Iterator<IN> iterator, Collector<OUT> c) {
  		SingleElementPushBackIterator<IN> i = new SingleElementPushBackIterator<>(iterator);
  		try {
  			int size;
  			if (i.hasNext()) {
  				while (true) {
  					int sig = in.readInt();
  					switch (sig) {
  						case SIGNAL_BUFFER_REQUEST:
  							if (i.hasNext()) {
  								size = sender.sendBuffer(i);
  								sendWriteNotification(size, i.hasNext());
  							} else {
  								throw new RuntimeException("External process requested data even though none is available.");
  							}
  							break;
  						case SIGNAL_FINISHED:
  							return;
  						case SIGNAL_ERROR:
  							try {
  								outPrinter.join();
  							} catch (InterruptedException e) {
  								outPrinter.interrupt();
  							}
  							try {
  								errorPrinter.join();
  							} catch (InterruptedException e) {
  								errorPrinter.interrupt();
  							}
  							throw new RuntimeException(
  								"External process for task " + function.getRuntimeContext().getTaskName() + " terminated prematurely due to an error." + msg);
  						default:
  							receiver.collectBuffer(c, sig);
  							sendReadConfirmation();
  							break;
  					}
  				}
  			}
  		} catch (SocketTimeoutException ignored) {
  			throw new RuntimeException("External process for task " + function.getRuntimeContext().getTaskName() + " stopped responding." + msg.get());
  		} catch (Exception e) {
  			throw new RuntimeException("Critical failure for task " + function.getRuntimeContext().getTaskName() + ". " + msg.get(), e);
  		}
  	}
docs: Sends all values contained in the iterator to the external process and collects all results.
  
  @param iterator input stream
  @param c        collector