Making a thread which cancels a InputStream.read() call if 'x' time has passed

Question

I currently have a working I/O stream from Android's BluetoothChat Example, but have run into problems. My application connects via bluetooth to a bluetooth module, which in turn sends a signal to a device the module is physically attached to.

My program calls read() on an input stream, and if there is data being sent the program executes smoothly with no problems. However, the way the stream is implemented there is no protection against an interrupted connection. If the module is physically removed from the device, or if the device doesn't send any signals back, my code simply sits and waits at the InputStream.read() call.

My read() call looks like this:

try {
    Log.i( "1) I/O", "available bits: " + mmInStream.available() );
    bytes = mmInStream.read(buffer, 0, length);
    Log.i( "2) I/O", "available bits: " + mmInStream.available() );
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                        .sendToTarget();
} catch (Exception e) {
    Log.i(TAG,  "Catch Statement" );
    Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
    Bundle bundle = new Bundle();
    bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
    msg.setData(bundle);
    mHandler.sendMessage(msg);
    Log.e(TAG, "disconnected a", e);
    connectionLost();

    // Start the service over to restart listening mode
    BluetoothService.this.start();
    //break;
}

When my program acts correctly, both of the Log calls in the try block return values of 0 for mmInStream.available(). When the input stream is interrupted, the initial Log call returns a 0, and the second is never called. My program then ends up crashing before the catch block is every reached.

I have been looking for several days now to fix this, and have found numerous solutions, but they have either not worked, or I do not understand them.

1) Using a scanner for the InputStream is shown below. This provided no help and also timed out while reading.

Scanner scan = new Scanner(new InputStreamReader(mmInStream));
scan.useDelimiter( "[\\r\\n]+" );
String readIn;

try {
    readIn = scan.next();
    scan = null;
    tempB = readIn.getBytes( Charset.forName( "US-ASCII" ) );
    append = "\r\n".getBytes( Charset.forName( "US-ASCII" ) );
    for( int i = 0; i < length; i++ ) {
        if( i == length - 1 ) {
            buffer[i] = append[1];
        } else if ( i == length - 2 ) {
            buffer[i] = append[0];
        } else {
            buffer[i] = tempB[i];
        }
    }
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                        .sendToTarget();
} catch (Exception e) {
    Log.i(TAG,  "Catch Statement" );
                Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
                Bundle bundle = new Bundle();
                bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
                msg.setData(bundle);
                mHandler.sendMessage(msg);
                Log.e(TAG, "disconnected a", e);
                connectionLost();

                // Start the service over to restart listening mode
                BluetoothService.this.start();
                //break;
            }

2) I have tried running a Thread which would cancel the read call after X amount of time, but it would not work correctly:

