Implementing Single Token Refresh Call with Two Networking Libraries (AsyncHttpClient & Retrofit)

Question

I'm working with a legacy codebase that utilizes two different networking libraries: AsyncHttpClient and Retrofit. My goal is to ensure that the token refresh API is called only once during the initial request, even if multiple network requests are made simultaneously.

Currently, I'm attempting to implement a queue-based system to manage token refreshes sequentially. However, I'm facing challenges integrating this approach with both AsyncHttpClient and Retrofit in a way that guarantees only a single token refresh call, regardless of the number of concurrent requests.

Any insights or suggestions on how to efficiently handle token refresh with these two libraries in a synchronized manner would be greatly appreciated.

class TokenManager @Inject constructor(
    private val mobileRepository: MobileRepository,
    private val userService: UserService
) {
    private val mutex = Mutex()
    private var lastEventTime = 0L
    private var sendEvent = false
    private val refreshQueue = TokenRefreshQueue()

    private val refreshedToken = AtomicReference("")
    val isRefreshing = AtomicBoolean(false)


    fun handleEvent(event: TokenRefreshFailEvent) {
        TokenStatusChecker.setTokenIsInValid(true)
        if(sendEvent)return
        CoroutineScope(Dispatchers.Main).launch {
            mutex.withLock {
                val currentTime = System.currentTimeMillis()
                if (currentTime - lastEventTime > 1500) {
                    EventBus.getDefault().postSticky(event)
                    lastEventTime = currentTime
                    sendEvent = true
                    Log.d("TAG", "handleEvent: sended event")
                }
            }
        }
    }

    fun refreshToken(onSuccess: (String) -> Unit, onFailed: (Int) -> Unit) {
        if (isRefreshing.get()) {
            // 이미 갱신 중이면 큐에 추가
            refreshQueue.enqueue { onSuccess(refreshedToken.get()) }
        } else {
            isRefreshing.set(true)
            actualTokenRefresh({ newToken ->
                onSuccess(newToken)
                processRemainingRequests()
            }, onFailed)
        }
    }

    private fun processRemainingRequests() {
        while (refreshQueue.isNotEmpty()) {
            refreshQueue.poll()?.invoke()
        }
        isRefreshing.set(false)
    }


    private fun actualTokenRefresh(onSuccess: (String) -> Unit, onFailed: (Int) -> Unit) {
        try {
            val refreshToken = mobileRepository.getRefreshToken()
            val response = userService.refreshTokenSync(RefreshTokenRequest(refreshToken)).execute()
            if (response.isSuccessful) {
                response.body()?.let {
                    mobileRepository.setUserToken(it.response?.accessToken ?: "")
                    mobileRepository.setRefreshToken(it.response?.refreshToken ?: "")
                    mobileRepository.setRefreshTokenExpiry(it.response?.refreshTokenExpiry ?: 0)
                    mobileRepository.setAccessTokenExpiry(it.response?.accessTokenExpiry ?: 0)
                    refreshedToken.set(it.response?.accessToken ?: "")
                    onSuccess(it.response?.accessToken ?: "")
                }
            } else {
                val json = JSONObject(response.errorBody()?.string())
                val errorCode = if (json.has("error") && json.getJSONObject("error").has("code")) {
                    json.getJSONObject("error").getInt("code")
                } else {
                    102
                }
                handleEvent(TokenRefreshFailEvent(errorCode))
                onFailed(errorCode)
            }
        } catch (e: Exception) {
            e.printStackTrace()
            handleEvent(TokenRefreshFailEvent(101))
            onFailed(101)
        } finally {
            isRefreshing.set(false)
            refreshQueue.onTaskComplete()
            Log.d("TAG", "actualTokenRefresh: refresh end")
        }
    }


    companion object {


        @Volatile
        private var INSTANCE: TokenManager? = null

        fun getInstance(
            mobileRepository: MobileRepository,
            userService: UserService
        ): TokenManager {
            return INSTANCE ?: synchronized(this) {
                INSTANCE ?: TokenManager(mobileRepository, userService).also { INSTANCE = it }
            }
        }
    }
}

class TokenRefreshQueue {
    private val queue: Queue<() -> Unit> = LinkedList()

