通常如何测试非 RxJava 可观察对象或异步代码？

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.

或者如果你使用 Observables，你可以使用 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