平面映射的可流动错误取消了Observable

Question

我有以下RxJava2链：

fun refreshList() {
        refreshDisposable?.dispose()
        refreshDisposable =
                mapHandler.filtersFlowable()
                        .doOnNext{
                            Timber.w("LISTFILTER - New filter! $it")
                        }
                        .switchMap {
                            lastPage = 0
                            refreshByPage(lastPage, it).toFlowable()
                        }
                        .doOnSubscribe {
                            Timber.w("LISTFILTER - In progress")
                            listRefreshState.postValue(Result.inProgress())
                        }
                        .map {
                            Result.success(it)
                        }
                        .ioUi(schedulerProvider)
                        .subscribeBy(
                                onNext = {
                                    Timber.w("LISTFILTER - Success!")
                                    listRefreshState.value = it
                                },
                                onError = {
                                    Timber.w("LISTFILTER - Error! ${it.message}")
                                    listRefreshState.value = Result.failure(it)
                                }
                        )
                        .addTo(disposables)

    }

也就是说，我的应用程序中有一组过滤器，当您更改它们时，应该针对后端API抛出新请求。

现在我发现有时请求会失败，然后我的filtersFlowable会停止发出。我希望它在再次更换过滤器后继续发光。

这是FiltersFlowable的定义：

private val filterSubject: BehaviorSubject<ApplyFilterRequest> =
        BehaviorSubject.createDefault(
                if (sharedPrefsManager.getFilters().isNotEmpty())
                    ApplyFilterRequest.fromJson(moshi, sharedPrefsManager.getFilters())
                else
                    ApplyFilterRequest.allPlacesAllTypes())

private val filterFlowable = filterSubject.hide().toFlowable(BackpressureStrategy.LATEST)

我怎样才能达到理想的行为？

Answer 1

你可以使用onErrorResumeNext：

fun provideFlowable(filter: Boolean): Flowable<Boolean> {
    val filterSubject: BehaviorSubject<Boolean> = BehaviorSubject.createDefault(filter)
    return filterSubject.hide().toFlowable(BackpressureStrategy.LATEST)
}

fun refreshList() {
    provideFlowable(true)
        .doOnNext { println("Before switch map") }
        .switchMap { apiCall(it) }
        .doOnNext { println("After switch map") }
        .subscribe(
            { println("Success") },
            { println("Error") }
        )
}

// will throw an error if true is passed
fun apiCall(filter: Boolean): Flowable<Boolean> {
    return Flowable.just(filter)
        .map { if (filter) throw Exception() else it }
        .onErrorResumeNext { error: Throwable ->
            println("Error happens")
            Flowable.just(false)
        }
}

refreshList()