RxSwift refreshToken使用重试(当：)

Question

我为网络请求令牌刷新编写了这样的代码.

enum NetworkingClient {
    private static func _makeUnaryCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
        call maker: @escaping () -> UnaryCall<Request, Response>
        ) -> Single<Response> {
        return Single<Response>.create { (handler) -> Disposable in
            let call = maker()
            call.response
                .whenComplete {
                    do {
                        handler(.success(try $0.get()))
                    } catch {
                        handler(.error(error))
                    }
                }
            return Disposables.create()
        }
        .subscribeOn(requestScheduler)
    }

    static func makeUnaryCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
        serverInfo: ServerInfo,
        call maker: @escaping () -> UnaryCall<Request, Response>
        ) -> Single<Response> {
        let gid = serverInfo.gid
        let uid = serverInfo.uid
        return self._makeUnaryCall(call: maker)
            .retryWhen { errorObservable -> Observable<Void> in
                return errorObservable.flatMap { error -> Observable<Void> in
                    if let errorCode = error as? NetworkingError, case .tokenExpired(let token) = errorCode {
                        return VoidToken.reLogin(token: token!, serverInfo: serverInfo)
                    }
                    throw error
                }            }
            .do(onError: { (err) in
                log.error("\(gid)-\(uid)-\(serverInfo.host)-\(serverInfo.port)-\(err)")
            })
            .observeOn(MainScheduler.instance)
    }
}

enum VoidToken {
    static var caches: [String: Observable<Void>] = [:]
    static let lock = NSLock()
    static func reLogin(token: Token, serverInfo: ServerInfo) -> Observable<Void> {
        VoidToken.lock.lock()
        if let refreshTokenReq = VoidToken.caches[token.accessToken] {
            VoidToken.lock.unlock()
            return refreshTokenReq
        }
        var service: LoginService? = LoginService(serverInfo: serverInfo)
        let refreshReq = service!
            .refreshToken(refreshToken: token.refreshToken)
            .retryWhen({ errorObservable in
                return errorObservable.flatMap { error -> Observable<Void> in
                    if let err = error as? GRPCStatus, err.code == .unavailable {
                        return .just(())
                    }
                    throw error
                }
            })
            .debug("refreshToken")
            .asObservable()
            .share(replay: 1, scope: .forever)
            .map { _ in }
            .do(afterCompleted: {
                service = nil
            })
        VoidToken.caches[token.accessToken] = refreshReq
        VoidToken.lock.unlock()
        return refreshReq
    }
}

enum NetworkingError: Error {
    case tokenExpired(Token?)
}

extension LoginService {
    func refreshToken(refreshToken: String, loginChatService: Bool = true) -> Single<Void> {
        let gid = self.gid
        let uid = self.uid
        return defaultImplementation
            .reLogin(refreshToken: refreshToken)
            .map { (rsp) -> Token in
                precondition(rsp.hasSess)
                AccountKit.shared.updateToken(pid: gid + uid, session: rsp.sess)
                let token = AccountKit.shared.fetchAuth(gid: gid, uid: uid)!
                log.warn("token refresh success, newToken: \(token.accessToken) newIMToken\(token.imToken)")
                return token
            }
            .flatMap { token -> Single<Void> in
                guard loginChatService else { return Single.just(()) }
                guard let chatService = AccountKit.shared.chatInfos.fetchWrapper(pid: gid + uid) else {
                    throw SourceContextError()
                }
                return chatService.rx.login(imToken: token.imToken, updateSyncTime: false)
                    .map { _ in () }
                    .catchErrorJustReturn(())
            }
    }
}

但是，我得到了一个类似于的日志

2021-07-19 16:24:24.208: refreshToken ->订阅 2021-07-19 16:24:24.494: refreshToken -> isDisposed 2021-07-19 16:24:25.037: refreshToken ->订阅 2021-07-19 16:24:25.251: refreshToken ->事件错误(NotLogin) 2021-07-19 16:24:25.251: refreshToken -> isDisposed

我不知道为什么Refresh Token isDisposed没有ON NEXT

这是个偶然的问题，但我不知道是什么原因造成的

它的日志输出通常应该如下所示：

2021-07-22 10:04:45.794: refreshToken ->订阅 2021-07-22 10:04:52.254: refreshToken -> Event next(()) 2021-07-22 10:04:52.266: refreshToken ->事件完成 2021-07-22 10:04:52.266: refreshToken -> isDisposed

如果你能帮我，我会很感激的。

Answer 1

查看调试语句上面的可观察链：

let refreshReq = service!
    .refreshToken(refreshToken: token.refreshToken)
    .retryWhen({ errorObservable in
        return errorObservable.flatMap { error -> Observable<Void> in
            if let err = error as? GRPCStatus, err.code == .unavailable {
                return .just(())
            }
            throw error
        }
    })

当refreshToken(refreshToken:)或retryWhen发出除GRPCStatus错误以外的任何类型的错误，且code等于.unavailable时，上述都会发出错误。

您说调试显示了一个名为notLogin的错误，它显然不是GRPCStatus错误，代码等于不可用。

另外，查看日志：

2021-07-19 16:24:24.208: refreshToken -> subscribed
2021-07-19 16:24:24.494: refreshToken -> isDisposed
2021-07-19 16:24:25.037: refreshToken -> subscribed
2021-07-19 16:24:25.251: refreshToken -> Event error(notLogin)
2021-07-19 16:24:25.251: refreshToken -> isDisposed

这告诉您，某个东西在订阅后释放了refreshToken一次性286毫秒，然后再订阅(或其他订阅) 543毫秒后，在另外214毫秒后发出错误.

在我看来，问题不在于您发布的代码，而在于过早释放您的reLogin的代码。

顺便说一句，我写了一篇关于这个主题的文章，你可能会觉得有帮助：RxSwift和处理无效令牌