Is it possible to read from a InputStream with a t

Question

Specifically, the problem is to write a method like this:

int maybeRead(InputStream in, long timeout)

where the return value is the same as in.read() if data is available within 'timeout' milliseconds, and -2 otherwise. Before the method returns, any spawned threads must exit.

To avoid arguments, the subject here java.io.InputStream, as documented by Sun (any Java version). Please note this is not as simple as it looks. Below are some facts which are supported directly by Sun's documentation.

1. The in.read() method may be non-interruptible.

2. Wrapping the InputStream in a Reader or InterruptibleChannel doesn't help, because all those classes can do is call methods of the InputStream. If it were possible to use those classes, it would be possible to write a solution that just executes the same logic directly on the InputStream.

3. It is always acceptable for in.available() to return 0.

4. The in.close() method may block or do nothing.

5. There is no general way to kill another thread.

Answer 1

If your InputStream is backed by a Socket, you can set a Socket timeout (in milliseconds) using setSoTimeout. If the read() call doesn't unblock within the timeout specified, it will throw a SocketTimeoutException.

Just make sure that you call setSoTimeout on the Socket before making the read() call.

Answer 2

Here is a way to get a NIO FileChannel from System.in and check for availability of data using a timeout, which is a special case of the problem described in the question. Run it at the console, don't type any input, and wait for the results. It was tested successfully under Java 6 on Windows and Linux.

import java.io.FileInputStream;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedByInterruptException;

public class Main {

    static final ByteBuffer buf = ByteBuffer.allocate(4096);

