如何将Combine的values属性向下兼容iOS14以支持async/await桥接

iOS 15 新增的Publisher.values本质是将Combine事件流封装为符合AsyncSequence协议的类型，要兼容iOS 14完全可以自行实现等价逻辑，不需要依赖系统新API。

实现逻辑 #

核心基于Swift并发标准库提供的AsyncStream/AsyncThrowingStream做桥接，需要覆盖几个核心能力：

• 绑定Combine订阅与异步序列的生命周期，避免内存泄漏
• 正常传递Publisher输出值、完成事件、错误事件
• 异步迭代取消时，自动取消对应Combine订阅
• 与系统原生API调用方式完全对齐，后续升级iOS15最低部署时无需修改业务代码

桥接代码 #

给Publisher添加扩展，分别实现普通流和抛错流两个版本的values属性：

import Combine
import Foundation

// 适配会抛出错误的Publisher
@available(iOS, deprecated: 15.0, message: "最低部署版本升级到iOS15后可移除该扩展，直接使用系统原生values")
extension Publisher {
    var values: AsyncThrowingStream<Output, Error> {
        AsyncThrowingStream(bufferingPolicy: .unbounded) { continuation in
            var cancellable: AnyCancellable?
            let cleanup: () -> Void = {
                cancellable?.cancel()
                cancellable = nil
            }
            
            continuation.onTermination = { @Sendable _ in
                cleanup()
            }
            
            cancellable = sink(
                receiveCompletion: { completion in
                    switch completion {
                    case .finished:
                        continuation.finish()
                    case .failure(let err):
                        continuation.finish(throwing: err)
                    }
                    cleanup()
                },
                receiveValue: { val in
                    continuation.yield(val)
                }
            )
        }
    }
}

// 适配不会抛出错误（Failure == Never）的Publisher
@available(iOS, deprecated: 15.0, message: "最低部署版本升级到iOS15后可移除该扩展，直接使用系统原生values")
extension Publisher where Failure == Never {
    var values: AsyncStream<Output> {
        AsyncStream(bufferingPolicy: .unbounded) { continuation in
            var cancellable: AnyCancellable?
            let cleanup: () -> Void = {
                cancellable?.cancel()
                cancellable = nil
            }
            
            continuation.onTermination = { @Sendable _ in
                cleanup()
            }
            
            cancellable = sink(
                receiveValue: { val in
                    continuation.yield(val)
                },
                receiveCompletion: { _ in
                    continuation.finish()
                    cleanup()
                }
            )
        }
    }
}

使用方式 #

原有代码不需要做任何修改，和iOS15下调用系统values的写法完全一致：

let publisher = [1, 2, 3, 4, 5]
  .publisher
  .delay(for: .seconds(3), scheduler: RunLoop.main)
  .eraseToAnyPublisher()

// 抛错类型Publisher用try await，Never类型Publisher直接用for await即可
for try await value in publisher.values {
  print(Date())
  print(value)
}

注意事项 #

• 该方案要求使用Xcode 13.2及以上版本（Swift 5.5+ 支持在iOS14等低版本部署Swift并发特性）
• 默认采用无界缓冲策略，和系统原生values行为对齐，如果需要自定义背压策略，可以修改AsyncStream初始化时的bufferingPolicy参数
• 后续项目最低部署版本升级到iOS15后，直接删除上述扩展即可，所有业务调用代码无需改动，会自动切换为系统原生实现

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