Combine PublisherをSwift Concurrencyに変換する

結論

Publisher→AsyncSequence→Swift Concurrencyの順番に変換してawaitで処理を行えるようにします。

概要

CombineのPublisherをSwift Concurrency(await)として扱いたい場合、以下のようなextensionを作成することにより、valueとして値を取得することができます。

それぞれ、エラーを返す場合とエラーを返さないNever向けのextension valueのプロパティを作成しています。
valuesは元々ある、AsyncSequenceプロトコル適合の AsyncPublisher or AsyncThrowingPublisher です。
詳しい説明はこちら。
最終的にAsyncSequenceの最初の値を返すことでPublisherのSwift Concurrency変換を行います。

extension Publisher where Failure == Never {
    var value: Output {
        get async {
            await self
                .first()
                .values
                .first { _ in true }!
        }
    }
}

extension Publisher {
    var value: Output {
        get async throws {
            try await self
                .first()
                .values
                .first { _ in true }!
        }
    }
}

用例

URLSessionに、Publisherを返してくれるdataTaskPublisherがあるので、Qiita APIからタグを取得する例をあげます。
Publisherに備わっているmapやdecodeを利用しつつ、Swift Concurrencyとして処理が可能です。

struct Tag: Decodable {
    let followers_count: Int
    let icon_url: String?
    let id: String
    let items_count: Int
}

loadWithError.swift

    func loadWithError() async throws -> [Tag] {
        try await URLSession.shared.dataTaskPublisher(for: URL(string: "https://qiita.com/api/v2/tags?sort=count")!)
            .map(\.data)
            .decode(type: [Tag].self, decoder: JSONDecoder())
            .value
    }

load.swift

    func load() async -> [Tag] {
        await URLSession.shared.dataTaskPublisher(for: URL(string: "https://qiita.com/api/v2/tags?sort=count")!)
            .map(\.data)
            .decode(type: [Tag].self, decoder: JSONDecoder())
            .replaceError(with: [])
            .value
    }