RxJava error handling for hot observable

Question

I'm pretty new to RxJava and have some questions on patterns etc. I'm creating an observable using the code below:

    public Observable<Volume> getVolumeObservable(Epic epic) {
        return Observable.create(event -> {
            try {
                listeners.add(streamingAPI.subscribeForChartCandles(epic.getName(), MINUTE, new HandyTableListenerAdapter() {
                    @Override
                    public void onUpdate(int i, String s, UpdateInfo updateInfo) {
                        if (updateInfo.getNewValue(CONS_END).equals(ONE)) {
                            event.onNext(new Volume(Integer.parseInt(updateInfo.getNewValue(LAST_TRADED_VOLUME))));
                        }
                    }
                }));
            } catch (Exception e) {
                LOG.error("Error from volume observable", e);
            }
        });
    }

Everything is working as expected, but I have some questions on error handling. If I understand correctly, this is to be viewed as a "hot observble", i.e. events will happen regardless of there being a subscription or not (onUpdate is a callback used by a remote server which I have no control over).

I've chosen not to call onError here since I don't want the observable to stop emitting events in case of a single exception. Is there a better pattern to be used? .retry() comes to mind, but I'm not sure that it makes sense for a hot observable?

Also, how is the observable represented when the subscriptions is created, but before the first onNext is called? Is it just an Observable.empty()

Where do you expect errors to come from? From listeners.add() or from onUpdate()? Can you give an example of error condition that you want subscribers to be notified of. — Yaroslav Stavnichiy
I guess you a little bit misunderstanding hot/cold Observable . This is not hot, every subscriber has their own listener to emit events. And even you didn't unregister your listener in dispose. The observable won't emit event after your dispose because of the Observable.create mechanism. — Phoenix Wang
Could be both listeners.add() and onUpdate(). Unfortunately the API I'm using is very poorly specified. — Daniel
Thanks, I realize now that my producer has to be created outside the Observable.create() for it to be considered hot — Daniel

Yaroslav Stavnichiy Yaroslav Stavnichiy · Accepted Answer · 2017-04-11T10:03:57

1) Your observable is not hot. The distinguishing factor is whether multiple subscribers share the same subscription. Observable.create() invokes subscribe function for every subscriber, i.e. it is cold.

It is easy to make it hot though. Just add share() operator. It will subscribe with first subscriber and unsubscribe with last one. Do not forget to implement unsubscribe functionality with something like this:

event.setCancellable(() -> listeners.remove(...));

2) Errors could be recoverable and not recoverable.

In case you consider an error to be self-recoverable (no action required from your side) you should not call onError as this will kill your observable (no further events would be emitted). You can possibly notify your subscribers by emitting special Volume message with error details attached.

In case an error is fatal, e.g. you have failed to add listener, so there could be no further messages, you should not silently ignore this. Emit onError as your observable is not functional anyway.

In case an error requires actions from you, typically retry, or retry with timeout, you can add one of retryXxx() operators. Do this after create() but before share().

3) Observable is an object with subscribe() method. How exactly it is represented depends on the method you created it with. See source code of create() for example.

RxJava error handling for hot observable

1 Answers