连续执行不同的Completables

Question

我目前正在尝试通过 java 中的响应式扩展实现特定结果，但是我无法做到，也许你们中的某个人可以帮助我。

firstCompletable
  .onErrorComplete(t -> specificErrorHandlingOne())
  .andThen(secondCompletable())
  .onErrorComplete(t -> specificErrorHandlingTwo())
  .andThen(thirdCompletable())
  .onErrorComplete(t -> specificErrorHandlingThree())
  .andThen(fourthCompletable())
  .onErrorComplete(t -> specificErrorHandlingFour())
  .subscribe(viewCallback::showSuccess)

但是，当secondCompletable出现错误时，会执行特定的错误处理，但其他Completable仍在调度中。如果其中一个 Completables 失败，我希望整个 Completables 链停止执行。我该怎么做？

我已经尝试改用 doOnError，但这只是在抛出的特定错误的堆栈跟踪上结束。

Answer 1

Completable.concat(
    completable1.doOnError(e -> {...}),
    completable2.doOnError(e -> {...}),
    completable3.doOnError(e -> {...}),
    completable4.doOnError(e -> {...})
).subscribe(action, errorConsumer);

Completables 将按指定顺序订阅
action 全部完成后调用
您可以为每一个指定错误处理程序（这是可选的）
任何错误都会破坏管道并传播到订阅者 (errorConsumer)

您原来的 andThen 链也应该有效，但您需要用 doOnError 替换 onErrorComplete，它用完成替换错误，它只调用指定的操作。或者只是 return false 来自你的 specificErrorHandlingXxx().

Answer 2

试试下面的方法：

public static void main(String[] args) {
    System.out.println("start");
    Completable c1 = Completable.fromAction(() -> printAndWait(1, 1));
    Completable c2 = Completable.fromAction(() -> printAndWait(2, 2));
    Completable c3 = Completable.fromObservable(Observable.timer(3, TimeUnit.SECONDS).concatWith(Observable.error(new RuntimeException())));
    Completable c4 = Completable.fromAction(() -> printAndWait(4, 2));

    c1.concatWith(c2).concatWith(c3).concatWith(c4).subscribe(e -> e.printStackTrace(), () -> System.out.println("done"));

    printAndWait(10, 10);//dont exit till program is completely executed

}

private static void printAndWait(int i, int j) {
    System.out.println(i);
    Observable.timer(j, TimeUnit.SECONDS).toBlocking().subscribe();//just add delay
}

连续执行不同的Completables

Execute different Completables in succession

java

rx-java

reactivex