Combine — 主要概念、Publisher、リアクティブプログラミング

Combineとは？

Combineは、Apple SDKに組み込まれた宣言型リアクティブプログラミングフレームワークです。リアクティブストリームパターンを実装し、Publisherが値を生成し、Subscriberがそれを消費し、オペレータがストリームを変換します。Combineは、コールバックやデリゲートの問題を解決し、ネットワーク応答からUI変更まで、あらゆる非同期イベントに対して統一された構成モデルを提供します。

Combine登場以前、iOS開発者はサードパーティライブラリ、主にRxSwiftを使用していました。AppleはCombineを、エコシステムへの深い統合を備えたネイティブな代替として作成しました。フレームワークは@objcブリッジを介してObjective-Cをサポートし、KVO（NSObject.keyValuePublisherによるKey-Value Observing）やNotificationCenterと連携し、SwiftUIの基盤としても機能します。SwiftUIで公開されるすべてのUIKitコンポーネントは、ビュー更新の内部でCombineを使用しています。

CombineはSwift Concurrencyを考慮して設計されています：iOS 15以降、Publisherは.valuesを介してAsyncSequenceに変換でき、for-await-inループで使用できます。非同期関数からPublisherへの逆変換はFutureを介して行われます。Apple WWDC 2024によると、単発の非同期呼び出しにasync/awaitが導入されたにもかかわらず、CombineはUIKitアプリケーションでのストリーミングデータ処理に推奨されるフレームワークであり続けています。

Combineの基本概念

Combineは3つのプロトコルに基づいています：Publisher（Output型の値を発行、Failure型のエラーで失敗可能）、Subscriber（値を受信、Demandを管理 — 要求された要素の数）、Subscription（キャンセル機能を備えたPublisher-Subscriber接続を表現）。データチャネルはsubscribe時に初期化され、キャンセル、完了、またはエラーで終了します。DemandはCombine独自の概念です：SubscriberはPublisherに、処理可能な要素数を伝え、プロトコルレベルでバックプレッシャーを実装します。

PublisherとSubscriber：リアクティブストリームアーキテクチャ

Publisherは2つの関連型を持つプロトコルです：Output（発行される値の型）とFailure（Errorに準拠するエラー型）。ストリームが失敗しない場合、FailureはNeverとして指定されます — これによりSubscriberはonReceiveがOutputでのみ呼び出されることが保証されます。組み込みPublisherには、Just（単一値）、Sequence（配列）、URLSession.DataTaskPublisher（ネットワークリクエスト）、NotificationCenter.Publisher、@Publishedプロパティラッパーが含まれます。

import Combine

// シーケンスからPublisherを作成
let publisher = [1, 2, 3, 4, 5].publisher

// 値処理を含むSubscriberを作成
class PrintSubscriber: Subscriber {
    typealias Input = Int
    typealias Failure = Never

    func receive(subscription: Subscription) {
        subscription.request(.unlimited)
    }

    func receive(_ input: Int) -> Subscribers.Demand {
        print("Received: \(input)")
        return .unlimited
    }

    func receive(completion: Subscribers.Completion<Never>) {
        print("完了")
    }
}

publisher.subscribe(PrintSubscriber())

SubscriptionとDemand

Subscriptionは、PublisherとSubscriber間のアクティブな接続を表すプロトコルです。Subscriberはreceive(subscription:)メソッドでSubscriptionを受け取り、request(_:)を呼び出してDemandを指定します：.unlimited（すべての値）、.max(N)（制限数）、.none（一時停止）。Demandは動的に変更可能で、Subscriberはデータ受信中に要求する要素数を増減できます。これにより、Publisher側でのバッファリングなしにバックプレッシャーを実現します。

SubjectとCurrentValueSubject

Subjectは、PublisherとSubscriberを組み合わせた型です。SubjectはPublisherとして（サブスクライバーが購読する）と同時にSubscriberとして（値が送信される）使用できます。Combineは2種類のSubjectを提供します：PassthroughSubject（状態を保持せず、新しい値のみを渡す）とCurrentValueSubject（現在の値を保持し、新しいサブスクライバーに渡す）。Subjectは、命令型コードをCombineのリアクティブチェーンに統合するために必要です。

let subject = PassthroughSubject<String, Never>()

