java - TestScheduler 不适用于 RxJava

Question

我正在尝试测试一个函数，其中流的元素在延迟后一个接一个地分派(dispatch)，我能够使用 Thread.sleep 使我的测试工作。但是，当我使用 TestScheduler.advanceTimeBy 时，我得不到任何结果。

查看代码:

public Observable<Object> getDelayedObjects(Observable<Observable<Object>> objectsStreams) {
    objectsStreams.concatMap(objectsStream ->
        objectsStream.repeat().concatMap(object ->
            Observable.just(object)
                      .delay(getDuration(object), TimeUnit.MILLISECONDS)));
}

和测试代码:

TestScheduler testScheduler = new TestScheduler();
BehaviorSubject<Observable<Object>> objectStreamSubject = BehaviorSubject.create(objectsStream);

model.getDelayedObjects(objectStreamSubject)
        .observeOn(testScheduler)
        .subscribeOn(testScheduler)
        .subscribe(testSubscriber);

testScheduler.triggerActions();
//Thread.sleep(900) works with the default scheduler
testScheduler.advanceTimeBy(900, TimeUnit.MILLISECONDS);
testSubscriber.assertReceivedOnNext(objects);

更新:

检查 TestScheduler 用法我发现通常将调度程序传递给 delay 函数。因此，通过将调度程序作为参数提供给方法 getDelayedObjects 然后再提供给 delay，我能够让测试通过。但是，我仍然不明白为什么它以前不起作用。

Answer 1

默认情况下，delay 运算符使用 computation scheduler用于执行基于时间的延迟。此信息可在 documentation 中找到的方法。在 @SchedulerSupport 注释中查找值，在本例中为 io.reactivex:computation。

出于测试目的，您必须将计算调度程序替换为 TestScheduler .为了能够替换，您必须使用 one of the many overrides包含 Scheduler 的 delay 运算符。