    public static void main(String[] args) {

        long timeout = 1000 * 5;

        try {
            InputStream in = extract(System.in);
            if (! (in instanceof FileInputStream))
                throw new RuntimeException(
                        "Could not extract a FileInputStream from STDIN.");

            try {
                int ret = maybeAvailable((FileInputStream)in, timeout);
                System.out.println(
                        Integer.toString(ret) + " bytes were read.");

            } finally {
                in.close();
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    /* unravels all layers of FilterInputStream wrappers to get to the
     * core InputStream
     */
    public static InputStream extract(InputStream in)
            throws NoSuchFieldException, IllegalAccessException {

        Field f = FilterInputStream.class.getDeclaredField("in");
        f.setAccessible(true);

        while( in instanceof FilterInputStream )
            in = (InputStream)f.get((FilterInputStream)in);

        return in;
    }

    /* Returns the number of bytes which could be read from the stream,
     * timing out after the specified number of milliseconds.
     * Returns 0 on timeout (because no bytes could be read)
     * and -1 for end of stream.
     */
    public static int maybeAvailable(final FileInputStream in, long timeout)
            throws IOException, InterruptedException {

        final int[] dataReady = {0};
        final IOException[] maybeException = {null};
        final Thread reader = new Thread() {
            public void run() {                
                try {
                    dataReady[0] = in.getChannel().read(buf);
                } catch (ClosedByInterruptException e) {
                    System.err.println("Reader interrupted.");
                } catch (IOException e) {
                    maybeException[0] = e;
                }
            }
        };

        Thread interruptor = new Thread() {
            public void run() {
                reader.interrupt();
            }
        };

        reader.start();
        for(;;) {

            reader.join(timeout);
            if (!reader.isAlive())
                break;

            interruptor.start();
            interruptor.join(1000);
            reader.join(1000);
            if (!reader.isAlive())
                break;

            System.err.println("We're hung");
            System.exit(1);
        }

        if ( maybeException[0] != null )
            throw maybeException[0];

        return dataReady[0];
    }
}

Interestingly, when running the program inside NetBeans 6.5 rather than at the console, the timeout doesn't work at all, and the call to System.exit() is actually necessary to kill the zombie threads. What happens is that the interruptor thread blocks (!) on the call to reader.interrupt(). Another test program (not shown here) additionally tries to close the channel, but that doesn't work either.

Answer 3

As jt said, NIO is the best (and correct) solution. If you really are stuck with an InputStream though, you could either

1. Spawn a thread who's exclusive job is to read from the InputStream and put the result into a buffer which can be read from your original thread without blocking. This should work well if you only ever have one instance of the stream. Otherwise you may be able to kill the thread using the deprecated methods in the Thread class, though this may cause resource leaks.

2. Rely on isAvailable to indicate data that can be read without blocking. However in some cases (such as with Sockets) it can take a potentially blocking read for isAvailable to report something other than 0.

Answer 4

I have not used the classes from the Java NIO package, but it seems they might be of some help here. Specifically, java.nio.channels.Channels and java.nio.channels.InterruptibleChannel.

Answer 5

I would question the problem statement rather than just accept it blindly. You only need timeouts from the console or over the network. If the latter you have Socket.setSoTimeout() and HttpURLConnection.setReadTimeout() which both do exactly what is required, as long as you set them up correctly when you construct/acquire them. Leaving it to an arbitrary point later in the application when all you have is the InputStream is poor design leading to a very awkward implementation.

Answer 6

Using inputStream.available()

It is always acceptable for System.in.available() to return 0.

I've found the opposite - it always returns the best value for the number of bytes available. Javadoc for InputStream.available():

Returns an estimate of the number of bytes that can be read (or skipped over) 
from this input stream without blocking by the next invocation of a method for 
this input stream.

An estimate is unavoidable due to timing/staleness. The figure can be a one-off underestimate because new data are constantly arriving. However it always "catches up" on the next call - it should account for all arrived data, bar that arriving just at the moment of the new call. Permanently returning 0 when there are data fails the condition above.

First Caveat: Concrete subclasses of InputStream are responsible for available()

InputStream is an abstract class. It has no data source. It's meaningless for it to have available data. Hence, javadoc for available() also states:

The available method for class InputStream always returns 0.

This method should be overridden by subclasses.

And indeed, the concrete input stream classes do override available(), providing meaningful values, not constant 0s.

Second Caveat: Ensure you use carriage-return when typing input in Windows.

If using System.in, your program only receives input when your command shell hands it over. If you're using file redirection/pipes (e.g. somefile > java myJavaApp or somecommand | java myJavaApp ), then input data are usually handed over immediately. However, if you manually type input, then data handover can be delayed. E.g. With windows cmd.exe shell, the data are buffered within cmd.exe shell. Data are only passed to the executing java program following carriage-return (control-m or <enter>). That's a limitation of the execution environment. Of course, InputStream.available() will return 0 for as long as the shell buffers the data - that's correct behaviour; there are no available data at that point. As soon as the data are available from the shell, the method returns a value > 0. NB: Cygwin uses cmd.exe too.

Simplest solution (no blocking, so no timeout required)

Just use this:

    byte[] inputData = new byte[1024];
    int result = is.read(inputData, 0, is.available());  
    // result will indicate number of bytes read; -1 for EOF with no data read.

OR equivalently,

    BufferedReader br = new BufferedReader(new InputStreamReader(System.in, Charset.forName("ISO-8859-1")),1024);
    // ...
         // inside some iteration / processing logic:
         if (br.ready()) {
             int readCount = br.read(inputData, bufferOffset, inputData.length-bufferOffset);
         }

Richer Solution (maximally fills buffer within timeout period)

Declare this:

public static int readInputStreamWithTimeout(InputStream is, byte[] b, int timeoutMillis)
     throws IOException  {
     int bufferOffset = 0;
     long maxTimeMillis = System.currentTimeMillis() + timeoutMillis;
     while (System.currentTimeMillis() < maxTimeMillis && bufferOffset < b.length) {
         int readLength = java.lang.Math.min(is.available(),b.length-bufferOffset);
         // can alternatively use bufferedReader, guarded by isReady():
         int readResult = is.read(b, bufferOffset, readLength);
         if (readResult == -1) break;
         bufferOffset += readResult;
     }
     return bufferOffset;
 }

Then use this:

    byte[] inputData = new byte[1024];
    int readCount = readInputStreamWithTimeout(System.in, inputData, 6000);  // 6 second timeout
    // readCount will indicate number of bytes read; -1 for EOF with no data read.