如何对 RxJava retryWhen 进行单元测试

Question

我正在尝试对 RxJava Completable 使用指数退避，并且在我尝试时代码运行良好。所以试图添加一个单元测试来覆盖它，但不确定我们如何在这里测试retryWhen 部分。

    fun makeCall(
        scheduler: Scheduler = scheduler.io()
    ) : Completable {
        val value = "myString"

        return myCompletable(value)
            .retryWhen { errors: Flowable<Throwable> ->
                errors
                    .zipWith(Flowable.range(1, 3))
                    .flatMap { errorRetryPair: Pair<Throwable, Int> ->
                        Flowable.timer(2.toDouble().pow(errorRetryPair.second.toDouble()).toLong(), TimeUnit.SECONDS)
                    }
            }.subscribeOn(scheduler)
    }

有人知道我们如何测试吗？ （无法找到很多关于测试的信息以确保 retryWhen 有效）

TIA

Answer 1

单元测试 RxJava 定时操作最好使用 TestScheduler。

// in your test code
simulateFailure();
long firstStepMs = 2_000; // first backoff time in ms
long epsilon = 40;        // 10ms is too small
scheduler.advanceTimeBy( firstStepMs - epsilon, TimeUnit.MILLISECONDS );
// test to see that resubscription has NOT occurred
scheduler.advanceTimeBy( 2 * epsilon, TimeUnit.MILLISECONDS );
// test to see that first resubscription HAS occurred

并重复不同的退避时间。推进时钟的粒度取决于您。 epsilon 必须大于 10 毫秒，因为 Java 中的定时器调度是如何工作的；你肯定无法有效地测量任何更短的间隔。