RxSwift Auth令牌刷新流程重复请求问题排查求助

构建RxSwift Auth令牌刷新服务时遇到异常：当用户无有效Auth Token，首次刷新请求失败、二次刷新成功后，会额外发起第三次刷新请求，待其完成才调用主接口。

实际网络请求序列 #

• 令牌刷新请求（失败）
• 令牌刷新请求（成功）
• 令牌刷新请求（成功）
• 主接口请求（成功）

预期请求序列 #

• 令牌刷新请求（失败）
• 令牌刷新请求（成功）
• 主接口请求（成功）

Authenticator实现代码 #

protocol AuthenticatorType {
    func authenticate() -> Observable<Void>
    func checkForValidAuthTokenOrRefresh(forceRefresh: Bool) -> Observable<Void>
}

extension AuthenticatorType {
    func checkForValidAuthTokenOrRefresh(forceRefresh: Bool = false) -> Observable<Void> {
        return checkForValidAuthTokenOrRefresh(forceRefresh: forceRefresh)
    }
}

final class Authenticator<Provider: RxMoyaProviderType> where Provider.Target == AuthAPI {
    private let provider: Provider
    private let cookiesStorageProvider: CookiesStorageProviderType
    private let queue = DispatchQueue(label: "Autenticator.\(UUID().uuidString)")
    
    private var refreshInProgressObservable: Observable<Void>?
    
    init(
        provider: Provider,
        cookiesStorageProvider: CookiesStorageProviderType
    ) {
        self.provider = provider
        self.cookiesStorageProvider = cookiesStorageProvider
    }
    
    func checkForValidAuthTokenOrRefresh(forceRefresh: Bool = false) -> Observable<Void> {
        return queue.sync { [weak self] in
            self?.getCurrentTokenOrRefreshIfNeeded(forceRefresh: forceRefresh) ?? .just(())
        }
    }
    
    func authenticate() -> Observable<Void> {
        provider.request(.authenticate(credentials: .defaultDebugAccount))
            .map(LoginResponse.self)
            .map { loginResponse in
                guard loginResponse.login else {
                    throw AuthenticationError.loginRequired
                }
            }
            .asObservable()
    }
}

// MARK: - Helper methods
private extension Authenticator {
    func getCurrentTokenOrRefreshIfNeeded(forceRefresh: Bool = false) -> Observable<Void> {
        if let refreshInProgress = refreshInProgressObservable {
            return refreshInProgress
        }
        
        if cookiesStorageProvider.isHaveValidAuthToken && !forceRefresh {
            return .just(())
        }
        
        guard cookiesStorageProvider.isHaveValidRefreshToken else {
            return .error(AuthenticationError.loginRequired)
        }
        
        let refreshInProgress = provider.request(.refreshToken)
            .share()
            .map { response in
                guard response.statusCode != 401 else {
                    throw AuthenticationError.loginRequired
                }
                return response
            }
            .map(RefreshReponse.self)
            .map { refreshResponse in
                guard refreshResponse.refresh else {
                    throw AuthenticationError.loginRequired
                }
            }
            .asObservable()
            .do(
                onNext: { [weak self] _ in self?.resetProgress() },
                onError: { [weak self] _ in self?.resetProgress() }
            )
        
        refreshInProgressObservable = refreshInProgress
        return refreshInProgress
    }
    
    func resetProgress() {
        queue.sync { [weak self] in
            self?.refreshInProgressObservable = nil
        }
    }
}

请求封装逻辑代码 #

func request(_ token: Target, callbackQueue: DispatchQueue?) -> Observable<Response> {
    authenticator.checkForValidAuthTokenOrRefresh()
        .flatMapLatest { [weak self] res -> Observable<Response> in
            self?.provider.request(token).asObservable() ?? .empty()
        }
        .map { response in
            guard response.statusCode != 401 else {
                throw AuthenticationError.loginRequired
            }
            return response
        }
        .retry { [weak self] error in
            error.flatMap { error -> Observable<Void> in
                guard let authError = error as? AuthenticationError, authError == .loginRequired else {
                    return .error(error)
                }
                
                return self?.authenticator.checkForValidAuthTokenOrRefresh(forceRefresh: true) ?? .never()
            }
        }
}

已尝试的解决方案：

• 用NSLock替代队列
• 调整subscribe/observe调度
• 修改do块时机
• 移除share()操作符

以上方法均未解决问题。后续尝试用Observable.deferred实现刷新逻辑，单请求场景正常，但多请求下刷新失败后无重试。最终添加.observe(on: SerialDispatchQueueScheduler(queue: queue, internalSerialQueueName: "test1"))调度后问题解决，但无法理解其原理，特此求助。

内容的提问来源于stack exchange，提问作者Richard