RxJava：忽略 Flowable 中的一些错误

Question

我有一个 Flowable<T> 和属于 IgnoreThisError 子类型的错误我想忽略（导致正常完成），所有其他错误应该传播到下游。

Kotlin 示例：

val f : Flowable<T> = ...
val g = f.onErrorComplete { it is IgnoreThisError }

onErrorComplete 这个函数是我需要的，不存在...

Answer 1

像这样的东西应该可以工作

f.onErrorResumeNext {
    if (it is IgnoreThisError)
        Flowable.empty()
    else
        Flowable.error(it)
}

由于重载和 Java 对签名方差的模拟，我不确定 Kotlin 是否可以在此处正确推断类型，如果不能，您可能需要更明确：

f.onErrorResumeNext(Function<Throwable, Publisher<T>> {
    if (it is IgnoreThisError)
        Flowable.empty()
    else
        Flowable.error(it)
})

Answer 2

下面是缺少onErrorComplete作为扩展函数的解决方法：

/**
 * Errors encountered in the stream for which the provided `predicate` returns true will be silently turned into graceful completion.
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
inline fun <T> Flowable<T>.onErrorComplete(crossinline predicate: (Throwable) -> Boolean): Flowable<T> =
  onErrorResumeNext { error: Throwable ->
    if (predicate(error)) Flowable.empty<T>() else Flowable.error<T>(error)
  }

请注意，我必须显式放置 error: Throwable ->，否则编译器会因为 onErrorResumeNext.

的重载而报错

RxJava：忽略 Flowable 中的一些错误

RxJava: Ignore some errors in a Flowable

kotlin

rx-java2

rx-kotlin2