RxAndroid - 使用 Zip 运算符处理错误

Question

我正在尝试找到一种方法来并行执行请求并在每个可观察对象完成时处理它们。尽管当所有可观察对象都给出响应时一切正常，但我没有看到在一切完成后处理所有错误的方法。

这是一个 zip 运算符的示例，它基本上并行执行 2 个请求：

Observable.zip(
                getObservable1()
                        .onErrorResumeNext { errorThrowable: Throwable ->
                            Observable.error(ErrorEntity(Type.ONE, errorThrowable))
                        }.subscribeOn(Schedulers.io()),
                getObservable2()
                        .onErrorResumeNext { errorThrowable: Throwable ->
                            Observable.error(ErrorEntity(Type.TWO, errorThrowable))
                        }.subscribeOn(Schedulers.io()),
                BiFunction { value1: String, value2: String ->
                    return@BiFunction value1 + value2
                })
                //execute requests should be on io() thread
                .subscribeOn(Schedulers.io())
                //there are other tasks inside subscriber that need io() thread
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        { result ->
                            Snackbar.make(view, "Replace with your own action " + result, Snackbar.LENGTH_LONG)
                                    .setAction("Action", null).show()
                        },
                        { error ->
                            Log.d("TAG", "Error is : " + (error as ErrorEntity).error.message)
                        }
                )


private fun getObservable1(): Observable<String> {
    return Observable.defer {
        throw Throwable("Error 1")
    }
}

private fun getObservable2(): Observable<String> {
    return Observable.defer {
        throw Throwable("Error 2")
    }
}

这种方法的问题在于，没有像 BiFunction 那样加入每个错误的机制来处理成功案例。因此，zip 运算符只会触发第一个错误，而忽略其他错误。

输出：

D/TAG: Error is : Error 1

是否有任何方法可以仅在 zip 中的每个可观察对象完成或出错后检索所有错误？

我的主要目标是查看哪些请求出错并仅在用户出现对话框询问他是否要重试失败的请求后才执行那些请求。

Answer 1

您可以使用数据类为您的可观察对象建模。例如

sealed class Response {
    data class Success(val data: String) : Response()
    data class Error(val t: Throwable) : Response()
}

然后您可以像这样将您的可观察对象映射到 Response：

val first: Observable<Response> = observable1
        .map<Response> { Response.Success(it) }
        .onErrorReturn { Response.Error(it) }

val second: Observable<Response> = observable2
        .map<Response> { Response.Success(it) }
        .onErrorReturn { Response.Error(it) }

你可以组合它们：

Observable.zip(
        first,
        second,
        BiFunction { t1: Response, t2: Response -> Pair(t1, t2) }
).subscribe({println(it)})

这会打印：

(Error(t=java.lang.Exception: Error 1), Error(t=java.lang.Exception: Error 2))

也看看this article。

RxAndroid - 使用 Zip 运算符处理错误

RxAndroid - Handle Errors with Zip operator

error-handling

kotlin

rx-java

rx-android

rx-java2