java - 如何等待异步 Observable 在另一个 Observable 上完成

Question

我有一些初始化时间较长的类，例如 Api。我的管理器类异步加载 Api 类，代码如下

class MyManager{
    private Observable<Api> apiProvider
    private Api mApi;

    MyManager(Observable<Api> apiProvider){
       this.apiProvider = apiProvider;
       apiProvider.subscribe(api -> mApi = api);
    }
}

但客户端可以调用 MyManager 类的方法，该类在创建之前需要 Api 实例。

public Observable<List<Value>> loadVlues() {
    return Observable.create(new Observable.OnSubscribe<List<Value>>() {
        @Override
        public void call(Subscriber<? super List<Value>> subscriber) {
            try {
                List<Value>> values = mApi.getValues();
                subscriber.onNext(values);
            } catch (Exception e) {
                subscriber.onError(e);
            }
        }
    });
}

我需要第二个可观察对象首先等待 apiProvider 可观察对象的执行。例如，在 Java 线程中，有简单的 join() 运算符。

使用 RXJava 执行此操作的最佳方法是什么？

<小时/>

现在我使用BlockingObservable解决了这个问题

 public Observable<List<Value>> loadValues() {

    return Observable.create(new Observable.OnSubscribe<List<Value>>() {
        @Override
        public void call(Subscriber<? super List<Value>> subscriber) {
                BlockingObservable.from(apiProvider).subscribe(new Subscriber<Api>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Api api) {
                        try {
                            List<Value>> values = mApi.getValues();
                            subscriber.onNext(values);
                        } catch (Exception e) {
                            subscriber.onError(e);
                        }
                    }
                });
        }
    });
}

这是正确的决定吗？

Answer 1

此时不需要阻止。

怎么样

class MyManager {

    Observable<Api> apiProvider;

    MyManager(Observable<Api> apiProvider){
       // cache so can be fetched several times
       this.apiProvider = apiProvider.cache();
    }
}

然后，在 loadValues() 上，将新的可观察值映射到 apiprovider 可观察值上，例如:

public Observable<List<Value>> loadValues() {
    return
       apiProvider.map(new Func1<Api, List<Value>>() {
            @Override
            public List<Value> call(Api api) {
                return api.getValues();
            }
       });
}

就像 @Cochi 在现已删除的答案中提到的那样，BlockingObservables should be used for test/demo purposes ，而不是真正的功能，因为它们违背了使用可观察量的目的。