public void run(int length) throws IOException {
    buffer = new byte[1024];
    length1 = length;
    Thread myThread = new Thread(new Runnable() {
        public void run() {
            try {
                bytes = mmInStream.read( buffer, 0, length1 );
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    });

    synchronized (myThread) {
        myThread.start();
        try {
            myThread.wait(500);
            if(myThread.isAlive()) {
                mmInStream.close();
                Log.i( "InStream", "Timeout exceeded!");
            }
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }
   try {
        myThread.run();
        mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                    .sendToTarget();
   } catch (IOException e) {
            Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
            Bundle bundle = new Bundle();
            bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
            msg.setData(bundle);
            mHandler.sendMessage(msg);
            connectionLost();
            BluetoothService.this.start();
   }

After those two options didn't work, I have been trying to look into Java NIO or AsyncTask, but all of this seems like way too much stuff to add for recognizing an I/O timeout. I have also seen that some Sockets support a timeout feature using .setSoTimeout(), however this is a BluetoothSocket and from what I've found they do not support this feature.

Since there is no I/O class which supports a read() method that takes a timeout length as a parameter, or timeout at all, it seems to me that adding a Thread would be the simplest implementation. Is this wrong? Any information on what I'm doing wrong with the above methods, or how to incorporate Java NIO/AsyncTask would be greatly appreciated.

EDIT:

This is the new thread code I tried, I am currently changing it to what the given answer shows and trying that. I will post that if it doesn't work after.

Thread myThread = new Thread(new Runnable() {
            public void run() {
                try {
                    bytes = mmInStream.read( buffer, 0, length1 );
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });

        synchronized (myThread) {
            try {
                myThread.wait(6000);
                Log.i( "InStream", "After wait" );
                if(myThread.isAlive()) {
                    Log.i( "InStream", "Timeout exceeded2!");
                    myThread.interrupt();
                    Log.i( "InStream", "Timeout exceeded!");
                } else {
                    myThread.interrupt();
                }
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                Log.i( "InStream", "Exception Caught" );
                e.printStackTrace();
            }

        }

EDIT 2:

I have tried the answer Dheerej has given below. I get an IllegalMonitorStateException on the wait() function call. I tried as it was shown in the answer, then also tried myThread.wait() instead of Thread.currentThread.wait(). I'm assuming this exception is being thrown because this is myThread object is being created and ran within another thread. Anyway, the code below is almost identical to Dheerej's answer.

        int length1 = length;
            Thread myThread = new Thread(new Runnable() {
                public void run() {
                    buffer = new byte[1024];
                    try {
                        bytes = mmInStream.read(buffer, 0, length1);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                                .sendToTarget();
                }
            });

            myThread.start();
            try {
                //Thread.currentThread().wait(500);
                myThread.wait( 1000 );              // Line 533
            } catch (InterruptedException e) {
                e.printStackTrace();
                //Log.i(TAG,  "Catch Statement" );
                Message msg = mHandler.obtainMessage(MainMenu.MESSAGE_TOAST);
                Bundle bundle = new Bundle();
                bundle.putString( TOAST, "Device has disconnected from the Bluetooth Module." );
                msg.setData(bundle);
                mHandler.sendMessage(msg);
                Log.e(TAG, "disconnected a", e);
                connectionLost();

                // Start the service over to restart listening mode
                BluetoothService.this.start();
            }

            if (myThread.isAlive()) {
                mmInStream.close(); // Alternatively try: myThread.interrupt()
            }

This is the resulting LogCat. The error says it starts in line 533, which is the wait() call above:

12-28 17:44:18.765: D/BLZ20_WRAPPER(3242): blz20_wrp_poll: return 1
12-28 17:44:18.765: D/BLZ20_WRAPPER(3242): blz20_wrp_write: wrote 3 bytes out of 3 on fd 62
12-28 17:44:18.769: W/NATIVE CODE(3242): -4) baud9600=1, goodbaud=1
12-28 17:44:18.769: D/AndroidRuntime(3242): Shutting down VM
12-28 17:44:18.769: W/dalvikvm(3242): threadid=1: thread exiting with uncaught exception (group=0x40015578)
12-28 17:44:18.773: E/AndroidRuntime(3242): FATAL EXCEPTION: main
12-28 17:44:18.773: E/AndroidRuntime(3242): java.lang.IllegalMonitorStateException: object not locked by thread before wait()
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.Object.wait(Native Method)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.Object.wait(Object.java:395)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.BluetoothService$ConnectedThread.run(BluetoothService.java:533)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.BluetoothService.read(BluetoothService.java:326)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.BluetoothService.changeitJava(BluetoothService.java:669)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.RelayAPIModel$NativeCalls.changeItJavaWrapper(RelayAPIModel.java:490)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.RelayAPIModel$NativeCalls.InitRelayJava(Native Method)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at my.eti.commander.MainMenu$1.handleMessage(MainMenu.java:547)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at android.os.Handler.dispatchMessage(Handler.java:99)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at android.os.Looper.loop(Looper.java:130)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at android.app.ActivityThread.main(ActivityThread.java:3687)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.reflect.Method.invokeNative(Native Method)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at java.lang.reflect.Method.invoke(Method.java:507)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:842)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:600)
12-28 17:44:18.773: E/AndroidRuntime(3242):     at dalvik.system.NativeStart.main(Native Method)
12-28 17:44:18.781: D/BLZ20_ASOCKWRP(3242): asocket_read
12-28 17:44:18.781: I/BLZ20_WRAPPER(3242): blz20_wrp_poll: nfds 2, timeout -1 ms
12-28 17:44:18.890: D/BLZ20_WRAPPER(3242): blz20_wrp_poll: transp poll : (fd 62) returned r_ev [POLLIN ] (0x1)
12-28 17:44:18.890: D/BLZ20_WRAPPER(3242): blz20_wrp_poll: return 1
12-28 17:44:18.890: D/BLZ20_WRAPPER(3242): blz20_wrp_read: read 5 bytes out of 5 on fd 62

Answer 1

Try this first:

try {
    int available = 0;

    while (true)
    {
        int available = mmInStream.available();
        if (available > 0) { break; }
        Thread.sleep(1);
        // here you can optionally check elapsed time, and time out
    }

    Log.i( "1) I/O", "available bits: " + available );
    bytes = mmInStream.read(buffer, 0, length);
    Log.i( "2) I/O", "available bits: " + mmInStream.available() );
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer).sendToTarget();
} catch (Exception e) {
    ...
}

In your original code, you call available() before read(), typically there is no data waiting to be read. Then you call read(), which blocks and waits for data, then reads all of it. Then you call available() again and once again there is no data, because it has all been read :) Better: sleep until available() returns nonzero, then read. However, this may not work, because available() is always allowed to return 0 (even if data is actually available).

If the above doesn't work, try the technique from this question: Is it possible to read from a InputStream with a timeout?

Callable<Integer> readTask = new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        return mmInStream.read(buffer, 0, length);
    }
}

try {
    Future<Integer> future = executor.submit(readTask);
    bytes = future.get(100, TimeUnit.MILLISECONDS);
    mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer).sendToTarget();
} catch (TimeoutException e) {
    // deal with timeout in the read call
} catch (Exception e) {
    ...
}

Lastly, the BluetoothSocket docs say that you can close the socket from any thread and that takes effect immediately. So you could simply have a watchdog thread and if the read call hasn't succeeded call close() on the socket which would cause the blocked read() to return with an error. This was what Dheeraj suggested above, but you only need to call close() when the other thread is stuck (due to network error/connection lost/etc): otherwise just check on its progress once in a while but don't close as long as your read hasn't taken too long.

It certainly looks like the lack of timeouts (and the impossibility of interrupting a blocked read() from the outside) has been a major ongoing pain point in Java for a long time.

See also:

Is it possible to read from a InputStream with a timeout? (uses Callable/Future)

Can I set a timeout for a InputStream's read() function? (uses Socket.setSoTimeout())

How to kill a BufferedInputStream .read() call (uses InterruptibleChannel)

How to stop a thread waiting in a blocking read operation in Java?

Answer 2

Try this code which expands on my comment above:

public void run(final int length) {
    Thread myThread = new Thread(new Runnable() {
        public void run() {
            buffer = new byte[1024];
            try {
                bytes = mmInStream.read(buffer, 0, length);
            } catch (IOException e) {
                e.printStackTrace();
            }
            mHandler.obtainMessage(MainMenu.MESSAGE_READ, bytes, -1, buffer)
                        .sendToTarget();
        }
    });

    myThread.start();
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    if (myThread.isAlive()) {
        mmInStream.close(); // Alternatively try: myThread.interrupt()
    }
}