在执行另一个 Observable 之前等待 Observable 完成？答案

【问题标题】：Wait for an Observable to finish before executing another Observable?在执行另一个 Observable 之前等待 Observable 完成？
【发布时间】：2016-10-21 21:10:00
【问题描述】：

问题

我有一个定期从 API 获取数据并显示接收到的数据的活动。 API 使用 OAuth，因此我收到了一个临时访问令牌，该令牌在一段时间（1 小时）后过期。如果应用程序尝试使用过期令牌获取数据，显然请求将失败。在我的应用程序的早期迭代中，我使用 AsyncTasks 处理网络请求，基本上只是执行了一个新的 AsyncTask，它会在调用从服务器获取数据的主 AsyncTask 之前获得一个新的访问令牌。这很有效，因为主 AsyncTask 会等到另一个完成后再执行。

我最近切换到 RxJava，基本上只是用 Observables 替换了 AsyncTasks。问题是获取数据的主要 Observable 不会等待刷新访问令牌的 Observable 完成。这是我的代码，感谢您的帮助。

代码

LiveThreadActivity.java

private Subscription subscription;
private Observable<List<CustomComment>> fetchData;

@Override
protected void onResume() {
    super.onResume();

    if (tokenExpired()) {
        auth.refreshToken();
    }

    subscription = fetchData
            .compose(bindToLifecycle())
            .retryWhen(new RetryWithDelay(5, 2000))
            .subscribe(list -> addNewComments(list), e -> handleFetchDataError(e));

}


// This method gets called in onCreate()
private void dataCollection() {
    fetchData = Observable.interval(0, REFRESH_RATE, TimeUnit.MILLISECONDS)
            .map(tick -> fetchNewComments())            // Run function every time a tick is emitted
            .retryWhen( new RetryWithDelay(2, 2000) )   // Retry twice with 2 second delay
            .subscribeOn(Schedulers.io())               // Network stuff in background thread
            .observeOn(AndroidSchedulers.mainThread()); // Other stuff on the main thread

}

Auth.java

public class Auth {
    ...

    public void refreshToken() {
        Observable.just(1)
                .map(y -> refreshAccessToken())
                .retryWhen( new RetryWithDelay(3, 2000) )
                .subscribeOn(Schedulers.io())
                .subscribe();
    }
}

【问题讨论】：

标签： android rx-java rx-android

【解决方案1】：

使用反应式库需要一种新的思维方式。您必须编写代码，因为它是同步的，但要注意它是异步执行的。

您的代码只是同步执行。它同时执行两个Observable。

函数refreshToken() 应如下所示：

public Observable<?> refreshToken() {
    return Observable.just(1)
            .map(y -> refreshAccessToken())
            .retryWhen( new RetryWithDelay(3, 2000) )
            .subscribeOn(Schedulers.io());
}

还有onResume():

@Override
protected void onResume() {
    super.onResume();

    Observable obs = fetchData
            .compose(bindToLifecycle())
            .retryWhen(new RetryWithDelay(5, 2000));

    if (tokenExpired()) {
        obs = obs.startWith(auth.refreshToken());
    }

    subscription = obs
            .subscribe(list -> addNewComments(list), e -> handleFetchDataError(e));

}

注意startWith() 操作员。它允许在另一个（刷新令牌）之后执行一个Observable（获取列表）。

【讨论】：

【解决方案2】：

.flatMap() 可能就足够了，即 tokenObservable.flatMap(/* return dataObservable */)

【讨论】：