RxJava数据流与SqlBrite和Retrofit

Question

在我的Android应用程序中，我使用域级存储库接口，它支持使用SqlBrite和带有Retrofit可观察对象的网络api实现的本地DB。所以我在Repository中有方法getDomains(): Observable<List<Domain>>，在Retrofit和SqlBrite中有两个相应的方法。我不想连接或合并这两个可观察对象。我希望我的存储库只从SqlBrite获取数据，因为SqlBrite返回QueryObservable，每次底层数据更改时都会触发onNext()，我可以独立运行我的网络请求并将结果存储到SqlBrite，并将我的Observable更新为从网络获取并存储到DB数据。所以我试图实现我的仓库的getDomains()方法如下:

fun getDomains(): Observable<List<Domain>> {
    return db.getDomains()
                   .doOnSubscribe {
                       networkClient.getDomains()
                                    .doOnNext { db.putDomains(it) }
                                    .onErrorReturn{ emptyList() }
                                    .subscribe()
                   }
}

但是在这种情况下，每次客户端都要订阅，每次都要发出网络请求，这就不太好了。我想到了其他do...操作符将请求移动到那里，但是在QueryObservable的情况下，doOnCompleted()永远不会被调用，直到我在某个地方调用toBlocking()，我不会，doOnEach()也不好，因为它每次从db提取项目时都会发出请求。我也尝试使用replay()操作符，但是尽管在这种情况下可观察对象缓存，订阅发生并导致网络请求。那么，如何以理想的方式组合这两个可观察对象呢?

Answer 1

好的，这取决于你的具体用例:例如，假设你想显示本地数据库的最新数据，并通过在后台执行网络请求不时更新数据库。

也许有更好的方法，但也许你可以这样做

 fun <T> createDataAwareObservable(databaseQuery: Observable<T>): Observable<T> =
      stateDeterminer.getState().flatMap {
        when (it) {
          State.UP_TO_DATE -> databaseQuery // Nothing to do, data is up to date so observable can be returned directly
          State.NO_DATA ->
            networkClient.getDomains() // no data so first do the network call
                .flatMap { db.save(it) } // save network call result in database
                .flatMap { databaseQuery } // continue with original observable
          State.SYNC_IN_BACKGROUND -> {
            // Execute sync in background
            networkClient.getDomains()
                .flatMap { db.save(it) }
                .observeOn(backgroundSyncScheduler)
                .subscribeOn(backgroundSyncScheduler)
                .subscribe({}, { Timber.e(it, "Error when starting background sync") }, {})
            // Continue with original observable in parallel, network call will then update database and thanks to sqlbrite databaseQuery will be update automatically
            databaseQuery
          }
        }
      }

因此，在最后创建SQLBrite Observable (QueryObservable)并将其传递给createDataAwareObservable()函数。如果这里没有数据，它将确保从网络加载数据，否则它将检查数据是否应该在后台更新(将其保存到数据库中，然后将自动更新SQLBrite QueryObservable)，或者数据是否为最新的。

基本上你可以这样使用:

createDataAwareObservable( db.getAllDomains() ).subscribe(...)

所以作为这个createDataAwareObservable()的用户，当你作为参数传入时，你总是得到相同类型的Observable<T>。所以基本上你总是订阅db.getAllDomains()。

Answer 2

如果你的问题是你必须订阅你的观察者，每次你想要获得数据，你可以使用中继，它永远不会取消订阅观察者，因为没有实现onComplete

   /**
 * Relay is just an observable which subscribe an observer, but it wont unsubscribe once emit the items. So the pipeline keep open
 * It should return 1,2,3,4,5 for first observer and just 3, 4, 5 fot the second observer since default relay emit last emitted item,
 * and all the next items passed to the pipeline.
 */
@Test
public void testRelay() throws InterruptedException {
    BehaviorRelay<String> relay = BehaviorRelay.create("default");
    relay.subscribe(result -> System.out.println("Observer1:" + result));
    relay.call("1");
    relay.call("2");
    relay.call("3");
    relay.subscribe(result -> System.out.println("Observer2:" + result));
    relay.call("4");
    relay.call("5");
}

另一个例子https://github.com/politrons/reactive/blob/master/src/test/java/rx/relay/Relay.java