VERY IMPORTANT NOTES *
NonBlockingInputStream out of a normal blocking
* stream. Note that this call also spawn a separate thread to perform the
* blocking I/O on behalf of the thread that is using this class. The
* {@link #shutdown()} method must be called in order to shut this thread down.
* @param in The input stream to wrap
* @param isNonBlockingEnabled If true, then the non-blocking methods
* {@link #read(long)} and {@link #peek(long)} will be available and,
* more importantly, the thread will be started to provide support for the
* feature. If false, then this class acts as a clean-passthru for the
* underlying I/O stream and provides very little overhead.
*/
public NonBlockingInputStream (InputStream in, boolean isNonBlockingEnabled) {
this.in = in;
this.nonBlockingEnabled = isNonBlockingEnabled;
if (isNonBlockingEnabled) {
Thread t = new Thread(this);
t.setName("NonBlockingInputStreamThread");
t.setDaemon(true);
t.start();
}
}
/**
* Shuts down the thread that is handling blocking I/O. Note that if the
* thread is currently blocked waiting for I/O it will not actually
* shut down until the I/O is received. Shutting down the I/O thread
* does not prevent this class from being used, but causes the
* non-blocking methods to fail if called and causes {@link #isNonBlockingEnabled()}
* to return false.
*/
public synchronized void shutdown() {
if (!isShutdown && nonBlockingEnabled) {
isShutdown = true;
notify();
}
}
/**
* Non-blocking is considered enabled if the feature is enabled and the
* I/O thread has not been shut down.
* @return true if non-blocking mode is enabled.
*/
public boolean isNonBlockingEnabled() {
return nonBlockingEnabled && !isShutdown;
}
@Override
public void close() throws IOException {
/*
* The underlying input stream is closed first. This means that if the
* I/O thread was blocked waiting on input, it will be woken for us.
*/
in.close();
shutdown();
}
@Override
public int read() throws IOException {
if (nonBlockingEnabled)
return read(0L, false);
return in.read ();
}
/**
* Peeks to see if there is a byte waiting in the input stream without
* actually consuming the byte.
*
* @param timeout The amount of time to wait, 0 == forever
* @return -1 on eof, -2 if the timeout expired with no available input
* or the character that was read (without consuming it).
* @throws IOException
*/
public int peek(long timeout) throws IOException {
if (!nonBlockingEnabled || isShutdown) {
throw new UnsupportedOperationException ("peek() "
+ "cannot be called as non-blocking operation is disabled");
}
return read(timeout, true);
}
/**
* Attempts to read a character from the input stream for a specific
* period of time.
* @param timeout The amount of time to wait for the character
* @return The character read, -1 if EOF is reached, or -2 if the
* read timed out.
* @throws IOException
*/
public int read(long timeout) throws IOException {
if (!nonBlockingEnabled || isShutdown) {
throw new UnsupportedOperationException ("read() with timeout "
+ "cannot be called as non-blocking operation is disabled");
}
return read(timeout, false);
}
/**
* Attempts to read a character from the input stream for a specific
* period of time.
* @param timeout The amount of time to wait for the character
* @return The character read, -1 if EOF is reached, or -2 if the
* read timed out.
* @throws IOException
*/
private synchronized int read(long timeout, boolean isPeek) throws IOException {
/*
* If the thread hit an IOException, we report it.
*/
if (exception != null) {
assert ch == -2;
IOException toBeThrown = exception;
if (!isPeek)
exception = null;
throw toBeThrown;
}
/*
* If there was a pending character from the thread, then
* we send it. If the timeout is 0L or the thread was shut down
* then do a local read.
*/
if (ch >= -1) {
assert exception == null;
}
else if ((timeout == 0L || isShutdown) && !threadIsReading) {
ch = in.read();
}
else {
/*
* If the thread isn't reading already, then ask it to do so.
*/
if (!threadIsReading) {
threadIsReading = true;
notify();
}
boolean isInfinite = (timeout <= 0L);
/*
* So the thread is currently doing the reading for us. So
* now we play the waiting game.
*/
while (isInfinite || timeout > 0L) {
long start = System.currentTimeMillis ();
try {
wait(timeout);
}
catch (InterruptedException e) {
/* IGNORED */
}
if (exception != null) {
assert ch == -2;
IOException toBeThrown = exception;
if (!isPeek)
exception = null;
throw toBeThrown;
}
if (ch >= -1) {
assert exception == null;
break;
}
if (!isInfinite) {
timeout -= System.currentTimeMillis() - start;
}
}
}
/*
* ch is the character that was just read. Either we set it because
* a local read was performed or the read thread set it (or failed to
* change it). We will return it's value, but if this was a peek
* operation, then we leave it in place.
*/
int ret = ch;
if (!isPeek) {
ch = -2;
}
return ret;
}
/**
* This version of read() is very specific to jline's purposes, it
* will always always return a single byte at a time, rather than filling
* the entire buffer.
*/
@Override
public int read (byte[] b, int off, int len) throws IOException {
if (b == null) {
throw new NullPointerException();
} else if (off < 0 || len < 0 || len > b.length - off) {
throw new IndexOutOfBoundsException();
} else if (len == 0) {
return 0;
}
int c;
if (nonBlockingEnabled)
c = this.read(0L);
else
c = in.read();
if (c == -1) {
return -1;
}
b[off] = (byte)c;
return 1;
}
//@Override
public void run () {
Log.debug("NonBlockingInputStream start");
boolean needToShutdown = false;
boolean needToRead = false;
while (!needToShutdown) {
/*
* Synchronize to grab variables accessed by both this thread
* and the accessing thread.
*/
synchronized (this) {
needToShutdown = this.isShutdown;
needToRead = this.threadIsReading;
try {
/*
* Nothing to do? Then wait.
*/
if (!needToShutdown && !needToRead) {
wait(0);
}
}
catch (InterruptedException e) {
/* IGNORED */
}
}
/*
* We're not shutting down, but we need to read. This cannot
* happen while we are holding the lock (which we aren't now).
*/
if (!needToShutdown && needToRead) {
int charRead = -2;
IOException failure = null;
try {
charRead = in.read();
}
catch (IOException e) {
failure = e;
}
/*
* Re-grab the lock to update the state.
*/
synchronized (this) {
exception = failure;
ch = charRead;
threadIsReading = false;
notify();
}
}
}
Log.debug("NonBlockingInputStream shutdown");
}
}