    @Synchronized
    fun enqueue(refreshTask: () -> Unit) {
        Log.d("TAG", "enqueue: queue size ${queue.size}")
        queue.add(refreshTask)
      if (queue.size == 1) {
            processNext()
        }
    }

    @Synchronized
    private fun processNext() {
        queue.peek()?.invoke()
    }

    @Synchronized
    fun onTaskComplete() {

        Log.d("TAG", "onTaskComplete:  ${queue.size}")
        if (queue.isNotEmpty()) {
            queue.remove() // 요소 제거 전에 큐가 비어있지 않은지 확인
        }

        if (queue.isNotEmpty()) {
            processNext()
        }
    }

    @Synchronized
    fun isNotEmpty(): Boolean {
        return queue.isNotEmpty()
    }

    @Synchronized
    fun poll(): (() -> Unit)? {
        return if (queue.isNotEmpty()) queue.poll() else null
    }
}

open class AuthAsyncHandler constructor(
    private val client: AsyncHttpClient,
    private val context: Context,
    private val userService: UserService,
    private val retryAction:()->Unit,
    private val logTag:String = ""
) : AsyncHttpResponseHandler() {

    private val tokenManager:TokenManager
    private val handlerCoroutineScope = CoroutineScope(Dispatchers.IO)
    init {
        val mobileRepository: MobileRepository = SharedPreferencesUtil(context)
        tokenManager = TokenManager.getInstance(mobileRepository,userService)
    }


    override fun onStart() {
        super.onStart()
        client.addHeader("x-platform", "ANDROID")

    }

    final override fun onSuccess(
        statusCode: Int, headers: Array<Header?>?,
        response: ByteArray?
    ) {

        if (response == null) {
            onFailure(statusCode, headers, response, null)
            return
        }
        val body: String = String(response)
        val jsonObject = JSONObject(body)
        if (jsonObject.has("error")) {
            val errorObject = jsonObject.getJSONObject("error")
            if (errorObject.has("code")) {
                val errorCode = errorObject.getInt("code")
                if(errorCode == 1003){
                    // here
                    Handler(Looper.getMainLooper()).run{
                        tokenManager.refreshToken(onSuccess = {
                            retryAction()
                        }, onFailed = {
                            client.cancelAllRequests(true)
                        })
                    }
                }else{
                    handleError(errorCode)
                }
            }
        } else if (jsonObject.has("response")) {
            handleSuccess(statusCode, headers, response)
        }
    }



    open fun handleSuccess(statusCode: Int, headers: Array<Header?>?, response: ByteArray?) {
    }

    private fun handleError(errorCode: Int) {
        tokenManager.handleEvent(TokenRefreshFailEvent(errorCode))
    }

    override fun onFailure(
        statusCode: Int,
        headers: Array<Header?>?,
        errorResponse: ByteArray?,
        e: Throwable?
    ) {
        handleError(statusCode)
    }

    override fun onRetry(retryNo: Int) {
        super.onRetry(retryNo)
    }


}

class AuthInterceptor(
    private val mobileRepository: MobileRepository,

    private val lazyUserService: Lazy<UserService>

) : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response {


        val tokenManager = TokenManager.getInstance(mobileRepository,lazyUserService.get())
        val originalRequest = chain.request()


        if(mobileRepository.getUserToken().isEmpty()){
            return chain.proceed(originalRequest)
        }


        val tokenAddedRequest = chain.request().putTokenHeader(mobileRepository.getUserToken())

        val response = chain.proceed(tokenAddedRequest)

        if (response.code == 200) {
            val responseBody = response.peekBody(Long.MAX_VALUE).string()
            val json = JSONObject(responseBody)
            if(json.has("error") && json.getJSONObject("error").has("code")){
                val error = json.getJSONObject("error")
                val code = error.getInt("code")
                if(code == 1003){
                    var refreshToken  = ""
                    tokenManager.refreshToken(onSuccess = {
                        refreshToken = it
                    }, onFailed = {
                        chain.call().cancel()
                    })
                    if(refreshToken.isNotEmpty()){
                        val refreshedRequest = originalRequest.putTokenHeader(refreshToken)
                        return chain.proceed(refreshedRequest)
                    }
                }
            }
        }

        return response
    }

        private fun Request.putTokenHeader(accessToken: String): Request {
        return this.newBuilder()
            .addHeader("token", accessToken)
            .build()
    }
}

please help.

I tried mutex but not working properly.