RxJava 结合数据库创建/更新调用以获取对象列表答案

【问题标题】：RxJava combine database create/update call for a list of objectsRxJava 结合数据库创建/更新调用以获取对象列表
【发布时间】：2019-12-18 00:59:22
【问题描述】：

我目前正在为我的第一个更复杂的 rx 链而苦苦挣扎，也许你可以帮助我。我有一个对象列表，我从 api 调用中获得。我需要检查它们是否在本地存在，如果存在，则更新数据集，如果不创建它们。我目前的做法是这样的：

 private fun insertUpdateFromServer(objectsToInsert: List<Model>): Completable {
    return Observable.fromIterable(objectsToInsert).flatMapCompletable { atl ->
        dao.getByServerId(atl.id).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .switchIfEmpty { obs: SingleObserver<in Model> -> createNewObject(atl) }.flatMapCompletable {
                    updateObject(atl, it).applySchedulers()
                }
    }

重要的是，我必须等待所有子任务完成。如果我只有要更新的对象，这是可行的，但是如果有一个新对象，整个事情就不会完成。

请注意，我只对操作完成感兴趣，发出的对象对我来说并不重要。函数“getByServerId”返回一个 Maybe。

所以我问你是否能指出我的逻辑错误并把我推向正确的方向，提前谢谢！

【问题讨论】：

标签： android kotlin rx-java2 android-room rx-android

【解决方案1】：

问题出在这一行：

switchIfEmpty { obs: SingleObserver<in Model> -> createNewObject(atl) }

Kotlin 的自动 SAM 转换试图通过生成从 SingleSource 实现 void subscribe 的 switchIfEmpty 重载来让您的生活更轻松。在并非所有情况下这都有用，但是您想要使用的是：

switchIfEmpty(createNewObject(atl))

here.

可以更好地解释为什么会发生这种情况

【讨论】：

感谢您指出这一点！但这并不能完全解决我的问题，因为现在添加对象并在此之后更新它的函数。有没有可能避免这种情况？

【解决方案2】：

所以我最终找到了一个解决方法，我找到了here。

TL;DR：可能会发出 null 并调用“onComplete”。所以如果我使用“onSuccess”和“onComplete”，我可以链接它。

有兴趣的人的代码：

     private fun insertUpdateFromServer(objects: List<Model>): Completable {
    return Observable.fromIterable(objects)
            .flatMapCompletable { obj ->
                dao.getByServerId(obj.id).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                        .doOnComplete { createNewObject(obj).subscribe() }
                        .doOnSuccess { updateObject(obj, it).applySchedulers().subscribe() }
                        .flatMapCompletable { Completable.complete() }
            }
}

这个表达式等待它的所有子任务终止，然后完全终止。

【讨论】：