class RemoteInputChannel extends InputChannel implements ChannelStateHolder {

	public void onBuffer(Buffer buffer, int sequenceNumber, int backlog) throws IOException {
		boolean recycleBuffer = true;

		try {
			if (expectedSequenceNumber != sequenceNumber) {
				onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber));
				return;
			}

			final boolean wasEmpty;
			boolean firstPriorityEvent = false;
			synchronized (receivedBuffers) {
				// Similar to notifyBufferAvailable(), make sure that we never add a buffer
				// after releaseAllResources() released all buffers from receivedBuffers
				// (see above for details).
				if (isReleased.get()) {
					return;
				}

				wasEmpty = receivedBuffers.isEmpty();

				if (buffer.getDataType().hasPriority()) {
					receivedBuffers.addPriorityElement(new SequenceBuffer(buffer, sequenceNumber));
					if (channelStatePersister.checkForBarrier(buffer)) {
						// checkpoint was not yet started by task thread,
						// so remember the numbers of buffers to spill for the time when it will be started
						numBuffersOvertaken = receivedBuffers.getNumUnprioritizedElements();
					}
					firstPriorityEvent = receivedBuffers.getNumPriorityElements() == 1;
				} else {
					receivedBuffers.add(new SequenceBuffer(buffer, sequenceNumber));
					channelStatePersister.maybePersist(buffer);
				}

				++expectedSequenceNumber;
			}
			recycleBuffer = false;

			if (firstPriorityEvent) {
				notifyPriorityEvent(sequenceNumber);
			}
			if (wasEmpty) {
				notifyChannelNonEmpty();
			}

			if (backlog >= 0) {
				onSenderBacklog(backlog);
			}
		} finally {
			if (recycleBuffer) {
				buffer.recycleBuffer();
			}
		}
	}

	public  RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		int networkBuffersPerChannel,
		Counter numBytesIn,
		Counter numBuffersIn);

}