在 RxJava2 中捕获错误并返回新类型答案

【问题标题】：Catch an error and return a new type in RxJava2在 RxJava2 中捕获错误并返回新类型
【发布时间】：2018-06-16 23:54:31
【问题描述】：

我有一个流：

val symbols: Single<List<Symbol>>

现在我想用 map() 将流转换为 UI 状态：

private fun cool(): Single<SymbolContract.State> =
    symbols.map { SymbolContract.State.Symbols(it) }

我想做的是在上游symbols单上捕获一个错误，这样我就可以捕获任何错误然后返回SymbolContract.State.GeneralError()。

我想要onErrorMap() 之类的东西。不幸的是，将onErrorResumeItem 放在symbols 上不起作用，因为它需要返回List<Symbol>。

我可以想到一些丑陋的方法来做到这一点，但最干净的方法是什么？

【问题讨论】：

我猜你的意思是onErrorReturnItem，因为onErrorResumeItem 不存在。无论如何，它为什么会返回一个符号列表？ onErrorReturnItem 将返回 SymbolContract.State。
它必须返回流返回的相同类型
我什么也没得到。如果你用 map 改变流，当前流必须返回一个SymbolContract.State，这就是我用onErrorReturnItem 指出的。至少，对于您发布的答案，它似乎对您有用，尽管显式类型不应该是必需的，因为 Kotlin 可以推断它。

标签： java kotlin rx-java rx-java2 reactivex

【解决方案1】：

我建议你使用全局处理错误。我给你一个样本，这样你就可以理解了。（它是 kotlin），您可以捕获任意数量的异常，其中一些是我的自定义异常。请记住，这个示例是关于 Reactive Webflux 的，但你明白了。其他人也会类似

@Configuration
class ExceptionTranslator {

    @Bean
    @Order(-1)
    fun handle(objectMapper: ObjectMapper): ErrorWebExceptionHandler {
        return ErrorWebExceptionHandler { exchange, ex ->
            if (exchange.response.isCommitted) {
                return@ErrorWebExceptionHandler Mono.error(ex)
            }

            val response = exchange.response
            response.statusCode = HttpStatus.INTERNAL_SERVER_ERROR
            response.headers.contentType = MediaType.APPLICATION_PROBLEM_JSON_UTF8

            val url: String

            var message = ex.message
            var params = mutableMapOf<String, Any>()

            when (ex) {
                is ParametricException -> {
                    url = ex.url
                    params = ex.params
                }
                is BaseException -> {
                    url = ex.url
                }
                is BadCredentialsException -> {
                    url = INVALID_CREDENTIAL_TYPE
                    message = ex.message ?: "Wrong Credentials"
                }
                is ConcurrencyFailureException -> {
                    url = INTERNAL_TYPE
                    message = ERR_CONCURRENCY_FAILURE
                }
                is MethodArgumentNotValidException -> {
                    val result = ex.bindingResult
                    val fieldErrors = result.fieldErrors.map {
                        FieldErrorVM(it.objectName, it.field, it.code ?: "Unknown")
                    }

                    url = CONSTRAINT_VIOLATION_TYPE
                    message = ERR_VALIDATION
                    params = Collections.singletonMap("errors", fieldErrors)
                }
                else -> url = INTERNAL_TYPE
            }

            if (ex is BaseException) {
                response.statusCode = HttpStatus.valueOf(ex.status.code())
            }

            val bytes = objectMapper.writeValueAsBytes(ProblemVM(url, message ?: "Internal Error", params))
            val buffer = response.bufferFactory().wrap(bytes)
            response.writeWith(Mono.just(buffer))
        }
    }

}

【讨论】：

【解决方案2】：

找到一个干净的答案：

private fun symbols(): Single<SymbolContract.State> =
       symbols.map<SymbolContract.State> { SymbolContract.State.Symbols(it) }
            .onErrorReturnItem(SymbolContract.State.Error(SymbolContract.State.Error.Type.Network))

onErrorReturnItem 必须在地图之后，地图需要显式类型参数。

【讨论】：