swift - 如何使用 Alamofire + RxSwift 同步刷新访问 token

Question

我有这个通用的 fetchData()我的 NetworkManager 类中的函数能够向网络发出授权请求，如果它失败(经过多次重试)会发出一个错误，该错误将重新启动我的应用程序(请求新的登录)。我需要同步调用这个重试 token ，我的意思是，如果多个请求失败，一次应该只有一个请求刷新 token 。如果一个失败，另一个请求必须被丢弃。我已经尝试了一些使用 DispatchGroup/NSRecursiveLock/以及调用描述波纹管的函数 cancelRequests 的方法(在这种情况下，任务计数始终为 0)。我怎样才能使这种行为在这种情况下起作用？

我的网络管理器类:

    public func fetchData<Type: Decodable>(fromApi api: TargetType,
                                           decodeFromKeyPath keyPath: String? = nil) -> Single<Response> {
        
        let request = MultiTarget(api)

        return provider.rx.request(request)
                .asRetriableAuthenticated(target: request)
    }

    func cancelAllRequests(){
        if #available(iOS 9.0, *) {
            DefaultAlamofireManager
                .sharedManager
                .session
                .getAllTasks { (tasks) in
                tasks.forEach{ $0.cancel() }
            }
        } else {
            DefaultAlamofireManager
                .sharedManager
                .session
                .getTasksWithCompletionHandler { (sessionDataTask, uploadData, downloadData) in
                    
                sessionDataTask.forEach { $0.cancel() }
                uploadData.forEach { $0.cancel() }
                downloadData.forEach { $0.cancel() }
            }
        }
    }

使重试有效的 Single 扩展名:

public extension PrimitiveSequence where TraitType == SingleTrait, ElementType == Response {
    
    private var refreshTokenParameters: TokenParameters {
        TokenParameters(clientId: "pdappclient",
                grantType: "refresh_token",
                refreshToken: KeychainManager.shared.refreshToken)
    }

    func retryWithToken(target: MultiTarget) -> Single<E> {
        self.catchError { error -> Single<Response> in
                    if case Moya.MoyaError.statusCode(let response) = error {
                        if self.isTokenExpiredError(error) {
                            return Single.error(error)
                        } else {
                            return self.parseError(response: response)
                        }
                    }
                    return Single.error(error)
                }
                .retryToken(target: target)
                .catchError { error -> Single<Response> in
                    if case Moya.MoyaError.statusCode(let response) = error {
                        return self.parseError(response: response)
                    }
                    return Single.error(InvalidGrantException())
                }
    }

    private func retryToken(target: MultiTarget) -> Single<E> {
        let maxRetries = 1
        return self.retryWhen({ error in
            error
                    .enumerated()
                    .flatMap { (attempt, error) -> Observable<Int> in
                        if attempt >= maxRetries {
                            return Observable.error(error)
                        }
                        if self.isTokenExpiredError(error) {
                            return Observable<Int>.just(attempt + 1)
                        }
                        return Observable.error(error)
                    }
                    .flatMap { _ -> Single<TokenResponse> in
                        self.refreshTokenRequest()
                    }
                    .share()
                    .asObservable()
        })
    }
    
    private func refreshTokenRequest() -> Single<TokenResponse> {
        return NetworkManager.shared.fetchData(fromApi: IdentityServerAPI
            .token(parameters: self.refreshTokenParameters)).do(onSuccess: { tokenResponse in
                    
            KeychainManager.shared.accessToken = tokenResponse.accessToken
            KeychainManager.shared.refreshToken = tokenResponse.refreshToken
        }, onError: { error in
            NetworkManager.shared.cancelAllRequests()
        })
    }

    func parseError<E>(response: Response) -> Single<E> {
        if response.statusCode == 401 {
            // TODO
        }

        let decoder = JSONDecoder()
        if let errors = try? response.map([BaseResponseError].self, atKeyPath: "errors", using: decoder,
                failsOnEmptyData: true) {
            return Single.error(BaseAPIErrorResponse(errors: errors))
        }

        return Single.error(APIError2.unknown)
    }

    func isTokenExpiredError(_ error: Error) -> Bool {
        if let moyaError = error as? MoyaError {
            switch moyaError {
            case .statusCode(let response):
                if response.statusCode != 401 {
                    return false
                } else if response.data.count == 0 {
                    return true
                }
            default:
                break
            }
        }
        return false
    }

