我有一个 Flowable<T> 和我想忽略的 IgnoreThisError 子类型的错误（导致正常完成），所有其他错误都应该向下游传播。
Kotlin 示例：
val f : Flowable<T> = ...
val g = f.onErrorComplete { it is IgnoreThisError }
onErrorComplete这个函数是我需要的，不存在的……
【问题讨论】：
kotlin f.onErrorComplete { when (it) { is IgnoreThisError -> ... else -> throw it } }
Flowable 没有onErrorComplete。
标签： kotlin rx-java2 rx-kotlin2
这样的东西应该可以工作
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)
})
【讨论】：
onErrorComplete扩展函数。
这里是缺少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 的重载而报错。
【讨论】：
it 以外的其他名称，而不是压制，例如error: Throwable -> ...。 @NonNull 严格来说也是多余的，因为 Kotlin 会添加 @org.jetbrains.annotations.NotNull 注释。