// Publisherとして購読
let cancellable = subject
    .map { $0.uppercased() }
    .sink { print($0) }

// Subscriberとして値送信
subject.send("hello")  // "HELLO"を出力
subject.send("world")  // "WORLD"を出力

CurrentValueSubjectは、初期値とvalueプロパティを持つ点でPassthroughSubjectと異なります：サブスクライバーは購読時に即座に現在の値を受け取り、その後すべての後続の更新を受け取ります。CurrentValueSubject.valueは読み書き可能で、valueを変更すると自動的にすべてのサブスクライバーに新しい値が送信されます。これにより、CurrentValueSubjectはMVVMアーキテクチャで状態を表現する理想的な選択肢となります：ViewModelがCurrentValueSubjectを公開し、Viewがsinkを通じて変更を購読します。

両方のSubjectは、send(completion: .finished)またはsend(completion: .failure(error))を呼び出してストリームを終了できます。終了後、Subjectはイベントの受け入れと発行を停止します。終了すべきでない長期間のストリーム（UIイベントなど）には、誤ってsend(completion:)を呼び出さないように、Never Failureと共にPassthroughSubjectを使用することをお勧めします。

変換と結合オペレータ

Combineオペレータは、新しいPublisherを返すPublisherのメソッドです。各オペレータは、アップストリームPublisherに購読し、変換された値をダウンストリームに発行する新しいオブジェクトを作成します。Publisherはジェネリック型であるため、オペレータは厳格な型付けを維持します：mapはOutput<A>をOutput<B>に変換し、tryMapはエラーの可能性を追加します。Combineには約100の組み込みオペレータがあります。

flatMapはCombineではRxSwift版と重要な違いがあります：同じFailure型を持つPublisherを返すクロージャを受け取り、ネストされたPublisherをメインストリームに平坦化します。maxPublishers: .max(1)を指定したflatMapはswitchMapのように動作し、新しい値が到着すると前のネストされたPublisherをキャンセルします。これは検索シナリオで重要です：新しい文字が入力されると、前のHTTPリクエストが自動的にキャンセルされます。

// 前のリクエストをキャンセルするデバウンス検索
searchTextField.textPublisher
    .debounce(for: .seconds(0.3), scheduler: RunLoop.main)
    .removeDuplicates()
    .flatMap(maxPublishers: .max(1)) { query in
        apiService.searchPublisher(query)
            .catch { _ in Just([]) }
    }
    .receive(on: DispatchQueue.main)
    .sink { results in
        self.tableView.reloadData()
    }
    .store(in: &cancellables)

結合オペレータ — combineLatestとzip — はRxSwiftと同様に動作します：combineLatestはいずれかのPublisherが変更されると、すべてのPublisherの最新値のタプルを発行します；zipはインデックスごとに値をペアにします。mergeは同じ型のPublisherを1つのストリームに結合し、順序の保持は保証されません。Combineにはselectもあります — 複数のうち最初に完了するPublisherを選択する稀なオペレータ — やshare — 再実行なしでストリームを複数のサブスクライバーにマルチキャストします。

Schedulerとスレッド管理

SchedulerはCombineにおいて、オペレータの実行コンテキストを定義するプロトコルです。5つ以上の組み込みSchedulerを持つRxSwiftとは異なり、Combineは既存のAppleメカニズムを使用します：DispatchQueue、RunLoop、OperationQueue。これらの各型はSchedulerプロトコルに準拠しており、追加のアダプターなしでreceive(on:)やsubscribe(on:)に直接渡すことができます。

receive(on:)はダウンストリームを指定されたSchedulerに切り替えます — RxSwiftのobserveOnに相当します。receive(on:)以降のすべてのオペレータは指定されたSchedulerで実行されます。subscribe(on:)はアップストリームを切り替え、Publisherの実行に影響します。典型的なパターン：バックグラウンド処理にはsubscribe(on: DispatchQueue.global())、UI更新にはreceive(on: DispatchQueue.main)。SwiftUIで.onReceiveを使用する場合、メインスレッドへの組み込みバインディングは不要ですが、sinkの場合は明示的なreceive(on:).mainが推奨されます。