    func filterUnauthorized() -> Single<E> {
        flatMap { (response) -> Single<E> in
            if 200...299 ~= response.statusCode {
                return Single.just(response)
            } else if response.statusCode == 404 {
                return Single.just(response)
            } else {
                return Single.error(MoyaError.statusCode(response))
            }
        }
    }

    func asRetriableAuthenticated(target: MultiTarget) -> Single<Element> {
        filterUnauthorized()
                .retryWithToken(target: target)
                .filterStatusCode()
    }

    func filterStatusCode() -> Single<E> {
        flatMap { (response) -> Single<E> in
            if 200...299 ~= response.statusCode {
                return Single.just(response)
            } else {
                return self.parseError(response: response)
            }
        }
    }
}

Answer 1

这是一个 RxSwift 解决方案:RxSwift 和 Handling Invalid Tokens
仅仅发布链接并不是最好的，所以我也会发布解决方案的核心:
关键是要创建一个与 ActivityMonitor 类非常相似但处理 token 刷新的类...

public final class TokenAcquisitionService<T> {

    /// responds with the current token immediatly and emits a new token whenver a new one is aquired. You can, for example, subscribe to it in order to save the token as it's updated.
    public var token: Observable<T> {
        return _token.asObservable()
    }

    public typealias GetToken = (T) -> Observable<(response: HTTPURLResponse, data: Data)>

    /// Creates a `TokenAcquisitionService` object that will store the most recent authorization token acquired and will acquire new ones as needed.
    ///
    /// - Parameters:
    ///   - initialToken: The token the service should start with. Provide a token from storage or an empty string (object represting a missing token) if one has not been aquired yet.
    ///   - getToken: A function responsable for aquiring new tokens when needed.
    ///   - extractToken: A function that can extract a token from the data returned by `getToken`.
    public init(initialToken: T, getToken: @escaping GetToken, extractToken: @escaping (Data) throws -> T) {
        relay
            .flatMapFirst { getToken($0) }
            .map { (urlResponse) -> T in
                guard urlResponse.response.statusCode / 100 == 2 else { throw TokenAcquisitionError.refusedToken(response: urlResponse.response, data: urlResponse.data) }
                return try extractToken(urlResponse.data)
            }
            .startWith(initialToken)
            .subscribe(_token)
            .disposed(by: disposeBag)
    }

    /// Allows the token to be set imperativly if necessary.
    /// - Parameter token: The new token the service should use. It will immediatly be emitted to any subscribers to the service.
    func setToken(_ token: T) {
        lock.lock()
        _token.onNext(token)
        lock.unlock()
    }

    /// Monitors the source for `.unauthorized` error events and passes all other errors on. When an `.unauthorized` error is seen, `self` will get a new token and emit a signal that it's safe to retry the request.
    ///
    /// - Parameter source: An `Observable` (or like type) that emits errors.
    /// - Returns: A trigger that will emit when it's safe to retry the request.
    func trackErrors<O: ObservableConvertibleType>(for source: O) -> Observable<Void> where O.Element == Error {
        let lock = self.lock
        let relay = self.relay
        let error = source
            .asObservable()
            .map { error in
                guard (error as? TokenAcquisitionError) == .unauthorized else { throw error }
            }
            .flatMap { [unowned self] in  self.token }
            .do(onNext: {
                lock.lock()
                relay.onNext($0)
                lock.unlock()
            })
            .filter { _ in false }
            .map { _ in }

        return Observable.merge(token.skip(1).map { _ in }, error)
    }

    private let _token = ReplaySubject<T>.create(bufferSize: 1)
    private let relay = PublishSubject<T>()
    private let lock = NSRecursiveLock()
    private let disposeBag = DisposeBag()
}

extension ObservableConvertibleType where Element == Error {

    /// Monitors self for `.unauthorized` error events and passes all other errors on. When an `.unauthorized` error is seen, the `service` will get a new token and emit a signal that it's safe to retry the request.
    ///
    /// - Parameter service: A `TokenAcquisitionService` object that is being used to store the auth token for the request.
    /// - Returns: A trigger that will emit when it's safe to retry the request.
    public func renewToken<T>(with service: TokenAcquisitionService<T>) -> Observable<Void> {
        return service.trackErrors(for: self)
    }
}

将上述内容放入应用程序后，您只需添加 .retryWhen { $0.renewToken(with: tokenAcquisitionService) }到您的请求结束。确保您的请求发出 ResponseError.unauthorized如果 token 未经授权并且服务将处理重试。