我有一个使用 PublishSubject
的分页解决方案,如下所示:
private val pages: PublishSubject<Int> = PublishSubject.create()
val observable: Observable<List<Data> = pages.hide()
.filter { !inFlight }
.doOnNext { inFlight = true }
.flatMap{
getPage(it) // Returns an Observable
}
.doOnNext(::onNextPage) // inFlight gets reset here
此 Observable
与其他 Observable
合并并扫描,如下所示:
fun stateObservable(): Observable<SavedState> {
return Observable.merge(listOf(firstPage(),
nextPage(),// The observable listed above
refresh()))
.scan(MyState.initialState(), StateReducer::reduce)
}
基本上我有一个单向设置,其中每个可观察对象都在累加器函数 reduce
的帮助下更新 MyState
及其相关更改。
在 ViewModel
中,这是以一种直接的方式使用的:
interactor.stateObservable()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeBy(onNext = ::render, onError = Timber::e)
.addTo(subscriptions)
此设置适用于 firstPage
以及 refresh
(也在 PublishSubject
的帮助下触发)但由于某些原因分页解决方案get 就像在 flatMap
中返回 getPage
Observable
但是此页面 Observable
永远不会被触发/订阅flatMap
之后的 doOnNext
显然也没有被调用。它似乎基本上不想订阅它,我只是不知道为什么。
getPage
函数如下所示:
private fun getPage(page: Long): Observable<PartialState<SavedState>> {
return repo.getPage(page).firstOrError().toObservable()
.subscribeOn(Schedulers.io())
.map<PartialState<MyState>> { NextPageLoaded(it) }
.onErrorReturn { NextPageError(it) }
.startWith { NextPageLoading() }
}
repo 中的 getPage
在 RxJavaInterop
通过以下方式:
public io.reactivex.Observable<List<Data>> getPage(long page) {
Observable<List<Data>> observable = getPage(page)
.map(dataList -> {
if(dataList == null){
dataList = new ArrayList<>();
}
return dataList;
});
return RxJavaInterop.toV2Observable(observable);
}
我没有收到任何错误,因此您可以排除这种情况。
我已经有了与 RxJava 1 完全相同的设置,它工作得很好,现在当我迁移到 2.x 时,我期待相同的解决方案可以工作,但我完全陷入了这个分页问题,并且在设置按预期工作的所有其他场景。
为了能够测试这个问题,我在 GitHub 上上传了一个示例项目演示问题。
有哪位 RxJava 专家知道它可能是什么? :)
谢谢
最佳答案
我发现了一个问题:过度使用 {}
为 startWith
创建一个 lambda,它什么都不做(因此永远不会切换到页面链) nextPageObservable
。
.startWith { NextPageLoading() }
相反:
.startWith ( NextPageLoading() )
但是,通过此更改,您的代码会由于其他地方的空值而产生 IAE:
java.lang.IllegalArgumentException: Parameter specified as non-null is null: method kotlin.jvm.internal.Intrinsics.checkParameterIsNotNull, parameter page
at com.paginationissue.ui.list.PaginationIssueInteractor$1.invoke(Unknown Source:2)
at com.paginationissue.ui.list.PaginationIssueInteractor$1.invoke(PaginationIssueInteractor.kt:15)
at com.paginationissue.paging.PageNumberTokenStrategy.generateNextPageToken(PageNumberTokenStrategy.kt:21)
at com.paginationissue.paging.PageNumberTokenStrategy.generateNextPageToken(PageNumberTokenStrategy.kt:12)
at com.paginationissue.paging.Pager.onNextPage(Pager.kt:92)
关于android - RxJava 2.x - 由 PublishSubject 触发并与其他 Observable 合并的 Observable 获取 flatMap 不会被订阅/执行,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58273141/