java - 从 Rxjava2 中的 Single.onErrorResumeNext 启动可观察链

Question

loadSingle返回单个对象，如果失败我想调用getObservable(rsList)哪个返回 Observable。我正在尝试，onErrorResumeNext但它需要单个对象。我怎样才能调用getObservable(rsList)失败loadSingle()？

提前致谢！！

                    repo.loadSingle()
                            .subscribeOn(Schedulers.io())
                            
                            .onErrorResumeNext {
                                repo.getObservable(rsList)
                            }
                            .flatMapObservable {
                                if (it != null && it.status == Status.SUCCESS) {
                                    upRsList(it.data)
                                }
                                repo.getObservable(rsList)
                            }
                          
                            ({ //observable success
                             }, {
                               //observable error 
                            })

API接口

interface HomeApi{
    fun getSingel():Single<List<String>>
    fun getObservable():Observable<HomeResponse>
}

标签： javaandroidrx-javareactive-programmingrx-java2

Answer 1

依赖项

testImplementation("org.junit.jupiter:junit-jupiter-api:5.6.2")
testRuntimeOnly("org.junit.platform:junit-platform-launcher:1.6.2")
testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.6.2")
testRuntimeOnly("org.junit.vintage:junit-vintage-engine:5.6.2")
implementation "io.reactivex.rxjava3:rxjava:3.0.4"
implementation "io.reactivex.rxjava3:rxkotlin:3.0.0"

必修课

internal interface Repo {
    fun loadSingle(): Single<Result<List<String>>>

    fun getObservable(list: List<String>): Observable<String>
}

internal class RepoImpl : Repo {
    override fun loadSingle(): Single<Result<List<String>>> {
        return Single.error(RuntimeException("fail"))
    }

    override fun getObservable(list: List<String>): Observable<String> {
        if (list === emptyList<String>()) {
            return Observable.just("42")
        }
        return Observable.just("success")
    }
}

internal sealed class Result<T> {
    data class Success<T>(val value: T) : Result<T>()

    data class Failure<T>(private val failure: Throwable) : Result<T>()
}

测试

通过#onErrorReturn 将错误包装为默认值，并相应地处理结果。

class So64751341 {
    @Test
    fun `64751341`() {
        val repo: Repo = RepoImpl()
        val testScheduler = TestScheduler()

        val flatMapObservable = repo.loadSingle()
            .subscribeOn(testScheduler)
            .onErrorReturn { failure -> Result.Failure(failure) }
            .flatMapObservable { result ->
                when (result) {
                    is Result.Success -> repo.getObservable(result.value)
                    is Result.Failure -> repo.getObservable(emptyList())
                }
            }


        val test = flatMapObservable.test()

        testScheduler.triggerActions()

        test // return default value 42 onError
            .assertValue("42")
    }
}

Repo#loadSingle() 同步抛出异常

internal class RepoExceptionImpl : Repo {
    override fun loadSingle(): Single<Result<List<String>>> {
        throw java.lang.RuntimeException("whatever")
    }

    override fun getObservable(list: List<String>): Observable<String> {
        if (list === emptyList<String>()) {
            return Observable.just("42")
        }
        return Observable.just("success")
    }
}

测试

Repo#loadSingle 必须用 Single#defer 包装。Single#defer 将捕获异常并将其作为 #onError 发送给订阅者，然后由 #onErrorReturn 处理

@Test
fun `64751341_exception`() {
    val repo: Repo = RepoExceptionImpl()
    val testScheduler = TestScheduler()

    val flatMapObservable = Single.defer {
        repo.loadSingle()
    }
        .subscribeOn(testScheduler)
        .onErrorReturn { failure -> Result.Failure(failure) }
        .flatMapObservable { result ->
            when (result) {
                is Result.Success -> repo.getObservable(result.value)
                is Result.Failure -> repo.getObservable(emptyList())
            }
        }

    val test = flatMapObservable.test()

    testScheduler.triggerActions()

    test // return default value 42 onError
        .assertValue("42")
}