RxKotlin (RxJava2) timeout() 不会抛出 TimeoutException答案

【问题标题】：RxKotlin (RxJava2) timeout() doesn't throw TimeoutExceptionRxKotlin (RxJava2) timeout() 不会抛出 TimeoutException
【发布时间】：2017-06-19 21:44:56
【问题描述】：

我试图获得一个使用两个不同超时值的示例。第一次排放的初始值较大，然后所有后续排放的值较小。该示例从 Java for RxJava v1x 转换为 Kotlin，尽管我正在尝试这是 v2x（不确定这是否有任何区别）。

问题是第一个事件的超时不会引发TimeoutException。将值设置为低于 500 毫秒，我希望打印堆栈跟踪，但我得到的输出好像没有发生超时（超时设置为 40 毫秒的后续发射会按预期产生堆栈跟踪）。以下示例阻止初始超时成功的原因是什么？

fun nextSolarEclipse(after: LocalDate): Observable<LocalDate> {
    return Observable
            .just(
                    LocalDate.of(2016, Month.MARCH, 9),
                    LocalDate.of(2016, Month.SEPTEMBER, 1),
                    LocalDate.of(2017, Month.FEBRUARY, 26),
                    LocalDate.of(2017, Month.AUGUST, 21),
                    LocalDate.of(2018, Month.FEBRUARY, 15),
                    LocalDate.of(2018, Month.JULY, 13),
                    LocalDate.of(2018, Month.AUGUST, 11),
                    LocalDate.of(2019, Month.JANUARY, 6),
                    LocalDate.of(2019, Month.JULY, 2),
                    LocalDate.of(2019, Month.DECEMBER, 26)
            )
            .skipWhile { date ->
                !date.isAfter(after)
            }
            .zipWith(
                    Observable.interval(500, 50, TimeUnit.MILLISECONDS),
                    { date, _ -> date }
            )
}

fun main(args: Array<String>) {
    nextSolarEclipse(LocalDate.now())
            .timeout<Long, Long>(
                    { Observable.timer(400, TimeUnit.MILLISECONDS) },
                    { Observable.timer(40, TimeUnit.MILLISECONDS) }
            )
            .subscribe(
                    { println(it) },
                    { it.printStackTrace() },
                    { println("Completed") }
            )

    TimeUnit.MILLISECONDS.sleep(2000)
}

编辑：2017 年 6 月 20 日

使用 Kotlin 1.1.2-5，使用 IntelliJ，并应用建议的更改，我仍然得到错误。正如我所期望的那样，无论如何尝试运行代码都会产生：

Error:(34, 21) Kotlin: Interface Function does not have constructors

【问题讨论】：

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

【解决方案1】：

这是 Kotlin 的另一种情况，暗示您需要一个函数而不是 Observable 实例。试试这个：

.timeout<Long, Long>(
     Observable.timer(400, TimeUnit.MILLISECONDS),
     Function { Observable.timer(40, TimeUnit.MILLISECONDS) }
)

【讨论】：

如果我这样做，我会得到一个接口 Function 没有构造函数，我不认为我想将它实现为匿名对象。
我没有使用 Kotlin 的经验，如果没有 Function 前缀，代码将无法为我编译。我猜这是臭名昭著的SAM问题。也许你可以改写{ it -> ... }。
如果我尝试除最初发布的以外的任何其他方式，我会收到错误 None of the following functions can be called with the arguments supplied。我想我想要的功能是@CheckReturnValue @SchedulerSupport public final fun <U : Any!, V : Any!> timeout(firstTimeoutIndicator: ObservableSource<Long!>!, itemTimeoutIndicator: Function<in LocalDate!, out ObservableSource<Long!>!>!): Observable<LocalDate!>! defined in io.reactivex.Observable，正如您在最初评论中所建议的那样，但是有/没有Function我仍然得到上述错误。
我在答案中给出的示例适用于 Kotlin 1.1.2。
您可能需要手动import io.reactivex.functions.Function。