java - 异步 Observable 内部的异步 Observable

Question

我遇到了一个小问题，我使用 RxJava 从服务器检索数据，然后将这些数据与本地的数据进行检查，问题是我开始检查本地文件并存储数据，然后启动我得到的可观察数据在数据完全下载之前成功，然后我在成功后收到“initRemoteData”结果。

所以我需要做的是，当我调用“InitRemoteData”时，当可观察给出结果时，我继续阻止可观察。

private Observable<CheckStatus> getCheckObservable() {
    return Observable.defer(() -> {
        DataDTO dto;
        try {
            dto = Utils.decryptData(
                localfile.getNonce(),
                localfile.getEncryptedData(),
                password);
        } catch (WrongPasswordException e) {
            return Observable.just(CheckStatus.WRONG_PASSWORD);
        }
        try {
            storeDataPrefs(dto);
        } catch (RuntimeException e) {
            return Observable.just(CheckStatus.OTHER_ERROR);
        }
        storeDatabase(dto);

        initRemoteData();//<-Here i did call for another observable but the observable keep going without waiting it to finish.



        return Observable.just(CheckStatus.SUCCESS);
    });
}

可观察的调用:

getCheckObservable()
        .subscribeOn(BackgroundSchedulers.getMultiThreadInstance())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(Subscribers.create(checkResult -> replaceFragment(
            CheckAccountCompleteFragment
                .newInstance(checkResult),
            AddToBackStack.FALSE)));

更新: 初始化远程数据代码:

Observable.zip(getObservable1()),//remote Data
            getObservable2(), // local Data
            (observableResult1,observableResult2)->{
                doSomethingWith(observableResult1,observableResult2); // compare between data
                return null;
            }).subscribeOn(BackgroundSchedulers.getMultiThreadInstance())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnCompleted(() -> {
                    doSomething2(); // update fields
                }
            })
            .subscribe();

Answer 1

如果我猜对了，您需要让 getCheckObservable() 来等待这两个结果吗？如果是:Observable.combineLatest() 可以帮助您。

更新: 好吧，那么为什么不直接返回这个“大”Observable 中的“等待”Observable 像这样:

Observable.zip(getObservable1()),//remote Data
        getObservable2(), // local Data
        (observableResult1,observableResult2)->{
            doSomethingWith(observableResult1,observableResult2); // compare between data

            return CheckStatus.SUCCESS;   // <-- CHANGES

        }).subscribeOn(BackgroundSchedulers.getMultiThreadInstance())
        .observeOn(AndroidSchedulers.mainThread())
        .doOnCompleted(() -> {
                doSomething2(); // update fields
            }
        });

和

private Observable<CheckStatus> getCheckObservable() {
return Observable.defer(() -> {
    DataDTO dto;
    try {
        dto = Utils.decryptData(
            localfile.getNonce(),
            localfile.getEncryptedData(),
            password);
    } catch (WrongPasswordException e) {
        return Observable.just(CheckStatus.WRONG_PASSWORD);
    }
    try {
        storeDataPrefs(dto);
    } catch (RuntimeException e) {
        return Observable.just(CheckStatus.OTHER_ERROR);
    }
    storeDatabase(dto);

    return initRemoteData();       // <-- CHANGES
});
}