RxJava：仅当第一个抛出错误并从第一个重复时才执行第二个 observable

Question

我正在使用retorift 访问getAricle api 并获取与用户相关的文章列表。如果传递的令牌过期，getArticle api 将抛出错误，如果是这样，我必须调用 refreshToken api 来获取新令牌，然后我必须再次调用 getArticle api

 ApiController.createRx().getArticle(token)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({ response -> toast(response.body().url) }, { e ->
                println(e.printStackTrace())
                if(e is HttpException && e.code() in  arrayOf(401,403)){                      
                   //Here I want to call refresh tolken api
                   toast("Auth error")
                }
                else
                   toast(R.string.something_went_wrong)
            })

编辑

尽管给出的答案显示了一些方向，但这些并不是我问题的直接答案。这就是解决它的方法，但我觉得这可以重构为更好的代码

ApiController.createRx().getArticle(Preference.getToken())
            .flatMap { value ->
                if (value.code() in arrayOf(403, 401)) {
                    ApiController.refreshToken()
                    ApiController.createRx().getArticle(Preference.getToken())
                } else Observable.just(value)
            }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({ response -> println("Success") }, { e ->
                e.printStackTrace()
                toast(R.string.something_went_wrong)
            })



fun refreshToken() {
        val token:String?=ApiController.createRx().refreshToken(Preferences.getRefreshToken()).blockingFirst()?.body()?.token
        if (token != null) Preferences.setAuthToken(token)
    }

编辑

我将代码重构为更简洁的版本

Observable.defer { ApiController.createRx().getArticle(Preferences.getToken()) }
            .flatMap {
                if (it.code() in arrayOf(401, 403)) {
                    ApiController.refreshToken()
                    Observable.error(Throwable())
                } else Observable.just(it)
            }
            .retry(1)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({println("Success") }, {
              it.printStackTrace()
              toast(R.string.something_went_wrong)
            })



 fun refreshToken() {
        var token: String? = null
        try {
            token = createRx().refreshToken(Preferences.getRefreshToken()).blockingFirst().body()!!.token
        } catch (e: Exception) {
            throw e
        }
        println("saving token")
        if (token != null) Preferences.setAuthToken(token)
    }

编辑

请查看我对最终重构代码的回答

Answer 1

我已经实现了这个确切的东西。这是该代码的略微修改版本：

private Observable<Object> refreshTokenIfNotAuthorized(Observable<? extends Throwable> errors) {
    final AtomicBoolean alreadyRetried = new AtomicBoolean(false);

    return errors.flatMap(error -> {

        boolean isAuthorizationError = /* some logic analyzing each error*/ ;

        if (isAuthorizationError && !alreadyRetried.get()) {
            try {
                alreadyRetried.set(true);
                String newToken = federatedTokenRefresher.refreshToken()
                                                         .toBlocking()
                                                         .first();

                setLogin(newToken);
                return Observable.just(null);

            } catch (Exception e) {
                return Observable.error(error);
            }

        }
        return Observable.error(error);
    });
}

你可以像这样使用这个方法：

doSomethingRequiringAuth().retryWhen(this::refreshTokenIfNotAuthorized);

Answer 2

您将收到什么样的错误？看来您可以使用onErrorResumeNext 运算符。

这个操作符一旦接收到一个 throwable，允许你返回一个 Observable 而不是 onError 中的 throwable

@Test
    public void observableOnErrorResumeException() {
        Integer[] numbers = {0, 1, 2, 3, 4, 5};

        Observable.from(numbers)
                .doOnNext(number -> {
                    if (number > 3) {
                        try {
                            throw new IllegalArgumentException();
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                    }

                })
                .onErrorResumeNext(t -> Observable.just(666))
                .subscribe(System.out::println);

    }

您可以在这里查看更多示例https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/errors/ObservableExceptions.java

Answer 3

我会给你另一个使用 groupBy 运算符的选项

/**
 * In this example we create a response code group.
 */
@Test
public void testGroupByCode() {
    Observable.from(Arrays.asList(401,403, 200))
            .groupBy(code -> code)
            .subscribe(groupByCode -> {
                switch (groupByCode.getKey()) {
                    case 401: {
                        System.out.println("refresh token");
                        processResponse(groupByCode);
                        break;
                    }
                    case 403: {
                        System.out.println("refresh token");
                        processResponse(groupByCode);
                        break;
                    }
                    default: {
                        System.out.println("Do the toast");
                        processResponse(groupByCode);
                    }
                }
            });
}

private void processResponse(GroupedObservable<Integer, Integer> groupByCode) {
    groupByCode.asObservable().subscribe(value -> System.out.println("Response code:" + value));
}

Answer 4

在阅读了有关RxJava 的更多信息后，我解决了我的问题，这就是我实现它的方式。 首先将retrofit 向onError 或onNext\onSuccess 抛出4xx 错误取决于我们如何定义它。 例如：

@GET("content") fun getArticle(@Header("Authorization") token: String):Single<Article>

这会将所有 4xx 错误抛出到 onError 而不是 Single<Article> 如果您将其定义为 Single<Response<Article>> 那么来自服务器的所有响应（包括 4xx）都将转到 onNext\onSuccess

Single.defer { ApiController.createRx().getArticle(Preferences.getAuthToken())}
                .doOnError {
                    if (it is HttpException && it.code() == 401)
                        ApiController.refreshToken()
                }
                .retry { attempts, error -> attempts < 3 && error is HttpException && error.code() == 401 }
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({println("Success") }, {
                  it.printStackTrace()
                  toast(R.string.something_went_wrong)
                })

我使用defer 作为我实际Observable 的包装器，因为我想在令牌刷新后重新创建可在重试时观察到的文章获取，因为我希望在我的刷新令牌代码存储新获取时再次调用Preferences.getAuthToken()令牌优先。

如果HttpException 为 401 且未尝试重试超过 2 次，则retry 返回 true