Rx-Kotlin awaitTerminalEvent永远不会完成

Question

我想更好地了解如何使用Rx-Kotlin进行单元测试，但我无法成功设置主题为“已完成”。因此，我总是等待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（或TestSubscriberFlowable），以便它监测排放并能够等待终端事件和断言值。在您的代码中，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()，然后打电话onComplete()，你可以订阅之前，然后发射到外面。这样的事情：

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