如何在RxJava2中链接两个Completable

Question

我有两个可完成的.我想做以下场景:如果第一个Completable到达onComplete,继续第二个Completable.最终结果将是第二次完成的完成.

当我有单个getUserIdAlreadySavedInDevice()和Completable login()时,我就是这样做的:

@Override
public Completable loginUserThatIsAlreadySavedInDevice(String password) {
    return getUserIdAlreadySavedInDevice()
            .flatMapCompletable(s -> login(password, s))

}

Answer 1

您正在寻找andThen运营商.

返回首先运行此Completable然后另一个可完成的Completable.

firstCompletable
    .andThen(secondCompletable)

通常,此运算符是flatMapon 的"替代" Completable:

Completable       andThen(CompletableSource next)
<T> Maybe<T>      andThen(MaybeSource<T> next)
<T> Observable<T> andThen(ObservableSource<T> next)
<T> Flowable<T>   andThen(Publisher<T> next)
<T> Single<T>     andThen(SingleSource<T> next)

Answer 2

TL; DR:其他答案错过了一个微妙之处.使用doThingA().andThen(doThingB()),如果你想的等效concat,使用doThingA().andThen(Completable.defer(() -> doThingB()),如果你想的等价flatMap.

上面的答案是正确的,但我发现它们具有误导性,因为它们错过了关于热切评价的微妙之处.

doThingA().andThen(doThingB())将doThingB()立即调用但仅订阅doThingB()当返回的observable doThingA()完成时返回的observable .

doThingA().andThen(Completable.defer(() -> doThingB())doThingB()只有在事物A完成后才会打电话.

只有doThingB()在订阅事件之前有副作用时,这才是重要的.例如Single.just(sideEffect(1)).toCompletable()

在订阅事件(真正的冷可观察)之前没有副作用的实现可能是Single.just(1).doOnSuccess(i -> sideEffect(i)).toCompletable().

如果只是咬了我的东西A是一些验证逻辑并doThingB()立即启动异步数据库更新,完成VertX ObservableFuture.这是不好的.按理说doThingB()应该写只更新在订阅数据库,我要去尝试,并在未来的设计的东西的方式.

Answer 3

尝试

Completable.concat

Returns a Completable which completes only when all sources complete, one after another.

http://reactivex.io/RxJava/javadoc/io/reactivex/Completable.html#concat(java.lang.Iterable)

Answer 4

请注意，这里获得更多票数的答案有点误导。请参阅下面的示例，我的想法是展示一些测试场景并展示操作符的可完成逻辑的andThen行为方式。

 private fun doThingAFail(): Completable {
        print("Do thingA Called\n")
        return Completable.fromCallable {
            print("calling stream A\n")
            throw(Exception("The excep"))
        }
    }

    private fun doThingB(): Completable {
        print("Do thingB Called\n")
        return Completable.fromCallable {
            print("calling stream B\n")

        }
    }

    private fun doThingA(): Completable {
        print("Do thingA Called\n")
        return Completable.fromCallable {
            print("calling stream A\n")
        }
    }

请观察以下测试：

@Test
fun testCallAPlusB() {
    doThingA().andThen(doThingB())
}

输出将是：

Do thingA Called
Do thingB Called

这里快速说明：请注意，我们没有订阅此代码片段中的这些 Completables。

对于测试：

@Test
fun theTestSubscribe() {
    doThingA().andThen(doThingB()).subscribe()
}

输出将是：

Do thingA Called
Do thingB Called
calling stream A
calling stream B

最后，如果第一个可完成失败，第二个可完成将不会被执行。

@Test
fun theTestFailThingA() {
    doThingAFail().andThen(doThingB()).subscribe()
}

输出将是：

Do thingA Called
Do thingB Called
calling stream A

这里的关键概念是方法内部的逻辑和可观察对象内部的逻辑不是同时执行的。doThingA()一旦我们调用和doThingB()方法，“Do thingA Called”和“Do thingB Called”行将被打印。而“调用流 A”和“调用流 B”行仅当有人订阅doThingA和doThingB方法时才会被调用。

这里的第二个概念是 andThen 运算符如何处理错误。在上面的示例中，如果doThingA()completable 最终出现错误，流将结束并且不会打印“调用流 B”行。