Exception handling in rxjava

Question

I am trying to get accustomed to rxjava and I am trying to call the below QuoteReader in an Observable. I am not sure how to handle the exception thrown,

public class QuoteReader {
   public Map<String, Object> getQuote() throws IOException{
       OkHttpClient okHttpClient = new OkHttpClient();
       Request request = new Request.Builder().url("http://quotes.rest/qod.json").build();
       Gson gson = new Gson();
       Map<String, Object> responseMap = null;

       try(Response response = okHttpClient.newCall(request).execute()) {
         responseMap = gson.fromJson(response.body().string(), Map.class);
         System.out.println("response map : "+responseMap);
       } catch(IOException ioe) {
         ioe.printStackTrace();
         throw ioe;
       } finally {
         okHttpClient = null;
         request = null;
       }
    return responseMap;
  }
}

The following is the rx code I am trying to write,

rx.Observable.just(new QuoteReader().getQuote()) //compile time error saying unhandled exception
              .subscribe(System.out::println);

How should I update the code to handle the exception. Thanks!

Answer 1

Use fromCallable that allows your method to throw (plus, it gets evaluated lazily and not before you even get into the Observable world):

rx.Observable.fromCallable(() -> new QuoteReader().getQuote())
          .subscribe(System.out::println, Throwable::printStackTrace);

Answer 2

Observable pipeline does not allow you throw Exceptions. You must use runtimeExceptions. So changing your code it should looks like.

     try(Response response = okHttpClient.newCall(request).execute()) {
        responseMap = gson.fromJson(response.body().string(), Map.class);
        System.out.println("response map : "+responseMap);
   } catch(IOException ioe) {
     ioe.printStackTrace();
     new RuntimeException(ioe);

You can see a practical example here https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/errors/ObservableExceptions.java

Answer 3

There is another factory method for an Observable, which is called create. This gives you an Observer as input of a lambda expression. Observer is a kind of callback object with three methods: onNext(T t), onError(Throwable e) and onCompleted. I read you're new to RxJava, so here is a little extra as a sidenote: the Rx contract specifies that an Observer can receive zero or more onNext calls, followed by maybe either an onError or onCompleted call. In regular expression for this looks like: onNext* (onError|onCompleted)?.

Now that you know this, you can implement your operation using Observable.create:

Observable.create(observer -> {
  try {
    observer.onNext(new QuoteReader.getQuote());
    observer.onCompleted();
  }
  catch (Throwable e) {
    observer.onError(e);
  }
}

Notice that this code doesn't do anything until you subscribe to this Observable as you already did in your question!