Как concatEagerDelayError в RxJava2

Question

Как реализовать Observable.concatEagerDelayError или эквивалент в RxJava2 / RxKotlin2?

Есть:

• Observable.concatEager
• Observable.concatDelayError

Но не:

• Observable.concatEagerDelayError

Что у меня есть:

fun getAll(): Observable<List<User>> = Observable.concatArrayDelayError(
    // from db
    userDAO
        .selectAll()
        .subscribeOn(ioScheduler),
    // from api
    userAPI
        .getAll()
        .doOnNext { lstUser -> Completable.concatArray(
            userDAO.deleteAll().subscribeOn(ioScheduler),
            userDAO.save(lstUser).subscribeOn(ioScheduler)
        ) }
        .subscribeOn(ioScheduler)
)

Я хочу того же поведения, но с нетерпением жду selectAll () и getAll (), потому что нет причин ждать от db для запуска сетевого вызова.

Answer 1

Использование concatMapEagerDelayError:

 Observable.fromIterable(sources)
 .concatMapEagerDelayError(v -> v, true);

 Observable.fromArray(source1, source2, source3)
 .concatMapEagerDelayError(v -> v, true);

JavaDoc .

Редактировать:

fun getAll(): Observable<List<User>> = Observable.fromArray(
    // from db
    userDAO
        .selectAll()
        .subscribeOn(ioScheduler),
    // from api
    userAPI
       .getAll()
       // --- this makes no sense by the way -------------------
       .doOnNext { lstUser -> Completable.concatArray(
            userDAO.deleteAll().subscribeOn(ioScheduler),
            userDAO.save(lstUser).subscribeOn(ioScheduler)
       )}
       // ------------------------------------------------------
       .subscribeOn(ioScheduler)
)
.concatMapEagerDelayError({ v -> v }, true)