java - 刷新访问 token Retrofit2 + RXJava2

Question

这种方法在更新 token 时总是有效的。也就是说，对于每个请求，如果我收到 error 401，运算符 retryWhen() 会触发它更新 token 。

代码如下:

private Observable<TokenModel> refreshAccessToken() {
    Map<String, String> requestBody = new HashMap<>();
    requestBody.put(Constants.EMAIL_KEY, Constants.API_EMAIL);
    requestBody.put(Constants.PASSWORD_KEY, Constants.API_PASSWORD);

    return RetrofitHelper.getApiService().getAccessToken(requestBody)
            .subscribeOn(Schedulers.io())
            .doOnNext((AccessToken refreshedToken) -> {
                PreferencesHelper.putAccessToken(mContext, refreshedToken);
            });
}

public Function<Observable<Throwable>, ObservableSource<?>> isUnauthorized (){
    return throwableObservable -> throwableObservable.flatMap((Function<Throwable, ObservableSource<?>>) (Throwable throwable) -> {
        if (throwable instanceof HttpException) {
            HttpException httpException = (HttpException) throwable;

            if (httpException.code() == 401) {
                return refreshAccessToken();
            }
        }
        return Observable.error(throwable);
    });
}

我在向服务器发出请求的 retryWhen() 运算符处调用了 isUn​​authorized()

class RetrofitHelper {

    static ApiService getApiService() {
        return initApi();
    }

    private static OkHttpClient createOkHttpClient() {
        final OkHttpClient.Builder httpClient = new OkHttpClient.Builder();
        httpClient.addInterceptor(chain -> {
            Request originalRequest = chain.request();

            AccessToken accessToken= PreferencesHelper.getAccessToken(BaseApplication.getInstance());
            String accessTokenStr = accessToken.getAccessToken();
            Request.Builder builder =
                    originalRequest.newBuilder().header("Authorization", "Bearer " + accessTokenStr);

            Request newRequest = builder.build();
            return chain.proceed(newRequest);
        });

        return httpClient.build();
    }

    private static ApiService initApi(){
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl(Constants._api_url)
                .addConverterFactory(GsonConverterFactory.create())
                .addConverterFactory(ScalarsConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .client(createOkHttpClient())
                .build();
        return retrofit.create(ApiService.class);
    }
}

但是我们最近添加了 Basic Auth，现在在第一次请求时我得到 401 并且 retryWhen() 尝试更新 token ，但仍然得到 401。也就是说，doOnNext() 不起作用，但 onError() 立即起作用

private static Observable<AccessToken> refreshAccessToken() {
    return RetrofitHelper.getApiService()
            .getAccessToken(
                    Credentials.basic(
                            Constants._API_USERNAME, Constants._API_PASSWORD
                    ),
                    Constants._API_BODY_USERNAME,
                    Constants._API_BODY_PASSWORD,
                    Constants._API_BODY_GRANT_TYPE
            )
            .doOnNext((AccessToken refreshedToken) -> {
                PreferencesHelper.putObject(BaseApplication.getInstance(), PreferenceKey.ACCESS_TOKEN_KEY, refreshedToken);
                }

            });
}

//API服务

public interface ApiService {
    // Get Bearer Token
    @FormUrlEncoded
    @POST("oauth/token")
    Observable<AccessToken> getAccessToken(@Header("Authorization") String basicAuth,
                                           @Field("username") String username,
                                           @Field("password") String password,
                                           @Field("grant_type") String grantType);
}

在这里，告诉我为什么这是一个错误？为什么在第一个请求时我得到 401，而在第二个请求中一切正常？

Answer 1

我想提出一个更好的解决方案。

public class RefreshTokenTransformer<T extends Response<?>> implements ObservableTransformer<T, T> {

    private class HttpCode {
        private static final int UNAUTHORIZED_HTTP_CODE = 401;
    }

    private ApiService mApiService;
    private UserRepository mUserRepository;

    public RefreshTokenTransformer(ApiService service, UserRepository userRepository) {
        mApiService = service;
        mUserRepository = userRepository;
    }

    @Override
    public ObservableSource<T> apply(final Observable<T> stream) {
        return stream.flatMap(new Function<T, ObservableSource<T>>() {
            @Override
            public ObservableSource<T> apply(T response) throws Exception {
                if (response.code() == HttpCode.UNAUTHORIZED_HTTP_CODE) {
                    return mApiService.refreshToken(mUserRepository.getRefreshTokenHeaders())
                            .filter(new UnauthorizedPredicate<>(mUserRepository))
                            .flatMap(new Function<Response<TokenInfo>, ObservableSource<T>>() {
                                @Override
                                public ObservableSource<T> apply(Response<TokenInfo> tokenResponse) throws Exception {
                                    return stream.filter(new UnauthorizedPredicate<T>(mUserRepository));
                                }
                            });
                }

                return stream;
            }
        });
    }

    private class UnauthorizedPredicate<R extends Response<?>> implements Predicate<R> {

        private UserRepository mUserRepository;

        private UnauthorizedPredicate(UserRepository userRepository) {
            mUserRepository = userRepository;
        }

        @Override
        public boolean test(R response) throws Exception {
            if (response.code() == HttpCode.UNAUTHORIZED_HTTP_CODE) {
                throw new SessionExpiredException();
            }

            if (response.body() == null) {
                throw new HttpException(response);
            }

            Class<?> responseBodyClass = response.body().getClass();
            if (responseBodyClass.isAssignableFrom(TokenInfo.class)) {
                try {
                    mUserRepository.validateUserAccess((TokenInfo) response.body());
                } catch (UnverifiedAccessException error) {
                    throw new SessionExpiredException(error);
                }
            }

            return true;
        }
    }
}

我已经编写了自定义运算符，它会执行下一步操作:

1. 第一个请求开始，我们得到 401 响应码；

2. 然后我们执行/refresh_token请求来更新token；

3. 之后，如果 token 刷新成功，我们重复 第一个请求。如果/refresh_token token 失败，我们抛出异常

然后，您可以在任何请求中轻松实现它:

 Observable
    .compose(new RefreshTokenResponseTransformer<Response<{$your_expected_result}>>
(mApiService, mUserRepository()));

还有一件重要的事情: 最有可能的是，您最初用于改造的 observable 具有参数，如下所示:

mApiService.someRequest(token)

如果参数预计在执行 RefreshTokenTransformer 期间发生变化(例如/refresh_token 请求将获得新的访问 token 并将其保存在某个地方，那么您想要使用新的访问 token 来重复请求)您将需要包装您的observable 与 defer 运算符强制创建这样的新 observable:

Observable.defer(new Callable<ObservableSource<Response<? extends $your_expected_result>>>() {
            @Override
            public Response<? extends $your_expected_result> call() throws Exception {
                return mApiService.someRequest(token);
            }
        })