Room的資料存取需在background thread進行，且撈出資料後通常也是要顯示在UI上，整個運作跟API連線幾乎一樣，如果用過RxJava處理API連線，RxJava和Room的搭配也能很快上手。

Room & RxJava2

Room對於RxJava2的支援可以讓DAO裡的@Query方法直接回傳Observable，接著就能用RxJava的各種功能來處理這個查詢。

Room支援Maybe、Single和Flowable三種Observable，其運作方式可查看Day25的說明。

先加入dependencies：

implementation "android.arch.persistence.room:rxjava2:1.0.0"

將DAO的回傳型態改成Observable，在此大致說明三種Observable的應用：

Maybe

@Query(“SELECT * FROM Users WHERE id = :userId”)
Maybe<User> getUserById(String userId);

如果沒撈到資料會進入onComplete，撈到資料的話進入onSuccess，屬於一次性的查詢，之後User table有更新的話不會收到通知，需重新查詢。

Single

@Query(“SELECT * FROM Users WHERE id = :userId”)
Single<User> getUserById(String userId);

沒撈到資料會進入onError，撈到資料則進入onSuccess，也是屬於一次性的查詢。

Flowable

@Query(“SELECT * FROM Users WHERE id = :userId”)
Flowable<User> getUserById(String userId);

沒撈到資料的話不會有任何反應，撈到資料時進入onNext，這是持續性的，當table更新時會發送新的結果，並且會自動在background thread進行。

應用到我們的程式，搜尋repo時要先找到id清單再查詢repo table，我們改成Flowable和flatMap來做：

@Query("SELECT * FROM RepoSearchResult WHERE query = :query")
public abstract Flowable<RepoSearchResult> rxSearch(String query);

@Query("SELECT * FROM Repo WHERE id in (:repoIds)")
public abstract Flowable<List<Repo>> rxLoadById(List<Integer> repoIds);

viewModel.rxSearch(query)
        .flatMap(new Function<RepoSearchResult, Publisher<List<Repo>>>() {
            @Override
            public Publisher<List<Repo>> apply(RepoSearchResult result) throws Exception {
                return viewModel.rxLoadById(result.repoIds);
            }
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new DisposableSubscriber<List<Repo>>() {
            @Override
            public void onNext(List<Repo> repos) {
                repoAdapter.swapItems(repos);
            }

            @Override
            public void onError(Throwable t) {
            
            }

            @Override
            public void onComplete() {
                        
            }
        });

不用指定io thread因為Room預設就會讓Flowable在背景執行，每當table更新時都會觸發onNext。

也許某些情況中會需要做同步(synchronize)查詢，可以透過RxJava的blockingGet()來完成。

@Query("SELECT * FROM RepoSearchResult WHERE query = :query")
public abstract Maybe<RepoSearchResult> rxSearchSync(String query);

RepoSearchResult result = viewModel.rxSearchSync(query)
                .subscribeOn(Schedulers.io())
                .blockingGet();

blockingGet()的括號中可以設置查詢不到資料時要回傳的預設值，例如blockingGet(new User())。這在API連線時也可以用喔，例如要立即更新token之類的情況。

除了@Query以外，Room的其他存取像是@Insert、@Delete也都要在背景執行，透過RxJava建立Observable就可以輕鬆地完成。

@Insert
public abstract void insert(User user);

Completable.fromAction(new Action() {
            @Override
            public void run() throws Exception {
                // UserDAO.insert(...)
            }
        })
        .subscribeOn(Schedulers.io())
        .subscribe();

Flowable的整體運作跟LiveData很像，優點是我們可以用RxJava的Operator來處理查詢結果，缺點則是失去了LiveData的lifecycle-aware特性，但沒關係，明天會提到RxJava和LiveData的整合，到時就能同時擁有兩方的優點。

GitHub source code:
https://github.com/IvanBean/ITBon2018/tree/day27-rxjava2-room

Reference:
Room & RxJava