android - RxJava : Is there some sort of chain which lets me "wait until observables A, B, C are complete before continuing"?

Question

在 jQuery 中，promise 的想法是通过以下方式完成的:

var taskA = $.Deferred();
var taskB = $.Deferred();
var taskC = $.Deferred();

// Callback starts when all tasks are complete/resolved
$.when(taskA, taskB, taskC).done(function(results) {
    var resultA = results[0];
    var resultB = results[1];
    var resultC = results[2];
    // ...
});

// Fire off long tasks
longTask1(function(result) { taskA.resolve("result"); });
longTask2(function(result) { taskB.resolve("result"); });
longTask3(function(result) { taskC.resolve("result"); });

我觉得 RxJava 中应该有一个等效项，但我不知道它的名称。

编辑:添加我的 Android Activity 代码以进行澄清

/**
 * What I expected to happen:
 * - User has to click buttons A, B and C in order to pass.
 *
 * What actually happens
 * - User clicks any button and passes.
 */
public class RxTestActivity extends AppCompatActivity {
    PublishSubject<Boolean> subjectClickedA;
    PublishSubject<Boolean> subjectClickedB;
    PublishSubject<Boolean> subjectClickedC;

    @Bind(R.id.btnA) Button btnA;
    @Bind(R.id.btnB) Button btnB;
    @Bind(R.id.btnC) Button btnC;
    @Bind(R.id.labelTest) TextView txtSuccess;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        setContentView(R.layout.activity_test);
        ButterKnife.bind(this);

        subjectClickedA = PublishSubject.create();
        subjectClickedB = PublishSubject.create();
        subjectClickedC = PublishSubject.create();

        // Create an observer which waits until all 3 tasks are complete before
        // triggering it's own "onCompleted".
        Observable.merge(subjectClickedA, subjectClickedB, subjectClickedC)
            // Ensure all subjects are completed
            // TODO: doesn't work
            .all(new Func1<Boolean, Boolean>() {
                @Override
                public Boolean call(Boolean aBoolean) {
                    // Won't let anything through until all of tasks are complete
                    return subjectClickedA.hasCompleted() && subjectClickedB.hasCompleted() && subjectClickedC.hasCompleted();
                }
            })
            .subscribe(new Subscriber<Boolean>() {
                @Override
                public void onError(Throwable e) {
                }

                @Override
                public void onNext(Boolean aBoolean) {
                    // aBoolean value comes in as false
                    Log.e("Subscriber onNext", String.valueOf(aBoolean));
                }

                @Override
                public void onCompleted() {
                    // This triggers on the first button click (no matter which one)
                    // TODO: reward user
                    Log.e("Subscriber", "all tasks completed!");
                    txtSuccess.setText("Great success!");
                }
            });
    }

    @OnClick(R.id.btnA)
    protected void onButtonAClicked() {
        subjectClickedA.onNext(true);
//        subjectClickedA.onCompleted();

        btnA.setEnabled(false);
    }

    @OnClick(R.id.btnB)
    protected void onButtonBClicked() {
        subjectClickedB.onNext(true);
//        subjectClickedB.onCompleted();

        btnB.setEnabled(false);
    }

    @OnClick(R.id.btnC)
    protected void onButtonCClicked() {
        subjectClickedC.onNext(true);
//        subjectClickedC.onCompleted();

        btnC.setEnabled(false);
    }
}

Answer 1

你可以通过多种方式组合多个Observable。

请看一下这个。 Combining Observables

本节介绍了可用于组合多个 Observable 的运算符。

• startWith( ) — 在开始之前发出指定的项目序列从 Observable 中发出项目
• merge( ) — 将多个 Observable 合并为一个
• mergeDelayError( ) — 将多个 Observable 合并为一个，允许无错误的 Observables 在传播错误之前继续
• zip( ) — 合并两个或多个 Observables 发出的项目集通过指定的函数一起并根据结果发出项目此函数的 (rxjava-joins) and( )、then( ) 和 when( ) —通过以下方式组合由两个或多个 Observables 发出的项目集模式和计划中介
• combineLatest( ) — 当两个项目中的任何一个发出一个项目时Observables，通过一个组合每个 Observable 发出的最新项目指定的函数并根据此结果发出项目功能
• join( ) 和 groupJoin( ) — 合并两个发出的项目每当一个 Observable 中的一项落入一个 Observable由另一个项目发出的项目指定的持续时间窗口可观察
• switchOnNext( ) — 将发出 Observable 的 Observable 转换为发出最近发出的项目的单个 Observable这些 Observables 的发射