RxJava 2 - 在另一个 Completable 之后调用 Completable

Question

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

我有两个 Completable 对象来存储一些数据。我想触发第一个，然后在第一个成功完成后才开始第二个。对第二个 Completable 的调用应该被阻止，直到第一个 Completable 成功完成。此外，如果第一个错误完成，则也应跳过另一个。

查看文档和其他 SO 问题似乎 concatWith 或 andThen 应该适合我。但是在手动测试和单元测试中我都可以看到第二个可完成的是与第一个并行触发的：/

第一次完成

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
}

我错过了什么？

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);
}

Answer 2

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

Returns 一个 Completable，首先运行此 Completable，然后运行另一个 Completable。

andThen()

firstCompletable
    .andThen(secondCompletable)

通常，此运算符是 "replacement" 用于 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)

RxJava 2 - 在另一个 Completable 之后调用 Completable

RxJava 2 - Call Completable after another Completable

java

rx-android

rx-java2