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

Question

问题

我有一个 activity，它定期从 API 获取数据并显示接收到的数据。 API 使用 OAuth，所以我收到一个临时访问令牌，该令牌会在一段时间（1 小时）后过期。如果应用程序尝试使用过期令牌获取数据，显然请求将失败。在我的应用程序的早期迭代中，我使用 AsyncTasks 处理网络请求，实际上只是执行了一个新的 AsyncTask，它会在调用从服务器获取数据的主 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();
    }
}

Answer 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（获取列表）。

Answer 2

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

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

Wait for an Observable to finish before executing another Observable?

android

rx-java

rx-android