kotlin - 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，因为它们将执行所有Completable。一旦希望成功，我希望它停止执行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()