iOS使用Combine处理多异步回调：fetchValue未捕获第二次回调

问题背景 #

在实现内存缓存时，使用Apollo GraphQL结合Combine的Future获取数据，遇到fetchValue的第二次回调无法被myRequest捕获的问题，第二次回调未触发预期的输出。

相关代码 #

myRequest方法

func myRequest<T, Q: GraphQLQuery>(query: Q) -> Future<Response<T>, CYGErrorType>? where T: RootSelectionSet {
    return Future<Response<T>, CYGErrorType> { promise in
        guard let futureFetchValue = self.fetchValue(query: query, cachePolicy: self.model.cachePolicy) as Future<Response<T>, CYGErrorType>? else {
            promise(.failure(.default))
            return
        }
        
        futureFetchValue
            .sink(receiveCompletion: { completion in
                switch completion {
                case .finished:
                    break
                case .failure(let error):
                    promise(.failure(error))
                }
            }, receiveValue: { result in
                print("API>> API>> response")
                promise(.success(result))
            })
            .store(in: &self.cancellable)
    }
}

fetchValue方法

private func fetchValue<T, Query: GraphQLQuery>(query: Query, cachePolicy: CachePolicy) -> Future<Response<T>, CYGErrorType>? where T: RootSelectionSet {
    return Future<Response<T>, CYGErrorType> { promise in
        print("API>> hit")
        apolloClient.fetch(query: query, cachePolicy: cachePolicy) { result in
            switch result {
            case .success(let graphQLResult):
               
                let validateResponse = graphQLResult.isValid(query: query)
                switch validateResponse {
                case .success:
                    guard let data = graphQLResult.data as? T else {
                        promise(.failure(CYGErrorType.default))
                        return
                    }
                    
                    let response = Response(value: data, response: nil, error: nil)
                    promise(.success(response))
                    print("API>> response")

                    
                case .failureWithError(let error):
                    promise(.failure(error))
                }
            case .failure(let error):
                let cygError = self.graphQLErrorParsing(error: error, queryName: Query.operationName)
                promise(.failure(cygError))
            }
        }
    }
}

当前输出 #

API>> hit
API>> response
API>> API>> response
API>> response

预期输出 #

API>> hit
API>> response
API>> API>> response
API>> response
API>> API>> response

问题解答 #

1. 为什么fetchValue无法捕获第二次回调？ #

核心原因是Combine的Future是一次性的：Future的promise只能被调用一次，一旦调用了promise(.success)或promise(.failure)，这个Future就进入完成状态，不会再响应后续的任何事件。

当Apollo的fetch回调第二次触发时，fetchValue里的Future已经因为第一次调用promise(.success)结束，第二次调用promise不会产生任何效果，自然无法被myRequest里的sink捕获到。

另外，myRequest中使用单个AnyCancellable存储订阅，每次调用myRequest都会覆盖之前的订阅，导致旧订阅被取消，这也会影响第二次回调的捕获。

2. 如何确保两次回调都被正确捕获和处理？ #

要支持多回调事件，不能用Future，需改用PassthroughSubject或CurrentValueSubject这类支持多值输出的Publisher类型，具体改造步骤如下：

步骤1：修改fetchValue返回PassthroughSubject

private func fetchValue<T, Query: GraphQLQuery>(query: Query, cachePolicy: CachePolicy) -> PassthroughSubject<Response<T>, CYGErrorType>? where T: RootSelectionSet {
    let subject = PassthroughSubject<Response<T>, CYGErrorType>()
    print("API>> hit")
    apolloClient.fetch(query: query, cachePolicy: cachePolicy) { result in
        switch result {
        case .success(let graphQLResult):
            let validateResponse = graphQLResult.isValid(query: query)
            switch validateResponse {
            case .success:
                guard let data = graphQLResult.data as? T else {
                    subject.send(completion: .failure(CYGErrorType.default))
                    return
                }
                let response = Response(value: data, response: nil, error: nil)
                subject.send(response) // 每次有数据就发送事件
                print("API>> response")
            case .failureWithError(let error):
                subject.send(completion: .failure(error))
            }
        case .failure(let error):
            let cygError = self.graphQLErrorParsing(error: error, queryName: Query.operationName)
            subject.send(completion: .failure(cygError))
        }
    }
    return subject
}

步骤2：修改myRequest并调整订阅存储方式

首先将类中的单个cancellable改为Set类型，避免覆盖订阅：

private var cancellables = Set<AnyCancellable>()

然后修改myRequest方法：

func myRequest<T, Q: GraphQLQuery>(query: Q) -> PassthroughSubject<Response<T>, CYGErrorType>? where T: RootSelectionSet {
    guard let subject = self.fetchValue(query: query, cachePolicy: self.model.cachePolicy) as PassthroughSubject<Response<T>, CYGErrorType>? else {
        let errorSubject = PassthroughSubject<Response<T>, CYGErrorType>()
        errorSubject.send(completion: .failure(.default))
        return errorSubject
    }
    
    subject
        .sink(receiveCompletion: { completion in
            switch completion {
            case .finished:
                break
            case .failure(let error):
                // 可在此处理全局错误逻辑
                break
            }
        }, receiveValue: { result in
            print("API>> API>> response")
        })
        .store(in: &self.cancellables) // 存入Set，保留所有订阅
    
    return subject
}

额外说明

• PassthroughSubject：可以多次发送value事件，直到调用send(completion:)结束，完美匹配Apollo fetch可能多次回调的场景（比如缓存更新后再次返回数据）。
• Set：存储订阅避免被覆盖，确保所有回调都能被处理。
• 若需要保留最新数据供新订阅者获取，可改用CurrentValueSubject，它会保存当前最新值，新订阅者会立即收到该值。

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