RxJavaでbufferオペレーターを使ったユニットテストでハマった

この記事は雑です。

結論

RxJavaPlugins.set***SchedulerHandler に trampoline を設定していたけど、 setIoSchedulerHandler だけにした。

RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }

なぜ

こんな感じのTestRuleを書いていたのだけど、

RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
RxJavaPlugins.setComputationSchedulerHandler { Schedulers.trampoline() }
...

テスト対象のコードに、 buffer オペレーターがあると、bufferオペレーターは内部的に computation スケジューラーを利用しているので、 trampoline スケジューラーに上書きしてしまうことで無限に待ち続けてしまって、テストが終わらなくなる。

@Test
fun testBuffer() {
    val subject = PublishSubject.create<Int>()
    val testObserver = subject
        .buffer(1, TimeUnit.SECONDS)
        .filter { it.isNotEmpty() }
        .test()

    subject.onNext(1)
    subject.onNext(2)

    testObserver.await(2, TimeUnit.SECONDS)
    testObserver.assertValues(listOf(1, 2))
}