Использование flatMap с Completable

Question

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

val startPositions = BehaviorSubject.createDefault(0)

startPositions.flatMap { startPos -> App.context.repository.getConnections(startPos) }
        .flatMap { connections -> App.context.repository.storeConnections(connections) }
        .doOnNext { startPos -> startPositions.onNext(startPos + 1) }
        .subscribe({ startPos -> println("Index $startPos") })

Вот метод API для обработки загрузки данных:

 override fun getConnections(startPos: Int): Observable<List<Connection>> {
            return myAPI.getConnections(startPos)
        }

И это метод API для хранения данных:

override fun storeConnections(connections: List<Connection>): Completable =
        Completable.fromAction {
            appDao.storeConnections(connections.map {
                mapper.toDb(it)
            })
        }

Я получаю ошибку компиляции:

Несоответствие типов: предполагаемый тип (List) -> Completable но ((List) -> ObservableSource!)! ожидалось

Если возможно, я не хочу менять типы возврата моих вызовов API. Я также не уверен в использовании flatMap. Данные, возвращаемые после загрузки, представляют собой список, и я хочу, чтобы этот список оставался в виде списка в потоке. Я не хочу выдавать отдельные элементы списка.

Answer 1

Чтобы исправить ошибку несоответствия типов, используйте оператор flatMapCompletable вместо flatMap:

    .flatMapCompletable { connections -> App.context.repository.storeConnections(connections) }

Тогда вам придется использовать doOnComplete вместо doOnNext:

    .doOnComplete { startPos -> startPositions.onNext(startPos + 1) }