Swift 5.9中如何编写使用可变参数泛型的Publishers.CombineLatest？

核心思路 #

CombineLatest的核心逻辑是：所有上游Publisher都至少发送一次值后，才会将最新的一组值组合输出；之后任何上游发送新值，都会和其他上游的最新值组合发送。结合Swift 5.9的可变参数泛型，我们可以实现支持任意数量上游的版本，核心要解决：

1. 统一约束所有上游的Failure类型一致
2. 类型安全地存储每个上游的最新值、接收状态和完成状态
3. 遍历可变参数元组完成上游订阅与事件转发

完整实现代码 #

import Combine

extension Publishers {
    struct CombineLatest<Publishers: Publisher...>: Publisher where each Publishers.Failure == Publishers.Failure {
        typealias Output = (each Publishers.Output)
        typealias Failure = each Publishers.Failure
        
        private let publishers: Publishers
        
        init(_ publishers: Publishers) {
            self.publishers = publishers
        }
        
        func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
            let subscription = CombineLatestSubscription(
                publishers: publishers,
                subscriber: subscriber
            )
            subscriber.receive(subscription: subscription)
        }
    }
}

private final class CombineLatestSubscription<Publishers: Publisher..., S: Subscriber>: Subscription
where each Publishers.Failure == S.Failure, (each Publishers.Output) == S.Input {
    private var publishers: Publishers?
    private weak var subscriber: S?
    private var cancellables = Set<AnyCancellable>()
    
    // 包装单个上游的状态
    private struct PublisherState<Output> {
        var latestValue: Output?
        var didReceiveValue = false
        var didFinish = false
    }
    
    // 类型擦除的状态容器，解决不同Output类型无法存入同一数组的问题
    private class AnyPublisherState {
        let index: Int
        private(set) var didReceiveValue = false
        private(set) var didFinish = false
        
        init(index: Int) {
            self.index = index
        }
        
        func updateValue<T>(_ value: T) { fatalError("Must override in subclass") }
        func getValue<T>() -> T? { fatalError("Must override in subclass") }
        func markFinished() { didFinish = true }
    }
    
    private final class ConcretePublisherState<Output>: AnyPublisherState {
        var latestValue: Output?
        
        override func updateValue<T>(_ value: T) {
            latestValue = value as? Output
            didReceiveValue = true
        }
        
        override func getValue<T>() -> T? {
            latestValue as? T
        }
    }
    
    private var states: [AnyPublisherState] = []
    
    init(publishers: Publishers, subscriber: S) {
        self.publishers = publishers
        self.subscriber = subscriber
        
        // 遍历可变参数元组，初始化状态并订阅每个上游
        var index = 0
        let mirror = Mirror(reflecting: publishers)
        for child in mirror.children {
            guard let publisher = child.value as? any Publisher else { continue }
            let state = ConcretePublisherState<Any>(index: index)
            states.append(state)
            
            publisher
                .sink(
                    receiveCompletion: { [weak self] completion in
                        guard let self = self else { return }
                        state.markFinished()
                        
                        switch completion {
                        case .failure(let error):
                            self.subscriber?.receive(completion: .failure(error as! S.Failure))
                            self.cancellables.removeAll()
                        case .finished:
                            if self.states.allSatisfy({ $0.didFinish }) {
                                self.subscriber?.receive(completion: .finished)
                            }
                        }
                    },
                    receiveValue: { [weak self] value in
                        guard let self = self else { return }
                        state.updateValue(value)
                        
                        // 所有上游都收到过值后，构建元组并发送
                        guard self.states.allSatisfy({ $0.didReceiveValue }) else { return }
                        let output = self.buildOutputTuple()
                        _ = self.subscriber?.receive(output)
                    }
                )
                .store(in: &cancellables)
            
            index += 1
        }
    }
    
    // 构建类型匹配的输出元组
    private func buildOutputTuple() -> (each Publishers.Output) {
        var output = (each Publishers.Output).init()
        let values = states.map { $0.getValue()! } // 安全：已确认所有上游都有值
        
        withUnsafeMutablePointer(to: &output) { ptr in
            let buffer = UnsafeMutableRawBufferPointer(start: ptr, count: MemoryLayout.size(ofValue: output))
            for (index, value) in values.enumerated() {
                let elementSize = MemoryLayout.size(ofValue: value)
                let elementPtr = buffer.baseAddress!.advanced(by: index * elementSize)
                elementPtr.copyMemory(from: &value, byteCount: elementSize)
            }
        }
        return output
    }
    
    func request(_ demand: Subscribers.Demand) {
        // 简单实现：转发需求给所有上游，可优化为按需请求
        let mirror = Mirror(reflecting: publishers)
        for child in mirror.children {
            guard let publisher = child.value as? any Publisher else { continue }
            publisher.subscribe(AnySubscriber(
                receiveSubscription: { $0.request(demand) },
                receiveValue: { _ in .none },
                receiveCompletion: { _ in }
            ))
        }
    }
    
    func cancel() {
        cancellables.removeAll()
        publishers = nil
        subscriber = nil
        states.removeAll()
    }
}

// 链式调用扩展（可选）
extension Publisher {
    static func combineLatest<Publishers: Publisher...>(_ publishers: Publishers) -> Publishers.CombineLatest<Self, Publishers> where each Publishers.Failure == Failure {
        Publishers.CombineLatest((self, publishers))
    }
}

关键实现细节 #

1. 泛型约束：通过each Publishers.Failure == Publishers.Failure确保所有上游的失败类型一致，输出类型自动推导为所有上游Output组成的元组。
2. 状态管理：用AnyPublisherState做类型擦除，将不同Output类型的上游状态存入同一数组，同时保留类型安全的读写能力。
3. 事件处理：
   • 收到值时更新对应上游状态，全部收到值后构建元组输出
   • 单个上游失败时立即转发失败事件，所有上游完成后才发送完成事件
4. 元组构建：通过unsafe内存操作将数组值转换为对应类型的元组，保证类型匹配且性能可控。

可优化方向 #

• 替换Mirror和unsafe操作：利用Swift 5.9的_VariadicTuple私有协议实现原生可变参数遍历（需注意API兼容性）
• 需求精细化处理：根据订阅者返回的Demand动态调整上游请求量，避免冗余数据
• 线程安全：添加锁保护状态读写，适配多线程场景

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