如何在一般情況下測試非RxJava observables或異步代碼？

Question

我在玩着實現我自己的觀察者或從其他語言移植他們爲樂趣和利潤。

我遇到的問題是，關於如何正確測試observables或異步代碼的信息很少。一旦

// Create a stream of values emitted every 100 milliseconds 
// `interval` uses Timer internally 
final Stream<Number> stream = 
    Streams.interval(100).map(number -> number.intValue() * 10); 

ArrayList<Number> expected = new ArrayList<>(); 

expected.add(0); 
expected.add(10); 
expected.add(20); 

IObserver<Number> observer = new IObserver<Number>() { 
    public void next(Number x) { 
    assertEquals(x, expected.get(0)); 
    expected.remove(0); 
    if(expected.size() == 0) { 
     stream.unsubscribe(this); 
    } 
    } 
    public void error(Exception e) {} 
    public void complete() {} 
}; 

stream.subscribe(observer); 

作爲流訂閱了，它發射的第一個值：

考慮以下測試代碼。調用onNext ...然後測試成功退出。

在JavaScript中，現在大多數測試框架都爲測試用例提供了可選的Promise，您可以在成功/失敗時異步調用該測試用例。有什麼類似的Java可用嗎？

Answer 1

由於執行異常，您必須等到完成。您只需等待一段時間在一個古老的時尚之路

your_code 
wait(1000) 
check results. 

或者，如果您使用的觀測量，你可以使用TestSubscriber 在這個例子中，你可以看到有一個異步操作如何，我們等到觀察者消耗的所有項目。

@Test 
public void testObservableAsync() throws InterruptedException { 
    Subscription subscription = Observable.from(numbers) 
      .doOnNext(increaseTotalItemsEmitted()) 
      .subscribeOn(Schedulers.newThread()) 
      .subscribe(number -> System.out.println("Items emitted:" + total)); 
    System.out.println("I finish before the observable finish. Items emitted:" + total); 


    new TestSubscriber((Observer) subscription) 
      .awaitTerminalEvent(100, TimeUnit.MILLISECONDS); 
} 

您可以在這裏看到更多的異步例子https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/scheduler/ObservableAsynchronous.java