RxJava 2 - 在一个Completable完成后调用另一个Completable

Question

我是RxJava的新手，遇到了以下问题：

我有两个Completable对象用于存储一些数据。我想先触发第一个Completable，然后只有在第一个Completable成功完成后才启动第二个Completable。对第二个Completable的调用应该被阻塞，直到第一个Completable成功完成。此外，如果第一个Completable以错误结束，那么另一个Completable也应该被跳过。

查看文档和其他SO问题，似乎concatWith或andThen适合我。但是，在手动测试和单元测试中，我发现第二个Completable与第一个Completable并行触发 :/

第一个Completable

public Completable doA() {
  Log.d(TAG, "class call");

  return db.countRows()
    .doOnSuccess(integer -> {
      Log.d(TAG, "found rows: "+integer);
    })
    .doOnError(Throwable::printStackTrace)
    .flatMapCompletable(this::customAction);
}

private Completable customAction(final int count) {
  Log.d(TAG, "count: "+count);
  if (count > 0) {
    Log.d(TAG, "no rows, skip");
    return Completable.complete();
  }

  final User user = ...
  return db.save(user); // return Completable
}

第二个可完成的操作

public Completable doB() {
  Log.d(TAG, "call to B");
  // ...
}

在执行A之后尝试调用B

public Completable someMethod() {
    Log.d(TAG, "someMethod");
    return doA()
        .andThen(doB());
        // this also doesn't work
        //.concatWith(doB());
}

订阅

someMethod()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .doOnComplete(() -> {
      Log.d(TAG, "complete");
      // ...
    })
    .doOnError(throwable -> {
      Log.d("Main", "error "+throwable.getMessage());
      // ...
    })
    .subscribe();

当我运行我的应用并检查日志时，我可以看到：

D/Some method: some method
D/DoA: class call
D/DoB: class call   // <- why here?
D/DoA: found rows: 0
D/DoA: count: 0

还有以下单元测试失败：

@Test
public void test() {
  when(doa.doA()).thenReturn(Completable.error(new Exception("test")));

  observe(); // subscription with TestObserver

  verify(dob, never()).doB(); // fails with NeverWantedButInvoked
}

我漏掉了什么？

- LR89

---

你能分享一下你是如何重写这段代码的吗？以便可以验证.andThen(...)在第一个completable完成后才发生。 - siger

Answer 1

因为你调用了 doB()，让我来重写你的流程：

public Completable someMethod() {
    Log.d(TAG, "someMethod");

    // doA() inlined
    LOG.d("class call");
    Completable a = ...

    // doB() inlined
    Log.d("class call");
    Completable b = ...

    return a.andThen(b);
}

- akarnokd

---

我没有看到我们的代码片段有什么不同，可能会影响可完成项的顺序。你能否再解释一下？ - LR89

你认为它们是并行运行的，但实际上，在它们任何一个实际执行之前，你将这两个“类调用”按顺序依次记录。 - akarnokd

@JemshitIskenderov 在什么情况下？请提交详细问题。 - akarnokd

我可以邀请您加入 Android 开发者的 Slack 群组吗？http://android-united.community/ - Jemshit Iskenderov

我不太喜欢这种频道，因为它们无法促进关于特定问题的有界对话。另外，Slack是半私人的，而StackOverflow至少允许其他人发现他们遇到的类似问题的答案。 - akarnokd

Answer 2

你可以使用 andThen() 或 concatWith() 运算符。

返回一个 Completable，首先运行此 Completable，然后运行另一个 Completable。 andThen()

firstCompletable
    .andThen(secondCompletable)

通常情况下，这个操作符是 Completable 上 flatMap 的“替代品”：

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)

concatWith:

firstCompletable
        .concatWith(secondCompletable)

- Rahil Ali