Rxjava - How to check if Maybe/Observable is empty properly?

Question

I am accessing database using Maybe and my goal is to call the network if Maybe will complete indicating that there is no data in database (I am using room). I have following functions:

@Override
public Observable<List<Item>> getResults() {
    return getResultsFromDatabase().toObservable().switchIfEmpty(getResultsFromNetwork());
}

@Override
public Observable<List<Item>> getResultsFromNetwork() {
    System.out.println("getting results from network");
    return api.getData().doOnNext(new Consumer<List<Item>>() {
        @Override
        public void accept(List<Item> items) throws Exception {
            itemDao.insert(items);
        }
    });
}

@Override
public Maybe<List<Item>> getResultsFromDatabase() {
    System.out.println("getting coins from database");
    return itemDao.getAllItems();
}

However, the switchIfEmpty works a little unexpectedly, as it is always called, even though there are objects in database.

return getResultsFromDatabase().toObservable().switchIfEmpty(getResultsFromNetwork());

1. Is it possible to transform Maybe into Observable using only toObservable()?
2. Is there another way to check if Maybe will be empty and then act upon it?

Answer 1

Use Single instead of Maybe. Because when there are no rows in db, Maybe will be completed directly without calling error. But Single will result in onError with EmptyResultSetException when there are no rows. We can then use Rx's power to return the data from network call. Here's how:

First change your itemDao.getAllItems() to return Single<List<Item>>. Then change getResultsFromDatabase() to return Single as well like:

@Override
public Single<List<Item>> getResultsFromDatabase() {
    System.out.println("getting coins from database");
    return itemDao.getAllItems();
}

Now you need to use onErrorResumeNext in order to return the data from network call, so change the getResults() to:

@Override
public Observable<List<Item>> getResults () {
            return getResultsFromDatabase()
                    .toObservable()
                    .onErrorResumeNext(new ObservableSource<List<Item>>() {
                        @Override
                        public void subscribe(io.reactivex.Observer<? super List<Item>> observer) {
                            getResultsFromNetwork();
                        }
                    });
        }