RxJava: запускать список завершаемых файлов, пока не получится

Question

У меня есть функция, которая принимает один вход и возвращает Completable.

Я хотел бы запустить функцию последовательно в списке входов-кандидатов, пока не найду функцию, для которой Completable завершает работу без ошибки.

fun doStuff(input: Int): Completable { ... }

Observable.just(1,2,3,4)
          .flatMapCompletable { doStuff(it) }
          .??? // Run completables until one succeeds

Я не использую merge или concat, потому что они будут выполнять все Completables.Я хочу, чтобы он прекратил выполнение Completables после того, как один из них будет найден, и это удастся.

Существует ли стандартная операция для достижения этого поведения?

Answer 1

Возможное решение, хотя требуется немного RxAcrobatics:

Observable
    .just(1, 2, 3, 4)
    .flatMapSingle { arg ->
        doStuff(arg)
            .toSingle { true }
            .onErrorReturn { false }
    }
    .filter { it }
    .firstElement()
    .subscribe()

Если у вас есть контроль над функцией, которая возвращает Completable, я предлагаю изменить это на Single<Boolean>, счто вы могли бы более элегантно использовать takeUntil, например:

Observable
    .just(1,2, 3, 4)
    .flatMapSingle(::doStuff)
    .takeUntil { it }
    .subscribe()

Если у вас нет контроля над функцией, которая возвращает Completable, вы все равно можете создать небольшую оболочку:

private fun doStuffWrap(foo: Int) = doStuff(foo)
    .toSingle { true }
    .onErrorReturn { false } 

И используйте элегантное решение:

Observable
    .just(1, 2, 3, 4)
    .flatMapSingle(::doStuffWrap)
    .takeUntil { it }
    .subscribe()

Answer 2

используйте takeUntil с вашим предикатом, проверяющим, успешно он или нет.