Rx-Kotlin awaitTerminalEvent 永远不会完成

Question

我试图更好地了解如何使用 Rx-Kotlin 进行单元测试，但我无法成功地将主题设置为“已完成”。结果，我总是在等待 5 秒的超时（onComplete 应该是立即的），然后在 assertComplete 上失败。

我对@9​​87654321@ 的理解是它应该只在调用 onComplete 之前阻塞。我也研究了TestScheduler,，但我不认为这里应该需要它。

任何可以引导我走向正确方向的帮助或文档将不胜感激。

@Test
fun testObservable() {
    val subject = BehaviorSubject.create<Int>()
    subject.onNext(0)

    TestSubscriber<Int>().apply {
        subject.subscribe({
            System.out.println(it)
            subject.onNext(1)
            subject.onComplete()
        })

        this.awaitTerminalEvent(5, TimeUnit.SECONDS)
        this.assertComplete()
        this.assertValue(1)
    }
}

Answer 1

你以错误的方式使用了错误的工具......

• TestSubscriber 用于测试Flowable，你应该在这里使用TestObserver。
• 您应该订阅TestObserver（或Flowable 中的TestSubscriber），以便它监控排放并能够等待终端事件和断言值。在您的代码中，TestSubscriber 未附加到任何流，因此它永远不会收到任何事件。

试图模仿你的代码，可能是这样的：

 @Test
fun testObservable() {
    val subject = BehaviorSubject.create<Int>()
    subject.onNext(0)

    TestObserver<Int>().apply {
        subject.doOnNext {
            System.out.println(it)
            subject.onNext(1)
            subject.onComplete()
        }
                .subscribe(this)

        this.awaitTerminalEvent(5, TimeUnit.SECONDS)
        this.assertComplete()
        this.assertValue(1)
    }
}  

如您所见，我使用TestObserver 订阅是通过TestObserver 对象完成的，而主题onNext()、onComplete() 移动到doOnNext()。测试将失败，因为您有两个发出的值，而测试断言仅适用于单个“1”值。

一般来说，你在onNext()中使用subject再次发出然后调用onComplete()，你可以先订阅然后再向外发出，这是错误的。像这样：

TestObserver<Int>().apply {
        subject.subscribe(this)
        subject.onNext(1)
        subject.onComplete()
        ....
}