如何将AnyPublisher<Void, Error>转换为async throws以实现协议方法？

我有一个协议方法：

func `do`() async throws

（注：do是Swift关键字，定义时需要用反引号包裹）

实现该方法时，必须调用返回值为AnyPublisher<Void, Error>的函数something()：

func something() -> AnyPublisher<Void, Error>

请问do()方法的函数体应如何编写，才能将AnyPublisher<Void, Error>的返回值适配成async throws的形式？当前函数体框架如下：

func `do`() async throws {
    try await ??? something()
}

解决方案 #

方案1：直接使用withCheckedThrowingContinuation实现

这是最直接的写法，无需额外扩展，直接在函数体内处理：

import Combine

func `do`() async throws {
    try await withCheckedThrowingContinuation { continuation in
        something()
            .sink(
                receiveCompletion: { completion in
                    switch completion {
                    case .finished:
                        continuation.resume()
                    case .failure(let error):
                        continuation.resume(throwing: error)
                    }
                },
                receiveValue: { _ in }
            )
            .store(in: &Set<AnyCancellable>())
    }
}

• 必须导入Combine框架才能使用相关API
• Set<AnyCancellable>用于存储订阅对象，避免订阅在任务完成前被提前释放

方案2：给Publisher添加异步扩展（复用性更强）

如果需要在多个地方进行这种转换，可以给Publisher写一个通用扩展：

import Combine

extension Publisher where Output == Void {
    func awaitCompletion() async throws {
        try await withCheckedThrowingContinuation { continuation in
            self.sink(
                receiveCompletion: { completion in
                    switch completion {
                    case .finished:
                        continuation.resume()
                    case .failure(let error):
                        continuation.resume(throwing: error)
                    }
                },
                receiveValue: { _ in }
            )
            .store(in: &Set<AnyCancellable>())
        }
    }
}

之后do()方法就能简化成：

func `do`() async throws {
    try await something().awaitCompletion()
}

核心逻辑说明 #

• withCheckedThrowingContinuation是Swift标准库提供的工具，用于把基于回调/事件的异步操作转换为async/await模式
• 当Publisher发送.finished事件时，调用continuation.resume()表示异步任务成功完成；如果发送.failure错误事件，则通过continuation.resume(throwing: error)抛出错误，适配throws要求

内容的提问来源于stack exchange，提问作者meaning-matters