RxJava: listen for events after error

I am not fond of long and wordy articles, thus I’ll try to be concise.

Problem

Imagine situation, when it is acceptable for the stream to include “error” events, and you want to continue listening to events regardless it is error or no, i.e. you do not want the stream to be interrupted.

As you know, the stream will be interrupted after once onError() has been called on a stream.

PublishSubject<String> publishSubject = PublishSubject.create();

publishSubject.subscribe(System.out::println, System.out::println);

publishSubject.onNext("1");
publishSubject.onError(new IllegalArgumentException("err1"));
publishSubject.onNext("2");

This will result in the following output:

1
java.lang.IllegalArgumentException: err1

Note, “2” is not present in the output, because stream is interrupted as soon as onError() is called.

Solution

Rx comes with a Notification wrapper class which can be helpful for this case.

PublishSubject<Notification<String>> publishSubject = PublishSubject.create();

publishSubject.subscribe(stringNotification -> {
    if(stringNotification.isOnError()) {
        System.out.println(stringNotification.getError());
    } else {
        System.out.println(stringNotification.getValue());
    }
});

publishSubject.onNext(Notification.createOnNext("1"));
publishSubject.onNext(Notification.createOnError(new IllegalArgumentException("err1")));
publishSubject.onNext(Notification.createOnNext("2"));

The output is following:

1
java.lang.IllegalArgumentException: err1
2

Pay attention, there is “2” after an error event.

Thanks for reading! Bookmark, because you’ll need this one day!