使用 RxJava、Retrofit 和 Room 从网络服务中获取实体并缓存在本地数据库中

如何解决使用 RxJava、Retrofit 和 Room 从网络服务中获取实体并缓存在本地数据库中

问题：如何将两个 RxJava observable 合并为一个 observable，以便

1. 每当两个源 observable 中的一个以交错方式发出新值时，生成的 observable 就会发出一个新值，
2. 第一个源的值保持不变并且
3. 第二个源的值是否根据第一个源的最近发出的值进行转换？

更正式的（我不能在这里画大理石图），我想要类似下面的东西。设 T1 是 observable 1 的类型和结果 observable，T2 是 observable 2 的类型，f 是映射 T1 x T2 --> T1。

Observable1<T1>:       o1            o2  o3  o4                      o5            o6  o7
Observable2<T2>:              u1                    u2        u3            u4                u5
---------------------------------------------------------------------------------------------------
Result Observable<T1:  o1  f(o1,u1)  o2  o3  o4  f(o4,u2)  f(o4,u3)  o5  f(o5,u4)  o6  o7  f(u7,u5)

背景故事（也许问题是笼统的，对于特定用例有更简单的解决方案）

问题

1. 来自网络服务后端的实体应存储在本地数据库中以进行缓存。
2. 如果 UI 请求实体，该实体应立即从数据库提供给 UI。
3. 应该永远观察数据库，即如果数据库发生更改，则更改应通过可观察对象传播到 UI
4. 当且仅当实体看起来已过时，才应从网络服务中获取新版本，插入数据库并重新发送到 UI。
5. 实体应被封装到一个“资源”包装器中，该包装器为实际实体配备一个状态以指示实体的“新鲜度”以及可选的错误消息。
6. 如果后台更新失败，数据库中实体的最新版本将与失败更新的错误消息一起提供。
7. 代码应使用 RxJava。

后端 API 的签名

class RetrofitClient {
  // Fetches the entity from the webserver,emits the entity and completes
  public Single<Entity> getEntity( long id );
}

class RoomEntityDao {
  // Inserts or updates the given entity in DB,emits the primary index on success and completes
  public Single<Long> insert( final @NonNull Entity entity );
  
  // Observes the entity,whenever the entity is updated in DB the updated entity is emitted; never completes
  public Observable<Entity> get( long id );
}

请注意，默认情况下 Single 和 Observer 都在后台调度程序上运行，即没有必要显式调用 subscribeOn。

这是从网络服务器获取实体并在成功时将实体插入数据库的方法。

private Completable fetchAndUpdateEntity( long id ) {
  return retrofitClient
    .getEntity( id )
    .flatMap( new Function<Entity,SingleSource<Long>>() {
      public SingleSource<Long> apply( @NonNull Entity entity ) {
        return appDatabase.entityDao().insert( entity );
      }
    } )
    .ignoreElement();
}

这是我未能实现的方法。 该方法从数据库中获取 Observable<Entity> 并将其转换为可观察的 Resource<Entity>。 当实体过期时，会调用并订阅 fetchAndUpdateEntity()。 当 fetchAndUpdateEntity() 成功将更新的实体插入到数据库中时，数据库返回的 Observable<Entity> 重新发出新条目。

public Observable<Resource<Entity>> getEntity( long id ) {
  return appDatabase.entityDao()
    .getEntity( id )
    .map( new Function<Entity,Resource<Entity>>() {
      public Resource<Entity> apply( @NonNull Entity entity ) {
        boolean final isOutdated = entity.lastSuccessfulSync < System.currentTimeMillis() - refreshInterval );
        if( isOutdated ) {
          fetchAndUpdateEntity( id ).subscribe();
          return Resource.updating( entity );
        }
        return Resource.upToDate( entity );
      }
    } );
}

问题是双重的：

1. 代码打破了观察者链，即未使用订阅 Disposable 的 fetchAndUpdateEntity( id )
2. 代码不使用 FAILED 的 Resource 状态（见下文），特别是方法返回的 Observable 不会传播失败的提取

这就是我想要实现的：

1. 以某种方式将 fetchAndUpdateEntity( id ) 放入链中，但仅在确实需要时才调用它，如上所示。
2. 如果 fetchAndUpdateEntity( id ) 返回的 Completable 成功完成，则不会发生任何特殊情况，因为 DB 将通过 DB 观察者重新发出更新后的实体。
3. 但是，如果 Completable 因失败而终止，则之前发出的 DB observable 实体应与 Resource 状态以及 {{1} 的消息一起放入新的 FAILED } 来自失败的可完成项。
4. 从 DB observable 发出的实体应立即发送到下游，即该方法返回的 observable 不应等待从服务器后端成功获取

我目前考虑的：

1. 要求 1 和 4 一起听起来像是 Throwable 转换的用例。合并的第一个来源显然是 DB observable。合并的第二个来源提出了问题。我不能使用 DB completable（并且以某种方式“神奇地”转换为正确类型的 observable），因为 a）只有在第一个发出的值已过时时才执行 DB completable 并且 b）如果 DB completable 失败我需要访问先前发出的值才能在失败时重新发出它
2. 我考虑了 merge 和 `switchMap'，因为它们都将 DB observable 的结果作为参数传递。但是，它们需要一个 observable 作为返回值，而我没有。

也许有人可以想出一个解决方案？这似乎是一个整天的问题。

---

附录：flatMap 类

Resource