Как сделать синхронизирующий вызов, используя RxJava

Question

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

class ApolloAuthenticator(private val authenticated: Boolean) : Authenticator {

  @Throws(IOException::class)
  override fun authenticate(route: Route, response: Response): Request? {
    // Refresh your access_token using a synchronous api request

    if (response.request().header(HEADER_KEY_APOLLO_AUTHORIZATION) != null) {
      return null  //if you've tried to authorize and failed, give up
    }

    synchronized(this) {

      refreshTokenSync() // This is blocked and never unblocked
      val newToken = getApolloTokenFromSharedPreference()
      return response.request().newBuilder()
        .header(HEADER_KEY_APOLLO_AUTHORIZATION, newToken)
        .build()
    }

    private fun refreshTokenSync(): EmptyResult {
      //Refresh token, synchronously
      val repository = Injection.provideSignInRepository()
      return repository
        .signInGraphQL()
        .toBlocking()
        .first()

   }

   fun signInGraphQL() : Observable<EmptyResult> =
sharedPreferencesDataSource.identifier
  .flatMap { result -> graphqlAuthenticationDataSource.getAuth(result) }
  .flatMap { result -> sharedPreferencesDataSource.saveApolloToken(result) }
  .onErrorReturn { EmptyResult() }

}

---------- Использование его

 val apollAuthenticator = ApolloAuthenticator(authenticated)
    val okHttpBuilder =
      OkHttpClient.Builder()
        .authenticator(apollAuthenticator)

Answer 1

Я не нашел способа сделать синхронизирующий вызов с использованием RxJava, но я могу сделать это с помощью kotlin coutorine runBlocking, который блокирует поток до завершения запроса:

synchronized(this) {
    runBlocking {
        val subscription = ApolloReauthenticator.signInGraphQl() // await until it's finished
          subscription.unsubscribe()
     }
}


fun signInGraphQl(): Subscription {
   return repository.refreshToken()
     .subscribe(
       { Observable.just(EmptyResult()) },
       { Observable.just(EmptyResult()) }
     )
 }