CompletableTransformer 未将 subscribeOn 和 observeOn 应用于上游答案

【问题标题】：CompletableTransformer is not applying subscribeOn and observeOn to the UpstreamCompletableTransformer 未将 subscribeOn 和 observeOn 应用于上游
【发布时间】：2019-10-11 20:21:33
【问题描述】：

我正在使用 rxKotlin、MVVM + Clean Architecture 创建离线第一个应用程序作为我的副项目，昨天我决定通过使用转换器摆脱样板 subscribeOn 和 observeOn。我很快意识到转换器的应用功能被忽略了。

这是我的基本可完成用例（交互器）的代码：

abstract class CompletableUseCase(private val transformer: CompletableTransformer) {

    abstract fun createCompletable(data: Map<String, Any>? = null) : Completable

    fun completable(data: Map<String, Any>? = null) : Completable {
        return createCompletable(data).compose(transformer)
    }
}

这里是特定交互器的实现：

class SaveRouteInteractor(
    transformer: CompletableTransformer,
    private val routeRepository: RouteRepository
) : CompletableUseCase(transformer) {

    companion object {
        private const val PARAM_ROUTE = "param_route"
    }

    fun saveRoute(route: Route) : Completable {
        val data = HashMap<String, Route>()
        data[PARAM_ROUTE] = route
        return completable(data)
    }

    override fun createCompletable(data: Map<String, Any>?): Completable {
        val routeEntity = data?.get(PARAM_ROUTE)

        routeEntity?.let {
            return routeRepository.saveRoute(routeEntity as Route)
        } ?: return Completable.error(IllegalArgumentException("Argument @route must be provided."))
    }
}

我的自定义转换器被传递给 SaveRouteInteractor 的构造函数：

class IOCompletableTransformer(private val mainThreadScheduler: Scheduler) : CompletableTransformer {

    override fun apply(upstream: Completable): CompletableSource {
        return upstream.subscribeOn(Schedulers.io()).observeOn(mainThreadScheduler)
    }
}

以及RouteRepository方法的实现：

override fun saveRoute(route: Route): Completable {
        return localRouteSource.saveRoute(route)
            .flatMap { localID ->
                route.routeId = localID
                remoteRouteSource.saveRoute(route)
            }
            .flatMapCompletable { localRouteSource.updateRouteID(route.routeId, it) }
    }

我使用 Room 作为我的本地源，所以在我的 ViewModel 中调用保存交互器后，我收到 IlligalStateException，告诉我不允许访问主线程上的数据库。

也许我遗漏了一些东西，但似乎转换功能被忽略了。我调试了这个方法，它正在将 subscribeOn 和 observeOn 应用到上游。

提前感谢您的帮助，节奏！

【问题讨论】：

标签： android kotlin rx-java2

【解决方案1】：

很难告诉你问题出在哪里，因为代码是不完整的。

例如这里：

    return localRouteSource.saveRoute(route)
        .flatMap { localID ->
            route.routeId = localID
            remoteRouteSource.saveRoute(route)
        }
        .flatMapCompletable { localRouteSource.updateRouteID(route.routeId, it) }

我想localRouteSource.saveRoute() 正在使用您向我们展示的交互器，但不清楚remoteRouteSource.saveRoute() 或localRouteSource.updateRouteID() 是如何实现的。

它们还需要在 IO 线程上订阅。

根据经验，您应该在知道需要时切换线程。

换句话说，你应该在你知道你正在做的 IO 尽可能接近实际工作的地方使用subscribeOn()。当您知道需要在 UI 线程中获得这些结果并且您可能会在其他线程中获得这些结果时，将使用 ObserveOn。

在你的例子中，绝对没有必要继续使用observeOn(MAIN_THREAD)，你唯一需要它的时候（我想）就是你想要显示结果的时候。

其他几件事：

这段代码

override fun createCompletable(data: Map<String, Any>?): Completable {
    val routeEntity = data?.get(PARAM_ROUTE)

    routeEntity?.let {
        return routeRepository.saveRoute(routeEntity as Route)
    } ?: return Completable.error(IllegalArgumentException("Argument @route must be provided."))
}

它是在调用方法而不是订阅可完成时评估的。

换句话说，当您调用该方法时，它会破坏 Rx 合约并计算 data?.get(PARAM_ROUTE)。如果它是不可变的，则没有太大区别，但如果它可以在执行期间更改值，则应将其包装在 Completable.defer { }

终于到了

        .flatMap { localID ->
            route.routeId = localID
            remoteRouteSource.saveRoute(route)
        }

你正在修改链外的东西 (route.routeId = localID)，这称为副作用。

小心这些东西，Rx 的构建方式可以更安全地用于不可变对象。

我个人不会太介意，只要您了解正在发生的事情以及何时会产生问题。

【讨论】：