Package org.apache.commons.io.input

Class ReadAheadInputStream

java.lang.Object

java.io.InputStream

org.apache.commons.io.input.ReadAheadInputStream

All Implemented Interfaces:

java.io.Closeable, java.lang.AutoCloseable

public class ReadAheadInputStream
extends java.io.InputStream

Implements InputStream to asynchronously read ahead from an underlying input stream when a specified amount of data has been read from the current buffer. It does so by maintaining two buffers: an active buffer and a read ahead buffer. The active buffer contains data which should be returned when a read() call is issued. The read ahead buffer is used to asynchronously read from the underlying input stream. When the current active buffer is exhausted, we flip the two buffers so that we can start reading from the read ahead buffer without being blocked by disk I/O.

This class was ported and adapted from Apache Spark commit 933dc6cb7b3de1d8ccaf73d124d6eb95b947ed19.

Since:

2.9.0

Field Summary

Fields

Modifier and Type	Field	Description
private java.nio.ByteBuffer	activeBuffer
private java.util.concurrent.locks.Condition	asyncReadComplete
private boolean	endOfStream
private java.util.concurrent.ExecutorService	executorService
private boolean	isClosed
private boolean	isReading
private boolean	isUnderlyingInputStreamBeingClosed
private java.util.concurrent.atomic.AtomicBoolean	isWaiting
private static java.lang.ThreadLocal<byte[]>	oneByte
private boolean	readAborted
private java.nio.ByteBuffer	readAheadBuffer
private java.lang.Throwable	readException
private boolean	readInProgress
private boolean	shutdownExecutorService
private java.util.concurrent.locks.ReentrantLock	stateChangeLock
private java.io.InputStream	underlyingInputStream

Constructor Summary

Constructors

Modifier	Constructor	Description
	ReadAheadInputStream(java.io.InputStream inputStream, int bufferSizeInBytes)	Creates an instance with the specified buffer size and read-ahead threshold
	ReadAheadInputStream(java.io.InputStream inputStream, int bufferSizeInBytes, java.util.concurrent.ExecutorService executorService)	Creates an instance with the specified buffer size and read-ahead threshold
private	ReadAheadInputStream(java.io.InputStream inputStream, int bufferSizeInBytes, java.util.concurrent.ExecutorService executorService, boolean shutdownExecutorService)	Creates an instance with the specified buffer size and read-ahead threshold

Method Summary

Modifier and Type	Method	Description
int	available()
private void	checkReadException()
void	close()
private void	closeUnderlyingInputStreamIfNecessary()
private boolean	isEndOfStream()
private static java.util.concurrent.ExecutorService	newExecutorService()	Creates a new daemon executor service.
private static java.lang.Thread	newThread(java.lang.Runnable r)	Creates a new daemon thread.
int	read()
int	read(byte[] b, int offset, int len)
private void	readAsync()	Read data from underlyingInputStream to readAheadBuffer asynchronously.
private void	signalAsyncReadComplete()
long	skip(long n)
private long	skipInternal(long n)	Internal skip function which should be called only from skip().
private void	swapBuffers()	Flips the active and read ahead buffer
private void	waitForAsyncReadComplete()

Methods inherited from class java.io.InputStream

mark, markSupported, nullInputStream, read, readAllBytes, readNBytes, readNBytes, reset, transferTo

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

oneByte

private static final java.lang.ThreadLocal<byte[]> oneByte

stateChangeLock

private final java.util.concurrent.locks.ReentrantLock stateChangeLock

activeBuffer

private java.nio.ByteBuffer activeBuffer

readAheadBuffer

private java.nio.ByteBuffer readAheadBuffer

endOfStream

private boolean endOfStream

readInProgress

private boolean readInProgress

readAborted

private boolean readAborted

readException

private java.lang.Throwable readException

isClosed

private boolean isClosed

isUnderlyingInputStreamBeingClosed

private boolean isUnderlyingInputStreamBeingClosed

isReading

private boolean isReading

isWaiting

private final java.util.concurrent.atomic.AtomicBoolean isWaiting

underlyingInputStream

private final java.io.InputStream underlyingInputStream

executorService

private final java.util.concurrent.ExecutorService executorService

shutdownExecutorService

private final boolean shutdownExecutorService

asyncReadComplete

private final java.util.concurrent.locks.Condition asyncReadComplete

Constructor Detail

ReadAheadInputStream

public ReadAheadInputStream(java.io.InputStream inputStream,
                            int bufferSizeInBytes)

Creates an instance with the specified buffer size and read-ahead threshold

Parameters:

inputStream - The underlying input stream.

bufferSizeInBytes - The buffer size.

ReadAheadInputStream

public ReadAheadInputStream(java.io.InputStream inputStream,
                            int bufferSizeInBytes,
                            java.util.concurrent.ExecutorService executorService)

Creates an instance with the specified buffer size and read-ahead threshold

Parameters:

inputStream - The underlying input stream.

bufferSizeInBytes - The buffer size.

executorService - An executor service for the read-ahead thread.

ReadAheadInputStream

private ReadAheadInputStream(java.io.InputStream inputStream,
                             int bufferSizeInBytes,
                             java.util.concurrent.ExecutorService executorService,
                             boolean shutdownExecutorService)

Creates an instance with the specified buffer size and read-ahead threshold

Parameters:

inputStream - The underlying input stream.

bufferSizeInBytes - The buffer size.

executorService - An executor service for the read-ahead thread.

shutdownExecutorService - Whether or not to shutdown the given ExecutorService on close.

Method Detail

newExecutorService

private static java.util.concurrent.ExecutorService newExecutorService()

Creates a new daemon executor service.

Returns:

a new daemon executor service.

newThread

private static java.lang.Thread newThread(java.lang.Runnable r)

Creates a new daemon thread.

Parameters:

r - the thread's runnable.

Returns:

a new daemon thread.

available

public int available()
              throws java.io.IOException

Overrides:

available in class java.io.InputStream

Throws:

java.io.IOException

checkReadException

private void checkReadException()
                         throws java.io.IOException

Throws:

java.io.IOException

close

public void close()
           throws java.io.IOException

Specified by:

close in interface java.lang.AutoCloseable

Specified by:

close in interface java.io.Closeable

Overrides:

close in class java.io.InputStream

Throws:

java.io.IOException

closeUnderlyingInputStreamIfNecessary

private void closeUnderlyingInputStreamIfNecessary()

isEndOfStream

private boolean isEndOfStream()

read

public int read()
         throws java.io.IOException

Specified by:

read in class java.io.InputStream

Throws:

java.io.IOException

read

public int read(byte[] b,
                int offset,
                int len)
         throws java.io.IOException

Overrides:

read in class java.io.InputStream

Throws:

java.io.IOException

readAsync

private void readAsync()
                throws java.io.IOException

Read data from underlyingInputStream to readAheadBuffer asynchronously.

Throws:

java.io.IOException

signalAsyncReadComplete

private void signalAsyncReadComplete()

skip

public long skip(long n)
          throws java.io.IOException

Overrides:

skip in class java.io.InputStream

Throws:

java.io.IOException

skipInternal

private long skipInternal(long n)
                   throws java.io.IOException

Internal skip function which should be called only from skip(). The assumption is that the stateChangeLock is already acquired in the caller before calling this function.

Parameters:

n - the number of bytes to be skipped.

Returns:

the actual number of bytes skipped.

Throws:

java.io.IOException

swapBuffers

private void swapBuffers()

Flips the active and read ahead buffer

waitForAsyncReadComplete

private void waitForAsyncReadComplete()
                               throws java.io.IOException

Throws:

java.io.IOException