IonReaderContinuableTopLevelBinary.java
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package com.amazon.ion.impl;
import com.amazon.ion.IonBufferConfiguration;
import com.amazon.ion.IonException;
import com.amazon.ion.IonReader;
import com.amazon.ion.IonCursor;
import com.amazon.ion.IonType;
import com.amazon.ion.OffsetSpan;
import com.amazon.ion.OversizedValueException;
import com.amazon.ion.RawValueSpanProvider;
import com.amazon.ion.SeekableReader;
import com.amazon.ion.Span;
import com.amazon.ion.SpanProvider;
import com.amazon.ion.SymbolTable;
import com.amazon.ion.system.IonReaderBuilder;
import java.io.IOException;
import java.io.InputStream;
/**
* An optionally continuable (i.e., incremental) binary {@link IonReader} implementation. Continuability is enabled
* using {@code IonReaderBuilder.withIncrementalReadingEnabled(true)}.
* <p>
* When continuable reading is enabled, if
* {@link IonReader#next()} returns {@code null} at the top-level, it indicates that there is not (yet) enough data in
* the stream to complete a top-level value. The user may wait for more data to become available in the stream and
* call {@link IonReader#next()} again to continue reading. Unlike the non-incremental reader, the continuable reader
* will never throw an exception due to unexpected EOF during {@code next()}. If, however, {@link IonReader#close()} is
* called when an incomplete value is buffered, the reader will raise an {@link IonException}.
* </p>
* <p>
* There is one caveat with the continuable reader implementation: it must be able to buffer an entire top-level value
* and any preceding system values (Ion version marker(s) and symbol table(s)) in memory. This means that each value
* and preceding system values must be no larger than any of the following:
* <ul>
* <li>The configured maximum buffer size of the {@link IonBufferConfiguration}.</li>
* <li>The heap memory available in the JVM.</li>
* <li>2GB, because the buffer is held in a Java {@code byte[]}, which is indexed by an {@code int}.</li>
* </ul>
* This will not be a problem for the vast majority of Ion streams, as it is
* rare for a single top-level value or symbol table to exceed a few megabytes in size. However, if the size of the
* stream's values risk exceeding the available memory, then continuable reading must not be used.
* </p>
*/
final class IonReaderContinuableTopLevelBinary extends IonReaderContinuableApplicationBinary implements IonReader, _Private_ReaderWriter, _Private_ByteTransferReader {
// True if continuable reading is disabled.
private final boolean isNonContinuable;
// True if input is sourced from a non-fixed stream and the reader is non-continuable, meaning that its top level
// values are not automatically filled during next().
private final boolean isFillRequired;
// True if a value is in the process of being filled.
private boolean isFillingValue = false;
// The type of value on which the reader is currently positioned.
private IonType type = null;
// The SymbolTable that was transferred via the last call to pop_passed_symbol_table.
private SymbolTable symbolTableLastTransferred = null;
/**
* Constructs a new reader from the given input stream.
* @param builder the builder containing the configuration for the new reader.
* @param alreadyRead the byte array containing the bytes already read (often the IVM).
* @param alreadyReadOff the offset into 'alreadyRead` at which the first byte that was already read exists.
* @param alreadyReadLen the number of bytes already read from `alreadyRead`.
*/
IonReaderContinuableTopLevelBinary(IonReaderBuilder builder, InputStream inputStream, byte[] alreadyRead, int alreadyReadOff, int alreadyReadLen) {
super(builder, inputStream, alreadyRead, alreadyReadOff, alreadyReadLen);
isNonContinuable = !builder.isIncrementalReadingEnabled();
isFillRequired = isNonContinuable;
}
/**
* Constructs a new reader from the given byte array.
* @param builder the builder containing the configuration for the new reader.
* @param data the byte array containing the bytes to read.
* @param offset the offset into the byte array at which the first byte of Ion data begins.
* @param length the number of bytes to be read from the byte array.
*/
IonReaderContinuableTopLevelBinary(IonReaderBuilder builder, byte[] data, int offset, int length) {
super(builder, data, offset, length);
isNonContinuable = !builder.isIncrementalReadingEnabled();
isFillRequired = false;
}
@Override
public SymbolTable pop_passed_symbol_table() {
SymbolTable currentSymbolTable = getSymbolTable();
if (currentSymbolTable == symbolTableLastTransferred) {
// This symbol table has already been returned. Since the contract is that it is a "pop", it should not
// be returned twice.
return null;
}
symbolTableLastTransferred = currentSymbolTable;
if (symbolTableLastTransferred.isLocalTable()) {
// This method is called when transferring the reader's symbol table to either a writer or an IonDatagram.
// Those cases require a mutable copy of the reader's symbol table.
return ((_Private_LocalSymbolTable) symbolTableLastTransferred).makeCopy();
}
return symbolTableLastTransferred;
}
@Override
public boolean hasNext() {
throw new UnsupportedOperationException("Not implemented");
}
/**
* Advances to the next value and attempts to fill it.
*/
private void nextAndFill() {
while (true) {
if (!isFillingValue && nextValue() == IonCursor.Event.NEEDS_DATA) {
return;
}
isFillingValue = true;
if (fillValue() == IonCursor.Event.NEEDS_DATA) {
return;
}
isFillingValue = false;
if (event != IonCursor.Event.NEEDS_INSTRUCTION) {
type = super.getType();
return;
}
// The value was skipped for being too large. Get the next one.
}
}
/**
* Handles the case where the current value extends beyond the end of the reader's internal buffer.
*/
private void handleIncompleteValue() {
if (event == Event.NEEDS_DATA) {
// The reader has already consumed all bytes from the buffer. If non-continuable, this is the end of the
// stream. If continuable, continue to return null from next().
if (isNonContinuable) {
endStream();
}
} else if (isNonContinuable) {
// The reader is non-continuable and has not yet consumed all bytes from the buffer, so it can continue
// reading the incomplete container until the end is reached.
// Each value contains its own length prefix, so it is safe to reset the incomplete flag before attempting
// to read the value.
isValueIncomplete = false;
if (nextValue() == IonCursor.Event.NEEDS_DATA) {
// Attempting to read the partial value required consuming the remaining bytes in the stream, which
// is now at its end.
isValueIncomplete = true;
endStream();
} else {
// The reader successfully positioned itself on a value within an incomplete container.
type = super.getType();
}
}
}
@Override
public IonType next() {
type = null;
if (isValueIncomplete) {
handleIncompleteValue();
} else if (!isSlowMode || isNonContinuable || parent != null) {
if (nextValue() == IonCursor.Event.NEEDS_DATA) {
if (isNonContinuable) {
endStream();
}
} else if (isValueIncomplete && !isNonContinuable) {
// The value is incomplete and the reader is continuable, so the reader must return null from next().
// Setting the event to NEEDS_DATA ensures that if the user attempts to skip past the incomplete
// value, null will continue to be returned.
event = Event.NEEDS_DATA;
} else {
isFillingValue = false;
type = super.getType();
}
} else {
nextAndFill();
}
return type;
}
@Override
public void stepIn() {
super.stepIntoContainer();
type = null;
}
@Override
public void stepOut() {
super.stepOutOfContainer();
type = null;
}
@Override
public IonType getType() {
return type;
}
/**
* Prepares a scalar value to be parsed by ensuring it is present in the buffer.
*/
@Override
void prepareScalar() {
if (!isValueIncomplete) {
if (!isSlowMode || event == IonCursor.Event.VALUE_READY) {
super.prepareScalar();
return;
}
if (isFillRequired) {
if (fillValue() == Event.VALUE_READY) {
super.prepareScalar();
return;
}
if (event == Event.NEEDS_INSTRUCTION) {
throw new OversizedValueException();
}
}
}
throw new IonException("Unexpected EOF.");
}
private static class IonReaderBinarySpan extends DowncastingFaceted implements Span, OffsetSpan {
final long bufferOffset;
final long bufferLimit;
final long totalOffset;
final SymbolTable symbolTable;
/**
* @param bufferOffset the offset of the span's first byte in the cursor's internal buffer.
* @param bufferLimit the offset after the span's last byte in the cursor's internal buffer.
* @param totalOffset the total stream offset of the span's first byte. This can differ from 'bufferOffset' if
* the cursor's internal buffer is refillable, such as when it consumes data from an input
* stream.
* @param symbolTable the symbol table active where the span occurs.
*/
IonReaderBinarySpan(long bufferOffset, long bufferLimit, long totalOffset, SymbolTable symbolTable) {
this.bufferOffset = bufferOffset;
this.bufferLimit = bufferLimit;
this.totalOffset = totalOffset;
this.symbolTable = symbolTable;
}
@Override
public long getStartOffset() {
return totalOffset;
}
@Override
public long getFinishOffset() {
return totalOffset + (bufferLimit - bufferOffset);
}
}
private class SpanProviderFacet implements SpanProvider {
@Override
public Span currentSpan() {
if (type == null) {
throw new IllegalStateException("IonReader isn't positioned on a value");
}
return new IonReaderBinarySpan(
valuePreHeaderIndex,
valueMarker.endIndex,
getTotalOffset(),
getSymbolTable()
);
}
}
private class RawValueSpanProviderFacet implements RawValueSpanProvider {
@Override
public Span valueSpan() {
if (type == null) {
throw new IllegalStateException("IonReader isn't positioned on a value");
}
return new IonReaderBinarySpan(
valueMarker.startIndex,
valueMarker.endIndex,
valueMarker.startIndex,
null
);
}
@Override
public byte[] buffer() {
return buffer;
}
}
private class SeekableReaderFacet extends SpanProviderFacet implements SeekableReader {
@Override
public void hoist(Span span) {
if (! (span instanceof IonReaderBinarySpan)) {
throw new IllegalArgumentException("Span isn't compatible with this reader.");
}
IonReaderBinarySpan binarySpan = (IonReaderBinarySpan) span;
if (binarySpan.symbolTable == null) {
throw new IllegalArgumentException("Span is not seekable.");
}
// Note: setting the limit at the end of the hoisted value causes the reader to consider the end
// of the value to be the end of the stream, in order to comply with the SeekableReader contract. From
// an implementation perspective, this is not necessary; if we leave the buffer's limit unchanged, the
// reader can continue after processing the hoisted value.
restoreSymbolTable(binarySpan.symbolTable);
slice(binarySpan.bufferOffset, binarySpan.bufferLimit, binarySpan.symbolTable.getIonVersionId());
type = null;
}
}
@Override
public boolean transferCurrentValue(_Private_ByteTransferSink writer) throws IOException {
if (hasAnnotations || !isByteBacked() || isInStruct()) {
return false;
}
writer.writeBytes(buffer, (int) valuePreHeaderIndex, (int) (valueMarker.endIndex - valuePreHeaderIndex));
return true;
}
@Override
public boolean isSymbolTableCompatible(SymbolTable symbolTable) {
return isSymbolTableSubsetOf(symbolTable);
}
@Override
public <T> T asFacet(Class<T> facetType) {
if (facetType == SpanProvider.class) {
return facetType.cast(new SpanProviderFacet());
}
// Note: because IonCursorBinary has an internal buffer that can grow, it is possible to relax the restriction
// that readers must have been constructed with a byte array in order to be seekable or provide raw value spans.
// However, it requires some considerations that do not fit well with the existing interfaces. Most importantly,
// because the cursor's internal buffer is a byte array, its size is limited to 2GB. Pinning all bytes after the
// first span is requested could lead to buffer overflow for large streams, so there would need to be a way for
// a user to release a span and allow the reader to reclaim its bytes. This functionality is not included in the
// existing Span interfaces. See: amzn/ion-java/issues/17
if (isByteBacked()) {
if (facetType == SeekableReader.class) {
return facetType.cast(new SeekableReaderFacet());
}
if (facetType == RawValueSpanProvider.class) {
return facetType.cast(new RawValueSpanProviderFacet());
}
}
return null;
}
@Override
public void close() {
if (!isNonContinuable) {
endStream();
}
super.close();
}
}