使用 RxJava 实现通用 Observable 的 TestScheduler

Question

我开始使用TestScheduler。像这样的东西一切正常：

@Test
public void test1() throws Exception {
    //when
    TestScheduler scheduler = new TestScheduler();
    TestObserver<Long> subscriber = new TestObserver<>();

    //run
    Observable
            .interval(1L, TimeUnit.SECONDS, scheduler)
            .subscribeWith(subscriber);

    //check
    scheduler.advanceTimeBy(200, TimeUnit.SECONDS);
    assertEquals(200, subscriber.valueCount());
}

如果我改用这个：

@Test
public void test2() throws Exception {
    //when
    TestScheduler scheduler = new TestScheduler();
    TestObserver<Long> subscriber = new TestObserver<>();

    //run
    Observable
            .interval(1L, TimeUnit.SECONDS)
            .observeOn(scheduler)
            .subscribeOn(scheduler)
            .subscribeWith(subscriber);

    //check
    scheduler.advanceTimeBy(200, TimeUnit.SECONDS);
    assertEquals(200, subscriber.valueCount());
}

测试失败，因为订阅者从未被调用。

我找到的所有示例，使用TestScheduler 和Observable.interval 并将调度程序传递给工厂方法，就像我在第一个示例中所做的那样。 我不能使用这种方法的原因是，在真正的应用程序中，observables 并不像这个那么简单，我无法通过调度程序。 我认为像在第二个示例中那样设置Scheduler 会很好，但看起来不是。

将 TestScheduler 用于更通用的 Observable 的正确方法是什么？

不使用 TestScheduler 我可以成功使用这些方法：

@Test
public void test3() throws Exception {
    //when
    Scheduler trampoline = Schedulers.trampoline();

    //run
    TestObserver<Long> test = Observable
            .interval(1L, TimeUnit.SECONDS)
            .observeOn(trampoline)
            .subscribeOn(trampoline)
            .test();

    //check
    test.await(3100,TimeUnit.MILLISECONDS);
    assertEquals(3, test.valueCount());
}

@Test
public void test4() throws Exception {
    //when
    Scheduler trampoline = Schedulers.trampoline();

    //run
    TestObserver<Long> test = Observable
            .fromArray(1L, 2L, 3L)
            .subscribeOn(trampoline)
            .observeOn(trampoline)
            .test();

    //check
    assertEquals(3, test.valueCount());
}

@Test
public void test5() throws Exception {
    //when
    Scheduler trampoline = Schedulers.trampoline();

    //run
    TestObserver<Long> test = Observable
            .fromArray(1L, 2L, 3L)
            .subscribeOn(trampoline)
            .observeOn(AndroidSchedulers.mainThread())
            .test();

    //check
    test.awaitTerminalEvent();
    assertEquals(3, test.valueCount());
}

编辑

如果我使用没有区别

@BeforeClass
public static void setupClass() {
    mScheduler = new TestScheduler();

    RxAndroidPlugins.setInitMainThreadSchedulerHandler(__ -> mScheduler);
    RxJavaPlugins.setIoSchedulerHandler(__ -> mScheduler);
}

@Test
public void test2() throws Exception {
    //when
    TestObserver<Long> subscriber = new TestObserver<>();

    //run
    Observable
            .interval(1L, TimeUnit.SECONDS)
            .observeOn(mScheduler)
            .subscribeOn(mScheduler)
            .subscribeWith(subscriber);

    //check
    mScheduler.advanceTimeBy(200, TimeUnit.SECONDS);
    assertEquals(200, subscriber.valueCount());
}

Answer 1

如果您想覆盖您可以使用的标准调度程序（用于间隔）。也可以覆盖其他标准调度程序。

RxJavaPlugins.setComputationSchedulerHandler(scheduler -> testScheduler);

testScheduler 将是您的 TestScheduler。设置好插件后，可以像test1

那样使用advanceTime

例子：

@Test
// fails because interval schedules on different thread then JUnit-Runner-thread -> fall through
void notWorkingTest1() throws Exception {
    TestScheduler scheduler = new TestScheduler();
    TestObserver<Long> subscriber = new TestObserver<>();

    Observable
            .interval(1L, TimeUnit.SECONDS)
            .observeOn(scheduler)
            .subscribeOn(scheduler)
            .subscribeWith(subscriber);

    //check
    scheduler.advanceTimeBy(200, TimeUnit.SECONDS);
    assertEquals(200, subscriber.valueCount());
}


@Test
// not working because interval will not be scheduled on virtual time -> JUnit-Runner-Thread will close because test observable emits on different thread
void notWorkingTest2() throws Exception {
    //when
    TestScheduler scheduler = new TestScheduler();

    //run
    TestObserver<Long> test = Observable
            .interval(1L, TimeUnit.SECONDS)
            .observeOn(scheduler)
            .subscribeOn(scheduler)
            .test();

    scheduler.advanceTimeBy(200, TimeUnit.SECONDS);

    test.assertValueCount(200);
}

@Test
// runs sync. -> no JUnit-Runner-thread blocking needed
void workingTest() throws Exception {
    TestScheduler scheduler = new TestScheduler();
    RxJavaPlugins.setComputationSchedulerHandler(s -> scheduler);

    TestObserver<Long> test = Observable
            .interval(1L, TimeUnit.SECONDS) // executed on Schedulers.computation()
            .observeOn(scheduler)
            .subscribeOn(scheduler)
            .test();

    scheduler.advanceTimeBy(200, TimeUnit.SECONDS);

    test.assertValueCount(200);
}