RxJava 中的链接

Question

我遇到了以下问题。 我正在使用 Room 和 RxJava，一切正常，但我需要按以下顺序链接 4 个 rx 操作：

1 - 插入一些数据

2 - 查询一些数据

3 - 现在查询的数据再做一次插入

4 - 更新

这是我的代码，但它不起作用。

Completable c = Completable.fromAction(() -> System.out.println("Inserting data"));
Flowable f = Flowable.fromArray(1);
Completable c1 = Completable.fromSingle((x) -> System.out.println("Inserting more data with: " + x));
Completable c2 = Completable.fromAction(() -> System.out.println("Updating"));

c.andThen(f).mergeWith(c1).mergeWith(c2).subscribe();

这是输出

Inserting data
Inserting more data with: io.reactivex.internal.operators.completable.CompletableFromSingle$CompletableFromSingleObserver@233c0b17
Updating

它跳过第二个 Observable

Answer 1

Completable insert = Completable.fromAction(() -> System.out.println("Inserting data"));
Single<Integer> query = Single.just(1);
Completable update = Completable.fromAction(() -> System.out.println("Updating"));
Completable insertMore = query.flatMapCompletable(x ->
        Completable.fromAction(() ->
                System.out.println("Inserting more data with: " + x)
        ));

insert.andThen(insertMore).andThen(update).subscribe();