Как выполнить код в doOnNext в rxjava?

Question

[Lordao]

Из dataSource получаю список в виде Flowable<List<Item>> от сервера. После этого мне необходимо каждый элемент добавить в локальный список cashes и затем добавить в БД. Но код во втором doOnNext не выполняется.

return remoteDataSource.loadItem(true).doOnNext {
            Log.d("Log", it.toString())
            caches.clear()
            localDataSource.clearData()
        }.take(1).flatMap { list -> Flowable.fromIterable(list) }.doOnNext {
            Log.d("Log", it.title)
            caches.add(it)
            localDataSource.addItem(it)
        }.toList().toFlowable()

Answer 1

[Денис Загаевский]

Не нужно так делать в Rx. В doOnNext не должно быть настолько мощных сайд-эффектов. В лог написать - ок, в бд сходить - не ок. Нужно завернуть clear/add в Completable.

remoteDataSource.loadItem(true)
    .flatMap {
         cache.clear().andThen(Observable.just(it))
    }
    .flatMap { list ->
         Observable.fromIterable(list)
            .flatMapCompletable { cache.add(it) }
            .andThen(Observable.just(list))
     }

Как-то так, примерно.

Comments

[Lordao]

Я не до конца понял каким образом завернуть локальный список cache в Completable.

В моём коде он инициализирован таким образом var cache: MutableList<Item> = arrayListOf()

И в случае того, если он не пустой, возвращаем его в видеFlowable.just<MutableList<Item>>(cache)

Когда пытаюсь применитьcache.clear().andThen(Observable.just(it)), то строчка не резолвится, так как cache простой список

[Денис Загаевский]

Lordao, в rx нельзя так работать с мутабельными данными. Заверни этот кэш в правильно rx- обёртку.
Это может быть либо класс с соответствующим интерфейсом, либо можно внутри rx цепочки его хранить с помощью scan, share. Зависит от задач.