RxSwift使用retryWhen实现refreshToken时偶发无ON NEXT即被释放问题

核心根因 #

1. 上游订阅提前销毁
   refreshToken 请求绑定在原始网络请求的 retryWhen 生命周期内，当原始请求的上层订阅（例如页面退出、业务层主动取消请求）调用 dispose 时，会同步取消 retryWhen 内部关联的 refreshToken 订阅，直接触发 isDisposed 回调，不会收到任何 next/error 事件。
2. gRPC 请求未绑定取消逻辑
   _makeUnaryCall 方法返回的 Disposable 为空实现，没有绑定 gRPC UnaryCall 的取消逻辑。当订阅被销毁时，后台 gRPC 请求仍在执行，回调时会触发已销毁的 handler，也会导致链路提前终止。
3. refreshToken 缓存无清理逻辑
   以旧 accessToken 为 key 缓存刷新请求，刷新成功/失败后没有清理对应缓存条目，极端情况下会复用已经终止的 Observable，导致订阅异常。
4. 错误处理逻辑触发提前终止
   refreshToken 内部的 retryWhen 只处理了 GRPCStatus.unavailable 错误，其余错误直接抛出会直接终止刷新序列，同步销毁订阅。

• 修复 gRPC 请求的取消逻辑：

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 {
            call.cancel()
        }
    }
    .subscribeOn(requestScheduler)
}

• 添加缓存清理逻辑，刷新完成/失败后自动清理旧缓存：

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.lock.lock()
        VoidToken.caches.removeValue(forKey: token.accessToken)
        VoidToken.lock.unlock()
    }, afterError: { _ in
        // 错误时也清理缓存
        VoidToken.lock.lock()
        VoidToken.caches.removeValue(forKey: token.accessToken)
        VoidToken.lock.unlock()
    })

• 替换 retryWhen 内的 flatMap 为 concatMap，避免多个请求同时触发 token 过期时的并发问题：

.retryWhen { errorObservable -> Observable<Void> in
    return errorObservable.concatMap { error -> Observable<Void> in
        if let errorCode = error as? NetworkingError, case .tokenExpired(let token) = errorCode {
            return VoidToken.reLogin(token: token!, serverInfo: serverInfo)
        }
        throw error
    }
}

• 若需要完全避免单个请求取消影响全局刷新逻辑，可以将 refreshToken 请求改为全局独立持有，不绑定到单个请求的生命周期。

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