Let's get started by creating a badly behaved observable that always throws an exception:

(defn exceptional-obs []
  (rx/observable*
   (fn [observer]
     (rx/on-next observer (throw (Exception. "Oops. Something went wrong")))
     (rx/on-completed observer))))

Now let's watch what happens if we subscribe to it:

(rx/subscribe (->> (exceptional-obs)
                   (rx/map inc))
              (fn [v] (prn-to-repl "result is " v)))

;; Exception Oops. Something went wrong  rx-playground.core/exceptional-obs/fn--1505

The exception thrown by exceptional-obs isn't caught anywhere so it simply bubbles up to the REPL. If this was a web application our users would be presented with a web server error such as the HTTP code 500 – Internal Server Error. Those users would probably not use our system again.

Ideally, we would like to get a chance to handle this exception gracefully, possibly rendering a friendly error message that will let ours users know we care about them.

As we have seen earlier in the chapter, the subscribe function can take up to 3 functions as arguments:

For our purposes we are interested in the onError handler, and using it is straightforward:

(rx/subscribe (->> (exceptional-obs)
                   (rx/map inc))
              (fn [v] (prn-to-repl "result is " v))
              (fn [e] (prn-to-repl "error is " e)))

;; "error is " #<Exception java.lang.Exception: Oops. Something went wrong>

This time, instead of throwing the exception, our error handler gets called with it. This gives us the opportunity to display an appropriate message to our users.

The use of onError gives us a much better experience overall but it isn't very flexible.

Let's imagine a different scenario where we have an observable retrieving data from the network. What if, when this observer fails, we would like to present the user with a cached value instead of an error message?

This is where the catch combinator comes in. It allows us to specify a function to be invoked when the observable throws an exception, much like OnError does.

Differently from OnError, however, catch has to return a new Observable that will be the new source of items from the moment the exception was thrown:

(rx/subscribe (->> (exceptional-obs)
                   (rx/catch Exception e
                       (rx/return 10))
                   (rx/map inc))
              (fn [v] (prn-to-repl "result is " v)))

;; "result is " 11

In the previous example, we are essentially specifying that, whenever exceptional-obs throws, we should return the value 10. We are not limited to single values, however. In fact, we can use any Observable we like as the new source:

(rx/subscribe (->> (exceptional-obs)
                   (rx/catch Exception e
                     (rx/seq->o (range 5)))
                   (rx/map inc))
              (fn [v] (prn-to-repl "result is " v)))

;; "result is " 1
;; "result is " 2
;; "result is " 3
;; "result is " 4
;; "result is " 5

The last error handling combinator we'll examine is retry. This combinator is useful when we know an error or exception is only transient so we should probably give it another shot by resubscribing to the Observable.

First, we'll create an observable that fails when it is subscribed to for the first time. However, the next time it is subscribed to, it succeeds and emits a new item:

(defn retry-obs []
  (let [errored (atom false)]
    (rx/observable*
     (fn [observer]
       (if @errored
         (rx/on-next observer 20)
         (do (reset! errored true)
             (throw (Exception. "Oops. Something went wrong"))))))))

Let's see what happens if we simply subscribe to it:

(rx/subscribe (retry-obs)
              (fn [v] (prn-to-repl "result is " v)))

;; Exception Oops. Something went wrong  rx-playground.core/retry-obs/fn--1476

As expected, the exception simply bubbles up as in our first example. However we know—for the purposes of this example—that this is a transient failure. Let's see what changes if we use retry:

(rx/subscribe (->> (retry-obs)
                   (.retry))
              (fn [v] (prn-to-repl "result is " v)))

;; "result is " 20

Now, our code is responsible for retrying the Observable and as expected we get the correct output.

It's important to note that retry will attempt to resubscribe indefinitely until it succeeds. This might not be what you want so Rx provides a variation, called retryWith, which allows us to specify a predicate function that controls when and if retrying should stop.

All these operators give us the tools we need to build reliable reactive applications and we should always keep them in mind as they are, without a doubt, a great addition to our toolbox. The RxJava wiki on the subject should be referred to for more information: https://github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators.