拉链室流动防止更新

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收集结果，然后将结果保存到数据库中，并返回一个flowable，其中这些位置按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
                }

    }

为什么会这样？ntzip应该把这两种来源的排放结合起来吗？

Answer 1

您的问题是试图为您的用例使用zip运算符。Zip通过对输入可观察的值进行配对来发出。它并不是对单个可观察物的每一次变化都发射，而是当它们都发射时。查看它的大理石，以帮助您可视化它的行为：

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

因此，在您的例子中，房间可观察到的是发射到您的zip函数中，但位置可观察到的是不更新的，因此，您不会调用您的函数。