在Room中使用RxJava

原文：Room and RxJava。

更少的重复代码，编译时检查的SQL查询，除此之外还有异步功能和可观察的查询－听起来是不是很牛？有了 Room，这些都成为可能。异步查询返回 LiveData 或者RxJava的Maybe, Single 或者 Flowable。它们都是可观察的查询，可以让你在在数据变更的时候自动获得更新，以便确保UI上反应的是数据库的最新数据。如果你已经在app中使用了RxJava 2 ，那么把Room和Maybe, Single 以及 Flowable一起使用应该会非常轻松。

让我们考虑下面的UI：用户可以查看和编辑他们的username。该数据以及其它的用户信息被保存在数据库里。

为了从数据库获得user，我们可以在UserDao中编写下面的查询：

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

这种方法有两个缺点：

它是一个阻塞式的，同步的调用。
每次数据变更的时候我们都需要手动调用一次这个方法。

Room提供了对数据库中数据进行观察的选项，并通过 RxJava的 Maybe, Single 和 Flowable 对象来执行异步查询。

要让Room支持RxJava 2,在build.gradle文件添加如下的dependencies：

// RxJava support for Room
implementation “android.arch.persistence.room:rxjava2:1.0.0-alpha5”
// Testing support
androidTestImplementation “android.arch.core:core-testing:1.0.0-alpha5”

Maybe

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

可能发生以下事情:

当数据库中没有user，查询没有返回行时，Maybe调用complete。
当数据库中有一个user时，Maybe将触发onSuccess并调用complete。
如果在Maybe的complete调用之后user被更新，什么也不会发生。

Single

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

可能发生以下事情:

当数据库中没有user，查询没有返回行时，Single触发onError（EmptyResultSetException.class）。
当数据库中有一个user时， Single触发onSuccess。
如果在Single.onComplete调用之后user被更新，什么也不会发生。

Flowable

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

可能发生以下事情：

当数据库中没有user，查询没有返回行时，Flowable不会发射，也不会触发onNext, 或者 onError。
当数据库中有一个user时，Flowable会触发onNext。
每当user更新之后，Flowable将自动发射，这样你就可以根据最新的数据来更新UI。

译者注：这里的getUserById返回的是Flowable，如果查询没有结果的话，什么信息也收不到，这在实际开发中是很蛋疼的。但是如果我们把返回类型改成Flowable<List>的话，如果查询没有结果是可以得到一个空list的。作者在回复中讲到之所以第一种情况没有通知是因为，在实现的过程中遇到了bug。不过我个人认为要解决这个问题应该不难吧。

测试

测试一个返回Maybe/Single/Flowable的query跟测试它相应的同步query没有什么大的区别。在UserDaoTest中，确保是用的是in-memory database，这样进程杀死之后存储的测试信息就自动被清理了。

@RunWith(AndroidJUnit4.class)
public class UserDaoTest {
…
private UsersDatabase mDatabase;
@Before
public void initDb() throws Exception {
    mDatabase = Room.inMemoryDatabaseBuilder(
                     InstrumentationRegistry.getContext(),
                     UsersDatabase.class)
            // allowing main thread queries, just for testing
            .allowMainThreadQueries()
            .build();
}
@After
public void closeDb() throws Exception {
    mDatabase.close();
}

在测试中添加InstantTaskExecutorRule，确保Room立即执行了所有的数据库操作。

@Rule
public InstantTaskExecutorRule instantTaskExecutorRule = 
                                      new InstantTaskExecutorRule();

让我们实现一个订阅getUserById的test并在user被插入时检查得到的是否就是我们插入的数据。

@Test
public void insertAndGetUserById() {
    // Given that we have a user in the data source
    mDatabase.userDao().insertUser(USER);
    // When subscribing to the emissions of user
    mDatabase.userDao()
             .getUserById(USER.getId())
             .test()
             // assertValue asserts that there was only one emission
             .assertValue(new Predicate<User>() {
                @Override
                public boolean test(User user) throws Exception {
                    // The emitted user is the expected one
                    return user.getId().equals(USER.getId()) &&
                      user.getUserName().equals(USER.getUserName());
                }
            });
}

好了，以上就是本文的全部。你可以在这里查看使用 Room 和 RxJava的 sample app。