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

Question

问题

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

我最近切换到 RxJava，基本上只是用 Observables 替换了 AsyncTasks。问题在于获取数据的主要 Observable 不等待刷新访问 token 的 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(获取列表)(刷新 token )。