Запрет передачи новых данных обсерверам

Question

Я делаю запрос к room бд в андроиде и получаю Flowable<List<MyClass>>. Некоторые экземпляры MyClass могут быть помечены в бд как устаревшие. Чтобы их обновлять, я делаю следующее: мой провайдер для данных, перед тем как отдать запросившему этот Flowable, сам подписывает на это Flowable обсервер, который проверяет данные на актуальность, и неактуальные отправляет на актуализацию через запрос на сервер. Когда сервер отвечает в бд записываются актуальные данные, и подписчики Flowable получают актуальные данные.

Но при таком подходе основной получатель данных после подписки на Flowable сначала получит неактуальные данные, а через пару мгновений актуальные, если удалось их обновить. Я хочу избавиться от этой ситуации. Я представляю себе это так, что мы можем встроить специфичный обсервер, который отрабатывает раньше остальных, и если он вернет true, то обычные обсерверы так же получают эти данные. Если специфичный обсервер вернет false, то обсерверы не получают этих данных.

В моем случае я в этом, хм, "чекере" проверяю данные на актуальность, если все ок - отдаю тру и все получают данные. Если нет - я отправляю запрос на обновление, и если он успешно выполнился, но возвращаю false и неактуальные данные не уходят, а в момент вставки обновленных данных генерируются новые данные, которые уже проходят чекер и уходят обсерверам. Ну и если чекер словил неактуальные данные, попытался их обновить, не смог потому что сервер не доступен, то возвращает true, и обсерверы получают не обновленные данные т.к. нет ничего лучшего.

Столько текста чтобы задать простой вопрос: есть ли что-то подобное в rxjava, а то опыта с ним у меня немного и не могу найти подобный функционал...

Answer 1

Возможно, не лучший вариант, но я бы написал примерно так:

fun getData() = database.getData()
        .flatMap {dbData -> 
            if (dbData.isOutDated()) loadFromServer(dbData).onErrorReturn { dbData }
            else Flowable.just(dbData)
        }

private fun loadFromServer(outDated: Data): Flowable<Data> =
    server.getNewDataFromOld(outDated)// returns Completable
            .toFlowable()

Answer 2

Таки да такая возможность есть, оператор filter

В моем случае выглядит так:

fun get(id: Int): Flowable<MyClass> {
            return Cache.database.client.getFlowable(id).filter {
                if (!it.actual())
                    try {
                        Cache.save(API.load(it.id))
                        return@filter false
                    } catch (e: Exception) {
                        true
                    }
                true
            }.subscribeOn(Schedulers.io())
        }

Делает ровно то что мне нужно

БЛОГ НА HUSL

Запрет передачи новых данных обсерверам

Ответы (2 шт):