等待一个可观察量完成，然后再执行另一个可观察量？

Question

问题是

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

我最近切换到RxJava，基本上只是用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

.flatMap（） 可能就足够了，即令牌可观察.flatMap（/* 返回数据可观察 */）

Answer 2

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

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

函数< code>refreshToken()应该如下所示:

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

和在恢复（）上：

@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（）操作符。它允许依次执行一个＜code＞可观察的＜code＞（获取列表）（刷新令牌）。