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 ) ; } }
Sends all values contained in the iterator to the external process and collects all results .