问题描述
问题:如何将两个 RxJava observable 合并为一个 observable,以便
更正式的(我不能在这里画大理石图),我想要类似下面的东西。设 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)
背景故事(也许问题是笼统的,对于特定用例有更简单的解决方案)
问题
- 来自网络服务后端的实体应存储在本地数据库中以进行缓存。
- 如果 UI 请求实体,该实体应立即从数据库提供给 UI。
- 应该永远观察数据库,即如果数据库发生更改,则更改应通过可观察对象传播到 UI
- 当且仅当实体看起来已过时,才应从网络服务中获取新版本,插入数据库并重新发送到 UI。
- 实体应被封装到一个“资源”包装器中,该包装器为实际实体配备一个状态以指示实体的“新鲜度”以及可选的错误消息。
- 如果后台更新失败,数据库中实体的最新版本将与失败更新的错误消息一起提供。
- 代码应使用 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 );
}
} );
}
问题是双重的:
- 代码打破了观察者链,即未使用订阅
disposable
的fetchAndUpdateEntity( id )
- 代码不使用
Failed
的Resource
状态(见下文),特别是方法返回的Observable
不会传播失败的提取
这就是我想要实现的:
- 以某种方式将
fetchAndUpdateEntity( id )
放入链中,但仅在确实需要时才调用它,如上所示。 - 如果
fetchAndUpdateEntity( id )
返回的 Completable 成功完成,则不会发生任何特殊情况,因为 DB 将通过 DB 观察者重新发出更新后的实体。 - 但是,如果 Completable 因失败而终止,则之前发出的 DB observable 实体应与
Resource
状态以及 {{1} 的消息一起放入新的Failed
} 来自失败的可完成项。 - 从 DB observable 发出的实体应立即发送到下游,即该方法返回的 observable 不应等待从服务器后端成功获取
我目前考虑的:
- 要求 1 和 4 一起听起来像是
Throwable
转换的用例。合并的第一个来源显然是 DB observable。合并的第二个来源提出了问题。我不能使用 DB completable(并且以某种方式“神奇地”转换为正确类型的 observable),因为 a)只有在第一个发出的值已过时时才执行 DB completable 并且 b)如果 DB completable 失败我需要访问先前发出的值才能在失败时重新发出它 - 我考虑了
merge
和 `switchMap',因为它们都将 DB observable 的结果作为参数传递。但是,它们需要一个 observable 作为返回值,而我没有。
附录:flatMap
类
Resource
解决方法
暂无找到可以解决该程序问题的有效方法,小编努力寻找整理中!
如果你已经找到好的解决方法,欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@)