Zipping Room Flowable 阻止更新

Question

我的存储库中有一个 getPlaces 方法：

override fun getPlaces(filter: FilterRequest): Flowable<List<Place>> {
    return from(placesApi.filter(filter))
            .doOnSuccess {
                placesDao.savePlaces(it)
            }
            .flatMapPublisher { it ->
                placesDao.getPlaces(it.map { it.placeId })
            }
}

此方法从 api 收集结果，然后将结果保存在数据库中，returns 一个流媒体，其中包含 id 从数据库中检索到的那些地方 Flowable:

@Query("select * from Places where placeId in (:placesIds)")
fun getPlaces(placesIds: List<String>) : Flowable<List<Place>>

现在每次我更改其中一个对象时，我都可以在整个应用程序中看到更改。

现在我想将这些结果与距当前位置的距离相结合，如下所示：

 override fun addDistanceToPlaces(req: Flowable<List<Place>>): Flowable<List<Place>> {
        return req
                .zipWith(getLastLocation().toFlowable(BackpressureStrategy.LATEST),
                        BiFunction<List<Place>, Location, List<Place>> { places, location ->
                            places.forEach {
                                var placeLocation = Location(it.placeName)
                                placeLocation.latitude = it.latitude
                                placeLocation.longitude = it.longitude

                                it.distance = location.distanceTo(placeLocation)
                            }
                            places.sortedBy {
                                it.distance
                            }
                        })
                .onErrorResumeNext { t: Throwable ->
                    req
                }

    }

这有效，但是如果我应用它，我会从 Room 中丢失 "updates"；更改不会通知观察者，所以我必须手动刷新。

为什么会这样？不应该 zip 只是合并两个来源的排放量吗？

Answer 1

您的问题是在您的用例中尝试使用 zip 运算符。 Zip 通过配对输入 observable 的值来发出。它不会针对单个 observable 的每次更改而发出，而是在它们都发出时发出。查看大理石以帮助您形象化其行为：

http://reactivex.io/documentation/operators/zip.html

所以在你的情况下，Room Observable 正在发射到你的 zip 函数中，但位置 Observable 没有更新，因此你没有调用你的函数。

我认为您正在寻找 combineLatest 运算符。这将等到 Room Observable 和 Location Observable vend 一次，然后这两个 observable 都可以发射并且你的组合函数将被调用并将后续值发射到你的应用程序。

Zipping Room Flowable 阻止更新

Zipping Room Flowable prevents updates

android

rx-java2

android-room