// バックグラウンドロード + メインスレッドのUI
URLSession.shared.dataTaskPublisher(for: url)
    .subscribe(on: DispatchQueue.global(qos: .background))
    .tryMap { data, response -> Data in
        guard let http = response as? HTTPURLResponse,
              http.statusCode == 200 else {
            throw URLError(.badServerResponse)
        }
        return data
    }
    .receive(on: DispatchQueue.main)
    .decode(type: User.self, decoder: JSONDecoder())
    .sink(receiveCompletion: { print($0) },
         receiveValue: { self.nameLabel.text = $0.name })
    .store(in: &cancellables)

RunLoop.mainはUI操作のためのDispatchQueue.mainの代替です。違いは、RunLoop.mainが現在のアプリケーションイベントループにバインドされているのに対し、DispatchQueue.mainはグローバルなメインスレッドキューにバインドされていることです。UIKitにはDispatchQueue.main、SwiftUIにはRunLoop.mainが推奨されます。ImmediateWhenSchedulerは現在のスレッドで同期的に操作を実行します — テストや単純なPublisherにデフォルトで使用されます。

CombineとSwiftUI：ObservableObjectによる統合

ObservableObjectは、変更を公開するオブジェクトのためのSwiftUIプロトコルです。ObservableObjectを実装するクラスは、@Publishedプロパティラッパーをプロパティに使用でき、その変更がSwiftUIに再描画の必要性を自動的に通知します。内部では、@PublishedはPublisherを作成し、wrappedValueが変更されるとobjectWillChange Publisherに通知します。SwiftUIは@StateObject、@ObservedObject、または@EnvironmentObjectを介してobjectWillChangeを購読します。

@Publishedと@StateObject

@Publishedは、CombineをSwiftUIに統合する最も一般的な方法です。@Publishedプロパティの値が変更されると、SwiftUIはそのオブジェクトを使用するすべてのViewを更新します。@StateObjectはObservableObjectインスタンスを作成し、その変更を購読します。@StateObjectで作成されたViewは、@Publishedプロパティが変更されると自動的に再描画されます。オブジェクトを複数のView間で共有する必要がある場合は、@ObservedObjectまたは@EnvironmentObjectが使用されます。

class UserViewModel: ObservableObject {
    @Published var name: String = ""
    @Published var age: Int = 0
    private var cancellables = Set<AnyCancellable>()

    init() {
        $name
            .debounce(for: .seconds(0.5), scheduler: RunLoop.main)
            .sink { [weak self] newName in
                AnalyticsService.logNameChange(newName)
            }
            .store(in: &cancellables)
    }
}

struct UserView: View {
    @StateObject var viewModel = UserViewModel()

    var body: some View {
        TextField("Name", text: $viewModel.name)
    }
}

AnyCancellableは、購読解除トークンを格納するCancellableの型消去ラッパーです。Set<AnyCancellable>は購読のライフサイクルを管理します：所有者がデイニシャライズされると、すべてのCancellableが自動的にキャンセルされます。SwiftUIプロジェクトでは、Set<AnyCancellable>はObservableObjectクラスで宣言され、購読は.store(in: &cancellables)を介して追加されます。UIKitでも、UIViewControllerでの&cancellablesによる保存や手動のcancel()呼び出しにより、同じメカニズムが使用されます。

Combine vs RxSwift

CombineとRxSwiftは同じリアクティブプログラミングのタスクを解決しますが、根本的なアーキテクチャの違いがあります。CombineはApple SDKの一部でiOS 13までの後方互換性があり、RxSwiftはiOS 8+をサポートするサードパーティライブラリです。CombineはFailureジェネリックによる厳格なエラー型付けを使用し、RxSwiftは単一のError型を使用します。CombineはプラットフォームレベルでSwiftUIと統合されており、RxSwiftはUI拡張にRxCocoaを必要とします。

CombineとRxSwiftの選択はプロジェクト要件に依存します。最小iOSバージョンが13以上でプロジェクトがSwiftUIを使用する場合、Combineは組み込み統合と追加依存関係がないため自然な選択です。プロジェクトがiOS 11-12をサポートする場合、既存のRxSwiftコードベースを含む場合、またはRxSwiftでのみ利用可能な特定のオペレータ（Observable.from(path:)など）が必要な場合、RxSwiftは有効なソリューションであり続けます。