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, затем сохраняет результаты в базе данных и возвращает текучую среду с местами, полученными по 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
                }

    }

Это работает, однако, если я применю это, я теряю "обновления" из комнаты;Об изменениях не сообщается наблюдателям, поэтому я должен выполнить обновление вручную.

Почему это происходит?Разве zip не должен объединять выбросы из обоих источников?

Answer 1

Ваша проблема пытается использовать оператор zip для вашего случая использования.Zip излучает путем сопряжения значений входной наблюдаемой.Он не излучает при каждом изменении одну из ваших наблюдаемых, а скорее, когда они оба излучают.Посмотрите на мрамор, чтобы он помог вам визуализировать его поведение:

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

Итак, в вашем случае наблюдаемая комната излучается в функцию zip, но наблюдаемая локация не обновляется, поэтомупоэтому вы не вызываете свою функцию.

Я думаю, вы ищете оператор combineLatest.Это подождет, пока обе единицы измерения: наблюдаемая комната и наблюдаемая местоположение, а затем после этого может сгенерироваться любая наблюдаемая, и будет вызвана ваша функция объединения, и последующее значение будет отправлено в ваше приложение.