Observable still running on the UI thread

Question

I'm having an issue with running my RxJava observables on a new thread. I'm trying to run zipped DB queries on Scheduler.io() but it blocks the UI thread regardless. Not sure why.

Code for DB:

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
        KeyMatchLevel keyMatchLevel, int queryLim) {
    return Observable.just(
            searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim));
}

Here's the code where zipping happens:

 public Observable<Set<Airport>> getSearchResultsForAll(String term, int queryLim) {
    List<String> terms = CollectionsUtil.asArrayList(term);
    Observable<List<Airport>> macObservable = onNewThread(
            searchModelsInView(getMacView(), terms,
                    KeyMatchLevel.PREFIXED, queryLim));
    Observable<List<Airport>> nameObservable = onNewThread(
            searchModelsInView(getNameView(), terms,
                    KeyMatchLevel.PREFIXED, queryLim));
    Observable<List<Airport>> codeObservable = onNewThread(
            searchModelsInView(getCodeView(), terms,
                    KeyMatchLevel.PREFIXED, NONE));
    Observable<List<Airport>> regionObservable = onNewThread(
            searchModelsInView(getCityView(), terms,
                    KeyMatchLevel.PREFIXED, NONE));

    ;
    return Observable.zip(macObservable, nameObservable, codeObservable, regionObservable,
            (macList, nameList, codeList, regionList) -> {

                Set<Airport> resultSet = new LinkedHashSet<Airport>();

                Airport mac = macList.get(0);

                if (term.length() > CODE_LEN) {
                    if (lowerCase(mac.getCity()).contains(lowerCase(term))) {
                        handleMACFound(resultSet, mac, regionList);
                        return resultSet;
                    }
                    resultSet.addAll(nameList);
                    resultSet.addAll(regionList);
                    return resultSet;
                }


                boolean macFound = StringUtil.isEquals(lowerCase(term),
                        lowerCase(mac.getCode()));

                if (macFound) {
                    handleMACFound(resultSet, mac, codeList);
                    return resultSet;
                }

                resultSet.addAll(codeList);
                resultSet.addAll(nameList);
                resultSet.addAll(regionList);

                return resultSet;
            });
}

private Observable<List<Airport>> onNewThread(Observable<List<Airport>> observable) {
    return observable.subscribeOn(Schedulers.io());
}

This is where it's called:

public void searchKey(String searchTerms, Airport... excluding) {
    TBDataBase.getAirportDB()
            .getSearchResultsForAll(searchTerms, MAX_AIRPORT_SEARCHED)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(airports -> {
                setResults(airports, excluding);
                listener.onRegularAirportSearch();
                listener.onSearchPerformed();
            });


}

It seems like it should run on the Scheduler.io() but it blocks the main thread so it doesn't.

Answer 1

It looks like the problem is in your searchModelsTask:

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
        KeyMatchLevel keyMatchLevel, int queryLim) {
    return Observable.just(
            searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim));
}

Observable.just expects that you already have the value. So it's executing searchModelsTask before the observable is even created. As a result it doesn't matter much that you are trying to put it on the I/O scheduler. Try using a defer with the just inside:

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
        KeyMatchLevel keyMatchLevel, int queryLim) {
    return Observable.defer(() -> Observable.just(
            searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim)));
}

or

public Observable<List<T>> searchModelsInView(View view, final Collection<String> searchTerms,
            KeyMatchLevel keyMatchLevel, int queryLim) {

return Observable.defer(new Func0<Observable<List<T>>>(){

        @Override
        public Observable<List<T>> call() {
            return Observable.just(
                    searchModelsTask(view, new HashSet<>(searchTerms), keyMatchLevel, queryLim));
        }
    });
}

You could also look into using Observable.callable instead of defer as well.