RxJava: Как обновить токен при одновременном отправке нескольких запросов?

Question

У меня есть приложение, которое аутентифицируется с использованием OAuth2 и извлекает данные из службы RESTful с помощью Retrofit. Теперь у меня есть токен, поиск и обновление. Маркер обновляется как так (планировщики опущены):

// Each Retrofit call observable is "wrapper" using this method 
protected <T> Observable<T> wrap(@NonNull final Observable<T> page) { 
    return authenticate() 
     .concatMap(token -> page) 
     .onErrorResumeNext(throwable -> { 
      Log.w(TAG, "wrap: ErrorResumeNext", throwable); 
      return refreshAccessToken() 
       .flatMap(accessToken -> page); 
     })); 
} 

// Retrieves the access token if necessary 
Observable<AccessToken> authenticate() { 
    // Already have token 
    if(accessToken != null) return Observable.just(accessToken); 
    // No token yet, fetch it 
    return api.getAccessToken(...); 
} 

// Refreshes the token 
Observable<AccessToken> refreshAccessToken() { 
    return api.refreshToken(...); 
} 

Это работает, но в некоторых случаях множественные запросы посылаются одновременно, и они оба вызов освежающего процесса - в основном мое приложение заканчивает освежающий маркер, как многий в то время как на данный момент были запросы.

Итак, вопрос в следующем: как я могу гарантировать, что, когда токен необходимо обновить, он выполняется только один раз, независимо от того, сколько текущих запросов требует обновления токена? Могу ли я каким-либо образом заставить другие запросы «ждать» до тех пор, пока первый запрос не будет успешно вызван и не получит новый токен?

Answer 1

Мы выполнили это поведение с помощью горячего наблюдаемого для обновления токена и обеспечения доступа к его экземпляру для всех запросов, которые не прошли аутентификацию.

Используйте операционную систему share, чтобы превратить ваш обычный холодный наблюдаемый элемент для обновления токена в горячем режиме, чтобы каждый другой пользователь делился своим результатом. Как только запрос вернется, все ожидающие наблюдатели получают уведомление и в этот момент (в цепочке операторов он приходит прямо до share() в обратный вызов для doOnUnsubscribe) уничтожает обновляемый наблюдаемый экземпляр, чтобы следующий подписчик создавал новый. Все это может быть легко достигнуто с помощью шаблона singleton, где вы переносите обновляемое наблюдаемое в класс оболочки singleton и просто запрашиваете его через getInstance(). Если запрос отсутствует, экземпляр null - getInstance должен создать новый.

Есть еще кое-что, о чем вам нужно позаботиться, ошибка во время обновления и недействительность маркера вместе, например, но это основы.

У меня нет много времени, чтобы подробно рассказать об этом, но если вы столкнетесь с некоторыми проблемами с реализацией этого самостоятельно, оставьте комментарий, и я отправлю некоторые примеры кода завтра. Они не имеют большого смысла без контекста.