@Override
	public BufferOrEvent getNextNonBlocked() throws Exception {
		while (true) {
			// process buffered BufferOrEvents before grabbing new ones
			Optional<BufferOrEvent> next;
			if (currentBuffered == null) {
				next = inputGate.getNextBufferOrEvent();
			}
			else {
				next = Optional.ofNullable(currentBuffered.getNext());
				if (!next.isPresent()) {
					completeBufferedSequence();
					return getNextNonBlocked();
				}
			}

			if (!next.isPresent()) {
				if (!endOfStream) {
					// end of input stream. stream continues with the buffered data
					endOfStream = true;
					releaseBlocksAndResetBarriers();
					return getNextNonBlocked();
				}
				else {
					// final end of both input and buffered data
					return null;
				}
			}

			BufferOrEvent bufferOrEvent = next.get();
			if (isBlocked(bufferOrEvent.getChannelIndex())) {
				// if the channel is blocked, we just store the BufferOrEvent
				bufferBlocker.add(bufferOrEvent);
				checkSizeLimit();
			}
			else if (bufferOrEvent.isBuffer()) {
				return bufferOrEvent;
			}
			else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {
				if (!endOfStream) {
					// process barriers only if there is a chance of the checkpoint completing
					processBarrier((CheckpointBarrier) bufferOrEvent.getEvent(), bufferOrEvent.getChannelIndex());
				}
			}
			else if (bufferOrEvent.getEvent().getClass() == CancelCheckpointMarker.class) {
				processCancellationBarrier((CancelCheckpointMarker) bufferOrEvent.getEvent());
			}
			else {
				if (bufferOrEvent.getEvent().getClass() == EndOfPartitionEvent.class) {
					processEndOfPartition();
				}
				return bufferOrEvent;
			}
		}
	}
@ Override public BufferOrEvent getNextNonBlocked ( ) throws Exception { while ( true ) { Optional < BufferOrEvent > next ; if ( currentBuffered == null ) { next = inputGate . getNextBufferOrEvent ( ) ; } else { next = Optional . ofNullable ( currentBuffered . getNext ( ) ) ; if ( ! next . isPresent ( ) ) { completeBufferedSequence ( ) ; return getNextNonBlocked ( ) ; } } if ( ! next . isPresent ( ) ) { if ( ! endOfStream ) { endOfStream = true ; releaseBlocksAndResetBarriers ( ) ; return getNextNonBlocked ( ) ; } else { return null ; } } BufferOrEvent bufferOrEvent = next . get ( ) ; if ( isBlocked ( bufferOrEvent . getChannelIndex ( ) ) ) { bufferBlocker . add ( bufferOrEvent ) ; checkSizeLimit ( ) ; } else if ( bufferOrEvent . isBuffer ( ) ) { return bufferOrEvent ; } else if ( bufferOrEvent . getEvent ( ) . getClass ( ) == CheckpointBarrier . class ) { if ( ! endOfStream ) { processBarrier ( ( CheckpointBarrier ) bufferOrEvent . getEvent ( ) , bufferOrEvent . getChannelIndex ( ) ) ; } } else if ( bufferOrEvent . getEvent ( ) . getClass ( ) == CancelCheckpointMarker . class ) { processCancellationBarrier ( ( CancelCheckpointMarker ) bufferOrEvent . getEvent ( ) ) ; } else { if ( bufferOrEvent . getEvent ( ) . getClass ( ) == EndOfPartitionEvent . class ) { processEndOfPartition ( ) ; } return bufferOrEvent ; } } }
------------------------------------------------------------------------