class RemoteInputChannel extends InputChannel implements ChannelStateHolder {

	@Override
	Optional<BufferAndAvailability> getNextBuffer() throws IOException {
		checkPartitionRequestQueueInitialized();

		final SequenceBuffer next;
		final DataType nextDataType;

		synchronized (receivedBuffers) {
			next = receivedBuffers.poll();
			nextDataType = receivedBuffers.peek() != null ? receivedBuffers.peek().buffer.getDataType() : DataType.NONE;
		}

		if (next == null) {
			if (isReleased.get()) {
				throw new CancelTaskException("Queried for a buffer after channel has been released.");
			}
			return Optional.empty();
		}

		numBytesIn.inc(next.buffer.getSize());
		numBuffersIn.inc();
		return Optional.of(new BufferAndAvailability(next.buffer, nextDataType, 0, next.sequenceNumber));
	}

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

}