Rx-Kotlin awaitTerminalEvent永远得不到onComplete

Question

我正试图更好地理解如何使用Rx进行单元测试，但我未能成功地将主题设置为“已完成”。因此，我总是等待5秒的超时( onComplete应该是立即的)，然后在assertComplete上失败。

我对awaitTerminalEvent的理解是，它只应该阻塞直到调用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对象完成的，subject onNext()，onComplete()移动到doOnNext()。测试将失败，因为您有两个发出的值，而测试断言仅针对单个“1”值。

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

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