ResizingPipedInputStream.java
package com.amazon.ion.impl;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
/**
* Manages a resizing buffer for production and consumption of data within a <strong>single thread</strong>.
* Buffered bytes may be consumed through the InputStream interface. This provides a few benefits over using
* a PipedOutputStream/PipedInputStream pair in a single thread:
* <ol>
* <li>There is no risk of deadlock. Piped streams, which are intended for producing data in one
* thread and consuming it in another, will block on read when no data is available and block on write
* when the buffer is full. In a single-threaded context, avoiding deadlock on read requires checking
* that bytes are available before every read. Avoiding deadlock on write would require checking that
* the buffer is not full before every write, but there is no built-in, publicly-accessible way of doing
* this with a PipedInputStream/PipedOutputStream.</li>
* <li>The buffer can grow. Piped streams use a fixed-size buffer that causes blocking when full. If
* used in a single-thread, this serves as a hard limit on the amount of data that can be written without
* a matching read. This can require arbitrary limits on data size to be imposed by the application. The
* ResizingPipedInputStream imposes no such limitation, but optionally allows for a maximum buffer
* size to be configured to protect against unbounded growth.</li>
* </ol>
*/
public class ResizingPipedInputStream extends InputStream {
/**
* Handler of notifications provided by the ResizingPipedInputStream.
*/
interface NotificationConsumer {
/**
* Bytes have been shifted to the start of the buffer in order to make room for additional bytes
* to be buffered.
* @param leftShiftAmount the amount of the left shift (also: the pre-shift read index of the first shifted
* byte).
*/
void bytesConsolidatedToStartOfBuffer(int leftShiftAmount);
}
/**
* A NotificationConsumer that does nothing.
*/
private static final NotificationConsumer NO_OP_NOTIFICATION_CONSUMER = new NotificationConsumer() {
@Override
public void bytesConsolidatedToStartOfBuffer(int leftShiftAmount) {
// Do nothing.
}
};
/**
* Mask to isolate a single byte.
*/
private static final int SINGLE_BYTE_MASK = 0xFF;
/**
* The initial size of the buffer and the number of bytes by which the size of the buffer will increase
* each time it grows, unless it must grow by a smaller amount to fit within 'maximumBufferSize'.
*/
private final int initialBufferSize;
/**
* The maximum size of the buffer. If the user attempts to buffer more bytes than this, an exception will be raised.
*/
private final int maximumBufferSize;
/**
* Whether to use a boundary to limit the number of available bytes. This can be used to buffer
* arbitrarily-sized chunks of bytes without making them available for consumption. When true,
* the boundary must be manually extended (see {@link #extendBoundary(int)} to make these bytes
* available. When false, all buffered bytes will be available to read.
*/
private final boolean useBoundary;
/**
* The NotificationConsumer currently registered.
*/
private NotificationConsumer notificationConsumer = NO_OP_NOTIFICATION_CONSUMER;
/**
* The raw buffer.
*/
private byte[] buffer;
/**
* View to the raw buffer.
*/
private ByteBuffer byteBuffer;
/**
* @see #capacity()
*/
private int capacity;
/**
* The index of the next byte in the buffer that is available to be read. Always less than or equal to `writeIndex`.
*/
private int readIndex = 0;
/**
* The index at which the next byte received will be written. Always greater than or equal to `readIndex`.
*/
private int writeIndex = 0;
/**
* @see #available()
*/
private int available = 0;
/**
* @see #size()
*/
private int size = 0;
/**
* @see #getBoundary()
*/
private int boundary = 0;
/**
* Constructor.
* @param initialBufferSize the initial size of the buffer. When full, the buffer will grow by this
* many bytes. The buffer always stores bytes contiguously, so growth requires
* allocation of a new buffer capable of holding the new capacity and copying of the
* existing bytes into the new buffer. As such, a size should be chosen carefully
* such that growth is expected to occur rarely, if ever.
*/
public ResizingPipedInputStream(final int initialBufferSize) {
this(initialBufferSize, _Private_IonConstants.ARRAY_MAXIMUM_SIZE, false);
}
/**
* Constructor.
* @param initialBufferSize the initial size of the buffer. When full, the buffer will grow by this
* many bytes. The buffer always stores bytes contiguously, so growth requires
* allocation of a new buffer capable of holding the new capacity and copying of the
* existing bytes into the new buffer. As such, a size should be chosen carefully
* such that growth is expected to occur rarely, if ever.
* @param maximumBufferSize the maximum size of the buffer. If a call to `receive` attempts to transfer an amount
* of bytes that would cause the buffer to exceed this size, a
* {@link BufferOverflowException} will be thrown. Must be greater than or equal to the
* initial buffer size.
* @param useBoundary whether to use a boundary to limit the number of available bytes. This can be used to buffer
* arbitrarily-sized chunks of bytes without making them available for consumption. When true,
* the boundary must be manually extended (see {@link #extendBoundary(int)} to make these bytes
* available. When false, all buffered bytes will be available to read.
*/
ResizingPipedInputStream(final int initialBufferSize, final int maximumBufferSize, final boolean useBoundary) {
if (initialBufferSize < 1) {
throw new IllegalArgumentException("Initial buffer size must be at least 1.");
}
if (maximumBufferSize > _Private_IonConstants.ARRAY_MAXIMUM_SIZE) {
throw new IllegalArgumentException("Initial buffer size must be at most " + _Private_IonConstants.ARRAY_MAXIMUM_SIZE + ".");
}
if (maximumBufferSize < initialBufferSize) {
throw new IllegalArgumentException("Maximum buffer size cannot be less than the initial buffer size.");
}
this.initialBufferSize = initialBufferSize;
this.maximumBufferSize = maximumBufferSize;
this.capacity = initialBufferSize;
buffer = new byte[initialBufferSize];
byteBuffer = ByteBuffer.wrap(buffer, 0, capacity);
this.useBoundary = useBoundary;
}
/**
* Moves all buffered (but not yet read) bytes from 'buffer' to the destination buffer. In total, {@link #size()}
* bytes will be moved.
* @param destinationBuffer the destination buffer, which may be 'buffer' itself or a new buffer.
*/
private void moveBytesToStartOfBuffer(byte[] destinationBuffer) {
if (size > 0) {
System.arraycopy(buffer, readIndex, destinationBuffer, 0, size);
}
if (readIndex > 0) {
notificationConsumer.bytesConsolidatedToStartOfBuffer(readIndex);
}
readIndex = 0;
boundary = available;
writeIndex = size;
}
/**
* @return the number of bytes that can be written at the end of the buffer.
*/
private int freeSpaceAtEndOfBuffer() {
return capacity - writeIndex;
}
/**
* Ensures that there is at least 'minimumNumberOfBytesRequired' bytes of free space in the buffer, growing the
* buffer if necessary. May consolidate buffered bytes, performing an in-order copy and resetting indices
* such that the `readIndex` points to the same byte and the `writeIndex` is positioned after the last
* byte that is available to read.
* @param minimumNumberOfBytesRequired the minimum amount of free space that needs to be available for writing.
*/
private void ensureSpaceInBuffer(int minimumNumberOfBytesRequired) {
if (size < 1 || freeSpaceAtEndOfBuffer() < minimumNumberOfBytesRequired) {
int shortfall = minimumNumberOfBytesRequired - freeSpaceAtEndOfBuffer() - readIndex;
if (shortfall <= 0) {
// Free up space by moving any unread bytes to the start of the buffer and resetting the indices.
moveBytesToStartOfBuffer(buffer);
} else {
// There is not enough space in the buffer even though all available bytes have already been
// moved to the start of the buffer. Growth is required.
int amountToGrow = Math.max(initialBufferSize, shortfall);
if (capacity + amountToGrow > maximumBufferSize) {
amountToGrow = shortfall;
if (capacity + amountToGrow > maximumBufferSize) {
throw new BufferOverflowException();
}
}
byte[] newBuffer = new byte[buffer.length + amountToGrow];
moveBytesToStartOfBuffer(newBuffer);
capacity += amountToGrow;
buffer = newBuffer;
byteBuffer = ByteBuffer.wrap(buffer, readIndex, capacity);
}
}
}
/**
* Buffers a single additional byte, growing the buffer if it is already full.
* @param b the byte to buffer.
*/
public void receive(final int b) {
ensureSpaceInBuffer(1);
buffer[writeIndex] = (byte) b;
writeIndex++;
size++;
if (!useBoundary) {
extendBoundary(1);
}
}
/**
* Buffers `len` additional bytes, growing the buffer if it is already full or if it would become full
* by writing `len` bytes.
* @param b the bytes to buffer.
* @param off the offset into `b` that points to the first byte to buffer.
* @param len the number of bytes to buffer.
*/
public void receive(final byte[] b, final int off, final int len) {
ensureSpaceInBuffer(len);
System.arraycopy(b, off, buffer, writeIndex, len);
writeIndex += len;
size += len;
if (!useBoundary) {
extendBoundary(len);
}
}
/**
* Buffers `b.length` additional bytes.
* @see #receive(byte[], int, int)
* @param b the bytes to buffer.
*/
public void receive(final byte[] b) {
receive(b, 0, b.length);
}
/**
* Buffers up to `len` additional bytes, growing the buffer if it is already full or if it would become full
* by writing `len` bytes. This method will block if and only if the given `InputStream`'s
* {@link InputStream#read(byte[], int, int)} blocks when trying to read `len` bytes. If this is not desired,
* the caller should ensure that the given `InputStream` has at least `len` bytes available before calling
* this method or provide an InputStream implementation that does not block.
* @param input the source of the bytes.
* @param len the number of bytes to attempt to write.
* @return the number of bytes actually written, which will only be less than `len` if
* {@link InputStream#read(byte[], int, int)} returns less than `len`.
* @throws IOException if thrown by the given `InputStream` during read, except for {@link EOFException}. If an
* EOFException is thrown by the `InputStream`, it will be caught and this method will return the number of bytes
* that were received before the exception was thrown.
*/
public int receive(final InputStream input, final int len) throws IOException {
ensureSpaceInBuffer(len);
int numberOfBytesRead;
try {
numberOfBytesRead = input.read(buffer, writeIndex, len);
} catch (EOFException e) {
// Some InputStream implementations (such as GZIPInputStream) will throw EOFException instead of
// returning -1.
numberOfBytesRead = -1;
}
if (numberOfBytesRead > 0) {
writeIndex += numberOfBytesRead;
size += numberOfBytesRead;
} else {
numberOfBytesRead = 0;
}
if (!useBoundary) {
extendBoundary(numberOfBytesRead);
}
return numberOfBytesRead;
}
/**
* {@inheritDoc}
* <p>
* NOTE: This method adheres to the documented behavior of {@link InputStream#read(byte[], int, int)}
* <strong>except</strong> that it never blocks. If a read is attempted before the first write,
* this method will return -1.
*/
@Override
public int read(final byte[] b, final int off, final int len) {
if (b.length == 0 || len == 0) {
return 0;
}
if (available < 1) {
return -1;
}
int bytesToRead = Math.min(available, len);
System.arraycopy(buffer, readIndex, b, off, bytesToRead);
readIndex += bytesToRead;
available -= bytesToRead;
size -= bytesToRead;
return bytesToRead;
}
/**
* Copies all of the available bytes in the buffer without changing the number of bytes available to subsequent
* reads.
* @param outputStream stream to which the bytes will be copied.
* @throws IOException if thrown by {@link OutputStream#write(byte[], int, int)}.
*/
public void copyTo(final OutputStream outputStream) throws IOException {
outputStream.write(buffer, readIndex, available);
}
/**
* Seeks the read index to the given position.
* @param index the index to which to seek. Must not be negative.
*/
void seekTo(int index) {
int amount = index - readIndex;
available -= amount;
size -= amount;
readIndex = index;
}
/**
* @return the current read index.
*/
int getReadIndex() {
return readIndex;
}
/**
* @return the current write index.
*/
int getWriteIndex() {
return writeIndex;
}
/**
* Rewinds the buffer to the given read index and sets 'available' to the given value. Subsequent
* behavior is undefined unless the values resulted from calling {@link #getReadIndex()} and
* {@link #available()} in immediate sequence, without any calls to 'receive' since.
* @param previousReadIndex the read index value to be set.
* @param previousAvailable the available value to be set.
*/
void rewind(final int previousReadIndex, final int previousAvailable) {
readIndex = previousReadIndex;
available = previousAvailable;
boundary = previousReadIndex + previousAvailable;
size = writeIndex - readIndex;
}
/**
* Truncates the buffer to the given write index and sets both 'available' and 'size' to the to the given value.
* Subsequent behavior is undefined unless the values resulted from calling {@link #getWriteIndex()} and
* {@link #available()} in immediate sequence, without any calls to 'read' or 'skip' since. It is the caller's
* responsibility to ensure that calling this method will not result in loss of important data beyond the boundary.
* @param previousWriteIndex the write index value to be set.
* @param previousAvailable the available value to be set.
*/
void truncate(final int previousWriteIndex, final int previousAvailable) {
writeIndex = previousWriteIndex;
available = previousAvailable;
boundary = writeIndex;
size = previousAvailable;
}
/**
* Skips up to `n` buffered bytes. Less than `n` bytes will be skipped if less than `n` bytes are
* available in the buffer.
* @param n the number of bytes to skip.
* @return the number of bytes actually skipped.
*/
@Override
public long skip(final long n) {
if (n < 1 || available < 1) {
return 0;
}
int bytesSkipped = (int) Math.min(available, n);
readIndex += bytesSkipped;
available -= bytesSkipped;
size -= bytesSkipped;
return bytesSkipped;
}
/**
* {@inheritDoc}
* <p>
* NOTE: This method adheres to the documented behavior of {@link InputStream#available()}
* <strong>except</strong> that it always returns the exact number of bytes that are available in the
* buffer.
* @return the exact number of bytes available in the buffer.
*/
@Override
public int available() {
return available;
}
/**
* @return the number of bytes actually buffered, which will be greater than or equal to 'available' if a boundary
* has been set, or equal to `available` if no boundary has been set.
*/
int size() {
return size;
}
/**
* @return the number of bytes buffered beyond the boundary. This is equivalent to subtracting {@link #available()}
* from {@link #size()}.
*/
int availableBeyondBoundary() {
return size - available;
}
/**
* @return the index of the boundary, which is used to mark the last buffered byte that is available for reading.
* The boundary must always fall within [readIndex, writeIndex].
*/
int getBoundary() {
return boundary;
}
/**
* Extends the boundary by the given number of bytes. It is the caller's responsibility to ensure that the
* resulting boundary includes only bytes that have been buffered (i.e. that it does not exceed `writeIndex`).
* @param numberOfBytes the number of bytes by which the boundary should be extended.
*/
void extendBoundary(int numberOfBytes) {
boundary += numberOfBytes;
available += numberOfBytes;
}
/**
* {@inheritDoc}
* <p>
* NOTE: This method adheres to the documented behavior of {@link InputStream#read(byte[], int, int)}
* <strong>except</strong> that it never blocks. If a read is attempted before the first write,
* this method will return -1.
*/
@Override
public int read() {
if (available < 1) {
return -1;
}
int b = buffer[readIndex];
readIndex++;
available--;
size--;
return b & SINGLE_BYTE_MASK;
}
/**
* Peeks the byte at the given index without modifying any internal indexes. It is the caller's responsibility to
* ensure that the given index points to an available byte.
* @param index the index of the byte to peek.
* @return the byte value.
*/
int peek(int index) {
return buffer[index] & SINGLE_BYTE_MASK;
}
/**
* @return the capacity of the buffer, which is always less than or equal to 'maximumBufferSize'.
*/
public int capacity() {
return capacity;
}
/**
* @return the initial capacity of the buffer.
*/
int getInitialBufferSize() {
return initialBufferSize;
}
/**
* Clears the buffer.
*/
void clear() {
readIndex = 0;
writeIndex = 0;
available = 0;
boundary = 0;
size = 0;
}
/**
* Returns a ByteBuffer view of the underlying buffer.
* @param position the start position of the ByteBuffer.
* @param limit the limit of the ByteBuffer.
* @return a ByteBuffer.
*/
ByteBuffer getByteBuffer(int position, int limit) {
// Setting the limit to the capacity first is required because setting the position will fail if the new
// position is outside the limit.
byteBuffer.limit(capacity);
byteBuffer.position(position);
byteBuffer.limit(limit);
return byteBuffer;
}
/**
* Copies bytes from the underlying buffer. It is the caller's responsibility to ensure the requested bytes
* are available.
* @param position the start position from which to read.
* @param destination the buffer to copy into.
* @param destinationOffset the offset of the buffer to copy into.
* @param length the number of bytes to copy.
*/
void copyBytes(int position, byte[] destination, int destinationOffset, int length) {
System.arraycopy(buffer, position, destination, destinationOffset, length);
}
/**
* Moves all bytes starting at 'fromPosition' to 'toPosition', overwriting the bytes in-between. It is the caller's
* responsibility to ensure that the overwritten bytes are not needed.
* @param fromPosition the position to move bytes from. Must be less than or equal to 'writeIndex' and 'boundary'.
* @param toPosition the position to move bytes to. Must be greater than or equal to 'readIndex'.
*/
void consolidate(int fromPosition, int toPosition) {
if (fromPosition > writeIndex || fromPosition > boundary || toPosition < readIndex) {
throw new IllegalArgumentException("Tried to consolidate using an index that violates the constraints.");
}
int indexShift = fromPosition - toPosition;
System.arraycopy(buffer, fromPosition, buffer, toPosition, writeIndex - fromPosition);
size -= indexShift;
available -= indexShift;
writeIndex -= indexShift;
boundary -= indexShift;
// readIndex does not need to change, because none of the consolidated bytes have been read yet.
}
/**
* Registers the given NotificationConsumer.
* @param consumer the NotificationConsumer to register.
*/
void registerNotificationConsumer(NotificationConsumer consumer) {
notificationConsumer = consumer;
}
}