21

我正在尝试测试以下 RxKotlin/RxJava 2 代码：

validate(data)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap { ... }

我正在尝试按如下方式覆盖调度程序：

// Runs before each test suite
RxJavaPlugins.setInitIoSchedulerHandler { Schedulers.trampoline() }
RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }

但是，运行测试时出现以下错误：

java.lang.ExceptionInInitializerError
...
Caused by: java.lang.NullPointerException: Scheduler Callable result can't be null
    at io.reactivex.internal.functions.ObjectHelper.requireNonNull(ObjectHelper.java:39)
    at io.reactivex.plugins.RxJavaPlugins.applyRequireNonNull(RxJavaPlugins.java:1317)
    at io.reactivex.plugins.RxJavaPlugins.initIoScheduler(RxJavaPlugins.java:306)
    at io.reactivex.schedulers.Schedulers.<clinit>(Schedulers.java:84)

有没有人遇到过这个问题？

使用 RxKotlin/RxJava 1 和以下调度程序覆盖时，测试工作正常：

RxAndroidPlugins.getInstance().registerSchedulersHook(object : RxAndroidSchedulersHook() {
    override fun getMainThreadScheduler() = Schedulers.immediate()
})

RxJavaPlugins.getInstance().registerSchedulersHook(object : RxJavaSchedulersHook() {
    override fun getIOScheduler() = Schedulers.immediate()
})
4

4 回答 4

24

我建议您采用不同的方法并为您的调度程序添加一个抽象层。这个人有一篇关于它的好文章。

在 Kotlin 中看起来像这样

interface SchedulerProvider {
    fun ui(): Scheduler
    fun computation(): Scheduler
    fun trampoline(): Scheduler
    fun newThread(): Scheduler
    fun io(): Scheduler 
}

然后你用你自己的 SchedulerProvider 实现覆盖它：

class AppSchedulerProvider : SchedulerProvider {
    override fun ui(): Scheduler {
        return AndroidSchedulers.mainThread()
    }

    override fun computation(): Scheduler {
        return Schedulers.computation()
    }

    override fun trampoline(): Scheduler {
        return Schedulers.trampoline()
    }

    override fun newThread(): Scheduler {
        return Schedulers.newThread()
    }

    override fun io(): Scheduler {
        return Schedulers.io()
    }
}

还有一个用于测试类：

class TestSchedulerProvider : SchedulerProvider {
    override fun ui(): Scheduler {
        return Schedulers.trampoline()
    }

    override fun computation(): Scheduler {
        return Schedulers.trampoline()
    }

    override fun trampoline(): Scheduler {
        return Schedulers.trampoline()
    }

    override fun newThread(): Scheduler {
        return Schedulers.trampoline()
    }

    override fun io(): Scheduler {
        return Schedulers.trampoline()
    }
}

在调用 RxJava 时，您的代码将如下所示：

mCompositeDisposable.add(mDataManager.getQuote()
        .subscribeOn(mSchedulerProvider.io())
        .observeOn(mSchedulerProvider.ui())
        .subscribe(Consumer<Quote> {
...

而且您将SchedulerProvider根据您的测试位置覆盖您的实现。这是一个示例项目供参考，我正在链接将使用可测试版本的测试文件SchedulerProvider：https ://github.com/Obaied/DingerQuotes/blob/master/app/src/test/java/com/obaied/ dingerquotes/QuotePresenterTest.kt#L31

于 2017-04-10T10:30:48.533 回答
14

弄清楚了！这与在此代码中的事实有关：

validate(data)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap { ... }

validate(data)正在返回一个Observable，它发出以下内容：emitter.onNext(null). 由于 RxJava 2 不再接受null值，flatMap因此没有被调用。我更改validate为返回 aCompletable并将调度程序覆盖更新为以下内容：

RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }

现在测试通过了！

于 2017-04-07T19:19:37.337 回答
6

作为提议的解决方案的替代方案，这在我的项目中已经运行了一段时间。您可以像这样在测试类中使用它：

@get:Rule
val immediateSchedulersRule = ImmediateSchedulersRule()

这个类看起来像这样：

class ImmediateSchedulersRule : ExternalResource() {

    val immediateScheduler: Scheduler = object : Scheduler() {

        override fun createWorker() = ExecutorScheduler.ExecutorWorker(Executor { it.run() })

        // This prevents errors when scheduling a delay
        override fun scheduleDirect(run: Runnable, delay: Long, unit: TimeUnit): Disposable {
            return super.scheduleDirect(run, 0, unit)
        }

    }

    override fun before() {
        RxJavaPlugins.setIoSchedulerHandler { immediateScheduler }
        RxJavaPlugins.setComputationSchedulerHandler { immediateScheduler }
        RxJavaPlugins.setNewThreadSchedulerHandler { immediateScheduler }

        RxAndroidPlugins.setInitMainThreadSchedulerHandler { immediateScheduler }
        RxAndroidPlugins.setMainThreadSchedulerHandler { immediateScheduler }
    }

    override fun after() {
        RxJavaPlugins.reset()
    }

}

您可以在此处找到从 TestRule 迁移到 ExternalResource 的方法，并在此处获取有关测试 RxJava 2 的更多信息。

于 2019-06-28T15:11:17.897 回答
5

这是对我有用的确切语法：

RxJavaPlugins.setIoSchedulerHandler(scheduler -> Schedulers.trampoline())
于 2018-11-27T12:28:19.740 回答