DataBufferInputStream violates InputStream contract

[access-control-rtfm]

Affects: 6.0.2

---

The methods asInputStream(...) of the interface DataBuffer have a default implementations. Every such method returns an instance of the class DataBufferInputStream.

Violation of backward compatibility

The constructor of the DataBufferInputStream marks the current write position of the underlying data buffer as the end position of the input stream, which the created instance represents after. The further written data into the underlying data buffer are not reflected by the input stream instance. An InputStream instance, returned by the DefaultDataBuffer of previous versions (e.g. 5.*.*) had this reflections.

Violation of design by contract of InputStream

The class DataBufferInputStream supports marking. But the method mark takes the argument, which is used as a certain read position instead of read limit, as the method reset is called. This violates the specification of the marking concept of InputStream.

Suggestion how it should be.

final class DataBufferInputStream extends InputStream {

	private final DataBuffer dataBuffer;

	private final boolean releaseOnClose;

	private boolean closed;

	private int markedPosition;

        private int readLimit;


	public DataBufferInputStream(DataBuffer dataBuffer, boolean releaseOnClose) {
		Assert.notNull(dataBuffer, "DataBuffer must not be null");
		this.dataBuffer = dataBuffer;
		this.releaseOnClose = releaseOnClose;
                this.markedPosition = -1;
	}

	@Override
	public int read() throws IOException {
		checkClosed();
		if (available() == 0) {
			return -1;
		}
		return this.dataBuffer.read() & 0xFF;
	}

	@Override
	public int read(byte[] b, int off, int len) throws IOException {
		checkClosed();
		int available = available();
		if (available == 0) {
			return -1;
		}
		len = Math.min(available, len);
		this.dataBuffer.read(b, off, len);
		return len;
	}

	@Override
	public boolean markSupported() {
		return true;
	}

	@Override
	public void mark(int readLimit) {
                if(readLimit < 1) {
                        throw new IllegalArgumentException("readLimit must be greater than 0");
                }
		this.markedPosition = this.dataBuffer.readPosition();
                this.readLimit = readLimit
	}

	@Override
	public int available() {
		return this.dataBuffer.readableByteCount();
	}

	@Override
	public void reset() {
		String errorMessage = null;
                int bytesReadSinceLastMark;
                if(this.markedPosition == -1) {
                         errorMessage = "The input stream was never marked";
                 } else if((bytesReadSinceLastMark = this.dataBuffer.readPosition() - this.markedPosition) > this.readLimit) {
                         errorMessage = "The read limit %d was exceeded since last mark at %d by %d bytes".formatted(this.readLimit, this.markedPosition, bytesReadSinceLastMark);
                 } else if(bytesReadSinceLastMark < 0) {
                         errorMessage = "The read position was set before marked position";
                 }
                 if(errorMessage != null) {
                           throw new IOException(errorMessage);
                 }
                 this.dataBuffer.readPosition(this.markedPosition);
	}

	@Override
	public void close() {
		if (this.closed) {
			return;
		}
		if (this.releaseOnClose) {
			DataBufferUtils.release(this.dataBuffer);
		}
		this.closed = true;
	}

	private void checkClosed() throws IOException {
		if (this.closed) {
			throw new IOException("DataBufferInputStream is closed");
		}
	}

}

[poutsma]

Next time, please feel free to create two separate issues for these two separate problems.

Violation of backward compatibility

The problem we had in 5.x is that the InputStream returned by the NettyDataBuffer had different behavior compared to the one returned by the DefaultDataBuffer. For instance, the stream returned by DefaultDataBuffer did not support marking, while the Netty one did. And the Netty one did not allow reading beyond the initial write position, while the default one did.

In 6.0, we added initial support for Netty 5's Buffer, which is quite restrictive when it comes to sharing buffer data, and adds even more variation. So it was decided make asInputStream return the same stream for all DataBuffer implementations, based on what was possible within the strict boundaries defined by Netty 5. Reading beyond the initial write position of the buffer is impossible in Netty 5, so it is impossible in the DataBufferInputStream. With varying implementations of the contract, the DataBuffer abstraction would be useless for our purposes.

Unfortunately, this meant that we had to break backward compatibility in 6.0, but that is to be expected when upgrading to a new major version.

Violation of design by contract of InputStream

Thank you for spotting this, I will commit a fix.

The wording on InputStream::reset do not require you to throw an IOException, just that it 'can be thrown'. So my fix will just ignore the readLimit parameter in DataBufferInputStream::mark, because all data is in memory regardless and there is no buffering. Setting the initial mark in the constructor to -1, instead of the start position, is also not required, as the same javadoc explicitly says that the start of the "file" can be used if no mark has been set.