Async Observable 内部的 Async Observable

Question

我遇到了一个小问题，我正在使用 RxJava 从服务器中检索数据，然后将这些数据与本地数据进行检查，问题是我开始在本地检查文件并存储数据，然后我启动我得到的 observable数据完全下载之前的SUCCESS，然后我在成功后收到“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)));

更新： initRemoteData 代码：

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() 可以帮助您。

更新： 好的，那么为什么不只是 return 这个“等待” 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
});
}