Innocent observable found dead! Programmer shocked: "but.. I thought that onErrorReturn implementation would keep him safe!"

[bradroid]

The event happened in his class on June 5h, 2015. After an intensive crime scene investigation, it is believed that PublishSubject was used to create an observable/observer, in order to create a search mechanism:

Fields:

    private PublishSubject<String> searchQuerySubject;
    private Subscription searchSubscription;

Callback called on user's text input change

   public void onSearchQueryChanged(String query) {
        searchQuerySubject.onNext(query);
    }

Code from the "initialize" method, which is called first

    searchQuerySubject = PublishSubject.create();
    searchSubscription = searchQuerySubject
                .doOnNext(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        if (Strings.isBlank(s)) {
                            view.hideSearchView();
                        } else {
                            view.showSearchView();
                            view.showProgress();
                        }
                    }
                })
                .debounce(750, TimeUnit.MILLISECONDS)
                .filter(new Func1<String, Boolean>() {
                    @Override
                    public Boolean call(String s) {
                        // Skip searches for empty queries
                        return !Strings.isBlank(s);
                    }
                })
                .flatMap(new Func1<String, Observable<List<ShoppingListItemOffer>>>() {
                    @Override
                    public Observable<List<ShoppingListItemOffer>> call(String searchQuery) {
                        return apiService.searchShoppingListItemsObservable(searchQuery);
                    }
                })
                .onErrorReturn(new Func1<Throwable, List<ShoppingListItemOffer>>() {
                    @Override
                    public List<ShoppingListItemOffer> call(Throwable throwable) {
                        return new ArrayList<>();
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new EndlessObserver<List<ShoppingListItemOffer>>() {
                    @Override
                    public void onNext(List<ShoppingListItemOffer> items) {
                        view.showItems(items);
                    }
                });

The critical point happened when error (mock network error) was returned upon requesting search results from the API:
"the 'onErrorReturn' was invoked, it returned default empty array to the onNext, and even thought I expected that the error was completely swallowed, the searchQuerySubject was killed in the process." - said the witness, who wanted to stay anonymous.

"The user desperately kept inputting letters into the search field, which kept triggering the
searchQuerySubject.onNext(query), but unfortunately, the searchQuerySubject just stopped responding to anything."

The question remains: isn't onErrorReturn implementation meant to swallow all the errors and keep the subject alive, even when network errors happen?

Related cases: ReactiveX/RxJava#1633

[dlew]

I don't see what this has to do with RxAndroid in particular, since onErrorReturn and PublishSubject are both part of the base RxJava project.

Regardless, I think it's a misunderstanding with how onErrorReturn works.

Every Observable must abide by the pattern of N calls to onNext followed by either the terminal onComplete or onError. As a result, if you call searchQuerySubject.onError() then searchQuerySubject will cease to emit anything else afterwards because you've put it into a terminal state.

What onErrorReturn does is just replace an onError with onNext for the downstream Subscriber. It doesn't say anything about what the upstream Observables are doing. So while your downstream Subscriber may not be completed yet, it the upstream Observables have. You can even see this behavior in the marble diagram: http://reactivex.io/documentation/operators/catch.html

[plastiv]

onError at RxJava means exceptional situation, just like Exception at Java. You should try-catch and provide default answer earlier if you want to keep things rolling.

searchQuerySubject
                .doOnNext(...)
                .debounce(...)
                .filter(...)
                .flatMap(new Func1<String, Observable<List<ShoppingListItemOffer>>>() {
                    @Override
                    public Observable<List<ShoppingListItemOffer>> call(String searchQuery) {
                        return apiService.searchShoppingListItemsObservable(searchQuery)
                                                   .onErrorReturn(new Func1<Throwable, List<ShoppingListItemOffer>>() {
                                                            @Override
                                                            public List<ShoppingListItemOffer> call(Throwable throwable) {
                                                                   return new ArrayList<>();
                                                            }
                                                   });
                    }
                })

[headinthebox]

moral of the story, before you teach a class, make sure you understand the basics of the Rx contract :-)

[bradroid]

Looks like there is no any issues with rxAndroid or rxJava - it was just a misunderstanding on how onErrorReturn works, so I'll close this issue. @dlew thank you for your comment.

@headinthebox What I meant was a java class. I am aware that my rx knowledge is too limited to teach.
@plastiv Wow, that worked :) thank you very much.