Not getting TimeoutException when I use rxJava timeout utility

Question

I am not getting TimeoutException if I don't enter any text within 5 seconds. The below code method will call getMsg() and wait for text input. I added 'timeout( 5, TimeUnit.SECONDS )' to wait only for 5 secs for the input. I want to show timeout error in case user did not enter msg with in 5 seconds.

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.concurrent.TimeUnit;

import rx.Observable;

public class TestRx {

   public static void main( String[] args ) throws IOException {
      Observable.just( getMsg() )
            .timeout( 5, TimeUnit.SECONDS )
            .subscribe( System.out::println,
                  e -> e.printStackTrace() );
      System.out.println( "End..." );
   }

   private static String getMsg() throws IOException {
      BufferedReader reader = new BufferedReader( new InputStreamReader( System.in ) );
      System.out.print( "Enter a msg:" );
      String msg = reader.readLine();
      return msg;
   }
}

akarnokd · Accepted Answer · 2017-09-28 13:10:52Z

5

getMsg() executes way before you get into RxJava. just() doesn't magically make the code in its parenthesis happen in a deferred manner. You need fromCallable:

public static void main( String[] args ) {
    Observable.fromCallable(() -> getMsg() )
        .timeout( 5, TimeUnit.SECONDS )
        .subscribe( System.out::println,
              e -> e.printStackTrace() );
    System.out.println( "End..." );
}

Update

The blocking happens on the main thread which is not interrupted in this setup. The alternative is to use subscribeOn and possible blockingSubscribe to wait for the data or termination:

Observable.fromCallable( () -> getMsg() )
          .subscribeOn(Schedulers.io())             // <----------------------
          .timeout( 5, TimeUnit.SECONDS )
          .blockingSubscribe( System.out::println,
                    e -> e.printStackTrace() );

System.out.println( "End..." );

edited Sep 28, 2017 at 13:10

answered Sep 27, 2017 at 13:31

akarnokd

70.1k14 gold badges161 silver badges199 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

venkata Narayana Over a year ago

Thanks I got it. I think just() works with static values or if it is statement that must be evaluated before calling just() method.

venkata Narayana Over a year ago

I want the program to end in case of timeout exception, but the current one is still in running even after timeout error. Please suggest how to end the program running.

akarnokd Over a year ago

Since the blocking happens on the Java main thread outside the reach of RxJava, you have to run the getMsg() on a background thread. Updated the answer accordingly.

Collectives™ on Stack Overflow

Not getting TimeoutException when I use rxJava timeout utility

1 Answer 1

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related