阻塞方法执行直到异步 RxJava 调用 returns

Question

我很确定以前一定有人问过这个问题，但我找不到了。我正在尝试做一些非常简单的事情，即使对于像我这样的 RxJava 初学者也是如此。我有以下方法：

 public ILocation findLocationDetails() {
    requestLocationDetails();
    return buildLocationFromDetails();
}

方法 requestLocationDetails 包含在另一个线程内执行的 REST 调用：

private void requestLocationDetails() {
    compositeDisposable.add(
        Observable
        .fromCallable((() -> JsonRestCaller.readJsonFromUrl(buildUrl())))
        .subscribeOn(Schedulers.io())
        .subscribeWith(new DisposableObserver<JsonObject>() {
            @Override
            public void onNext(JsonObject jsonObject) {
                try {
                    parseJson(jsonObject);
                } catch (IOException e) {
                    Log.e(TAG, e.getMessage());
                }
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, e.getMessage());
            }
            @Override
            public void onComplete() {
            }
    }));
}

现在我要做的就是等待 parseJson() 完成，这样 buildLocationFromDetails 方法就可以处理在 parseJson 中检索到的详细信息。

我读了一些关于阻塞运算符的内容，但我不确定如何让它们在我的案例中起作用。我还发现了一些例子，有人只是等待了一段时间以确保结果可用，但我不知道要等待多长时间，这并没有让我觉得这是正确的方法。

Answer 1

你可以试试下面的方法

//Create a method which will parse json and return the result in String 
//(I am assuming string as parse response for simplicity)
public Observable<String> parseJson(JsonObject jsonObject) {
    return Observable
        .create(
            e -> {
                //parse json code goes here.
                // Once parsing done pass the details
                e.onNext("Details from parsing json");
                e.onComplete();
            }
        );
}

现在修改上面你写的代码，

Observable
    .fromCallable((() -> JsonRestCaller.readJsonFromUrl(buildUrl())))
    //Pass the jsonObject from above to parseJson() method
    .flatMap(jsonObject -> parseJson(jsonObject))
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    //details is the string from parsing json
    .subscribe(
        details -> buildLocationFromDetails(details),
        error -> new Throwable(error)
    );

希望对您有所帮助。

阻塞方法执行直到异步 RxJava 调用 returns

Blocking method execution until asynchronous RxJava call returns

methods

block

rx-java2