kotlin - RxJava：忽略 Flowable 中的一些错误

Question

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

科特林示例：

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.