Combineを理解するための10本ノック

Posted at 2025-02-17

Combineは、Appleが提供するリアクティブプログラミングフレームワークで、非同期イベントの管理やデータのストリーム処理を簡潔に記述できる点が特徴です。以下、基本的な概念について解説します。

1. Publisher（パブリッシャー）

役割:

時間とともに変化するデータやイベントの「発信源」として機能します。

例:

URLSession.DataTaskPublisher（ネットワークリクエスト）、配列やコレクションから作成されるpublisherなど。
特徴: Publisherは値を順次発行し、完了（finished）または失敗（failure）の状態でストリームを終了します。

2. Subscriber（サブスクライバー）

役割:

Publisherから発行された値を受け取って処理する「受信者」です。

実装例:

Combineでは、sinkやassignなどのメソッドを使って簡単にSubscriberを実装できます。

動作:

Subscriberは、発行された各イベントに対して自動的に処理を実行し、必要に応じてエラー処理や完了時の処理も行います。

3. Subscription（サブスクリプション）

役割:

PublisherとSubscriberを接続し、データの流れを管理する「橋渡し役」です。

特徴:

データの需要（Backpressure）に応じた制御が行われるため、受信側の処理能力に合わせてデータが供給されます。
サブスクリプションをキャンセルすることで、ストリームの処理を中断することができます。

4. Operators（オペレーター）

役割:

データの変換、フィルタリング、結合、平坦化（flatten）など、ストリームに対する様々な操作を実行するための関数群です。

主なオペレーター:

map: 各値に対して変換処理を適用
filter: 条件に合致する値だけを通過
flatMap: 入れ子になったPublisherを平坦化して一つのストリームにまとめる
merge/combineLatest/zip: 複数のPublisherからの値を組み合わせる

5. Schedulers（スケジューラー）

役割:

どのスレッドやディスパッチキュー上で処理を実行するかを指定します。

利用例:

バックグラウンドで重い処理を実行し、結果をメインスレッドに戻してUIを更新する、といった用途に利用されます。

主なスケジューラー:

DispatchQueue, RunLoop, OperationQueueなど。

6. 基本的な流れ

Publisherの生成:

データの発信源となるPublisherを作成します。

オペレーターのチェーン:

Publisherに対して、必要な変換やフィルタリングを行うオペレーターを適用します。

Subscriberの設定:

sinkやassignなどを用いて、Publisherが発行するデータを受け取るSubscriberを設定します。

Subscriptionの管理:

必要に応じて、サブスクリプションをキャンセルすることで処理を中断することが可能です。

コード例

1：基本

import Combine

// 1. Publisherの生成: 配列からPublisherを作成
let numbersPublisher = [1, 2, 3, 4, 5].publisher

// 2. オペレーターのチェーン: 各値を2倍にする
let doubledNumbersPublisher = numbersPublisher
    .map { $0 * 2 }

// 3. Subscriberの設定: 値を受け取って出力
let subscription = doubledNumbersPublisher
    .sink(
        // このクロージャは、publisherからの完了イベントを処理します。
        // 値の発行が全て終了した時や、エラーが発生した時に呼び出されます。
        receiveCompletion: { completion in
            switch completion {
            case .finished:
                print("処理完了")
            case .failure(let error):
                print("エラー発生: \(error)")
            }
        },
        
        receiveValue: { value in
            print("受信した値: \(value)")
        }
    )
// 実行結果
// Recieved value: 2
// Recieved value: 4
// Recieved value: 6
// Recieved value: 8
// Recieved value: 10
// 処理完了

解説

この例では、配列 [1, 2, 3, 4, 5] をPublisherに変換し、各要素を2倍にしてから、sink を使って結果を出力しています。

2:Just と sink の基本

import Combine
import Foundation

// Justは1回だけ値を発行し、その後完了するPublisherです
let justPublisher = Just("Hello Combine!")

// sinkで値を受け取り、コンソールに出力します
let subscription = justPublisher
    .sink { value in
        print("Received value: \(value)")
    }
// 実行結果
// Received value: Hello Combine!

解説

Just: 与えた値（ここでは "Hello Combine!"）を1回だけ発行し、直後に完了します。
sink: Publisherからの値を受け取るためのSubscriberです。受け取った値をコンソールに出力しています。

3:Future を使った非同期処理

import Combine
import Foundation

// 非同期処理を行い、結果をFutureで返す関数
func performAsyncOperation() -> Future<String, Never> {
    return Future { promise in
        // 非同期で処理を実行（ここでは1秒後に完了するシミュレーション）
        DispatchQueue.global().asyncAfter(deadline: .now() + 1) {
            promise(.success("Async result"))
        }
    }
}

let futurePublisher = performAsyncOperation()
let subscription = futurePublisher
    .sink { value in
        print("Received async value: \(value)")
    }

解説

Future: 非同期に計算された結果を1回だけ発行するPublisherです。
promise: 非同期処理が完了した際に、成功（または失敗）の結果を通知するために使用します。

この例では、1秒後に "Async result" を発行しています。

4:PassthroughSubject を使った手動の値送信

import Combine
import Foundation

// 任意のタイミングで値を送信できるSubject
let subject = PassthroughSubject<String, Never>()

// sinkで受信
let subscription = subject
    .sink { value in
        print("Subject emitted: \(value)")
    }

// 手動で値を送信
subject.send("First message")
subject.send("Second message")

// 実行結果
// Subject emitted: First message
// Subject emitted: Second message

解説

PassthroughSubject: 外部から値を送信できるPublisherです。
subject.send(...) を呼び出すことで、任意のタイミングで値を発行できます。
ここでは "First message" と "Second message" の2つのメッセージが送信され、sinkで受信されます。

5:オペレーター (filter, map など) の利用

import Combine
import Foundation

// 配列からPublisherを作成
let numbersPublisher = [1, 2, 3, 4, 5].publisher

let subscription = numbersPublisher
    .filter { $0 % 2 == 0 } // 偶数だけを抽出
    .map { $0 * 10 }        // 各値を10倍する
    .sink { value in
        print("Processed value: \(value)")
    }

// 実行結果
// Processed value: 20
// Processed value: 40

解説

filter : 条件に合致する要素だけを次のストリームに流します。
map: 各要素に対して変換処理（ここでは10倍）を実行します。
この例では、元の配列 [1, 2, 3, 4, 5] から偶数（2, 4）を抽出し、それぞれを10倍して20と40が出力されます。

6:URLSession と Combine を使ったネットワークリクエスト

import Combine
import Foundation

// APIから取得するJSONデータに対応するモデル
struct Post: Decodable {
    let userId: Int
    let id: Int
    let title: String
    let body: String
}

let url = URL(string: "https://jsonplaceholder.typicode.com/posts/1")!

let publisher = URLSession.shared.dataTaskPublisher(for: url)
    .map { $0.data } // レスポンスからデータ部分を抽出
    .decode(type: Post.self, decoder: JSONDecoder()) // JSONデコード
    .receive(on: DispatchQueue.main) // メインスレッドで受け取る
    .sink(
        receiveCompletion: { completion in
            switch completion {
            case .finished:
                print("Request finished")
            case .failure(let error):
                print("Error: \(error)")
            }
        },
        receiveValue: { post in
            print("Received post: \(post)")
        }
    )

解説

dataTaskPublisher: 指定したURLから非同期でデータを取得するPublisherです。
map: ネットワークレスポンスから必要なデータ部分（Data）を抽出します。
decode: JSONデータを指定したDecodableな型（ここでは Post）に変換します。
receive(on:): UI更新などのためにメインスレッドで結果を受け取るよう指定します。
sink: ネットワークリクエストの完了状態や、取得した値を処理します。

7:storeとサブスクリプションのライフサイクル

import Combine

class ExampleViewModel {
    // サブスクリプションを保持するためのコンテナ
    private var cancellables = Set<AnyCancellable>()
    
    func fetchData() {
        // 値を発行するPublisherの例（ここではJustを使用）
        let publisher = Just("Hello, Combine!")
        
        // Publisherから値を受け取るサブスクリプションを作成
        publisher
            .sink { completion in
                // 完了時の処理
                switch completion {
                case .finished:
                    print("Completed")
                case .failure(let error):
                    print("Error: \(error)")
                }
            } receiveValue: { value in
                // 値を受け取った際の処理
                print("Received value: \(value)")
            }
            .store(in: &cancellables)  // サブスクリプションをcancellablesに格納
    }
}

解説

cancellables プロパティ
Set 型のプロパティを用意しています。ここに各サブスクリプションの AnyCancellable インスタンスが格納され、ViewModelのライフタイムに合わせて購読が維持されます。
Publisherの作成
Just("Hello, Combine!") を使用して、単一の値を発行し、その後完了するPublisherを作成しています。
receiveCompletion クロージャ
Publisherが完了またはエラーになった際の処理を実装しています。
receiveValue クロージャ
発行された値を受け取り、処理（ここでは単にコンソールに出力）を行っています。
store(in: &cancellables) により、このサブスクリプションを cancellables に格納します。これにより、cancellables が有効な間は購読が維持され、ViewModelが解放されるタイミングで自動的にキャンセルされます。

storeを使うことでサブスクリプションのライフサイクルがパブリッシャーのライフサイクルからstoreのライフサイクルに一致します

8:サブスクリプションのキャンセル

import Foundation
import Combine

// 購読を保持するための変数
var cancellable: AnyCancellable?

// 1秒ごとに時刻を発行するTimerパブリッシャーを作成し、
// 自動接続（autoconnect）で購読を開始
cancellable = Timer.publish(every: 1.0, on: .main, in: .common)
    .autoconnect()
    .sink { currentTime in
        print("Timer fired at: \(currentTime)")
    }

// 5秒後に購読をキャンセルしてライフサイクルを終了させる
DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
    print("Cancelling subscription.")
    cancellable?.cancel()  // 購読をキャンセル
    cancellable = nil      // 参照を解放
}

// コマンドラインツール等で実行する場合、RunLoopを走らせる必要がある
RunLoop.main.run()

解説

Timer.publish(every: on: in:)_: 指定した間隔でイベント（ここでは現在時刻）を発行するパブリッシャーを生成します。

autoconnect(): Publisherへの購読が開始されたと同時に、自動的にタイマーが動き出すようにします。
sink(receiveValue:): 発行された値（現在時刻）を受け取り、コンソールに出力します。
DispatchQueue.main.asyncAfter: 5秒後にキャンセル処理を実行します。
cancel(): 購読を終了し、以降のイベントが発行されなくなります。
RunLoop.main.run(): コマンドラインツール等でメインランループを走らせ、非同期処理が実行されるようにします。

このように、明示的に購読をキャンセルすることで、Combineパイプラインのライフサイクルを終了させることができます。

9:SwiftUIにおけるビューを跨いだ非同期処理

import SwiftUI
import Combine

// MARK: - 遷移元のView（ビューモデルなし）
struct SourceView: View {
    @State private var data: String = "初期データ"
    @State private var cancellable: AnyCancellable?

    var body: some View {
        NavigationView {
            VStack(spacing: 20) {
                Text("ソースビュー")
                    .font(.largeTitle)
                Text(data)
                    .padding()

                NavigationLink(destination: DestinationView()) {
                    Text("遷移する")
                        .padding()
                        .background(Color.blue)
                        .foregroundColor(.white)
                        .cornerRadius(8)
                }
            }
            .padding()
            .navigationTitle("Source")
            .onAppear {
                // 2秒後にデータを更新する非同期処理
                cancellable = Just("ソースから取得したデータ")
                    .delay(for: .seconds(2), scheduler: DispatchQueue.main)
                    .sink { receivedValue in
                        data = receivedValue
                    }
            }
        }
    }
}

// MARK: - 遷移先のView（ビューモデルなし）
struct DestinationView: View {
    @State private var info: String = "読み込み中..."
    @State private var cancellable: AnyCancellable?

    var body: some View {
        VStack(spacing: 20) {
            Text("デスティネーションビュー")
                .font(.largeTitle)
            Text(info)
                .padding()
        }
        .padding()
        .navigationTitle("Destination")
        .onAppear {
            // 3秒後に処理結果を更新する非同期処理
            cancellable = Just("遷移先の非同期処理完了")
                .delay(for: .seconds(3), scheduler: DispatchQueue.main)
                .sink { value in
                    info = value
                }
        }
    }
}

// MARK: - エントリーポイント
struct ContentView: View {
    var body: some View {
        SourceView()
    }
}

@main
struct MyApp: App {
    var body: some Scene {
        WindowGroup {
            ContentView()
        }
    }
}

解説

SourceView
- @Stateでdataを定義し、初期値を設定しています。
  onAppear内でJustパブリッシャーを利用し、2秒後にdataを更新しています。
DestinationView
- 同様に、@Stateでinfoを定義し、onAppear内で3秒後にinfoを更新する非同期処理を実装しています。

10:SwiftUIにおけるデバウンス処理

import SwiftUI
import Combine

// ObservableObjectプロトコルに準拠したViewModelクラス
class ViewModel: ObservableObject {
    // @Publishedプロパティラッパーにより、textの変更が監視対象になる
    @Published var text: String = ""
    
    // Combineの購読（subscription）を管理するためのセット
    private var cancellables = Set<AnyCancellable>()

    init() {
        // $textはtextプロパティのPublisher
        $text
            // 0.5秒のデバウンス処理：連続した入力のうち、最後の入力のみを通知する
            .debounce(for: .seconds(0.5), scheduler: DispatchQueue.main)
            // 値が流れてきたときの処理。ここでは新しいテキストをコンソールに出力する
            .sink { newText in
                print("Text changed: \(newText)")
            }
            // 購読をキャンセル可能なセットに格納
            .store(in: &cancellables)
    }
}

// SwiftUIのViewを定義するContentView
struct ContentView: View {
    // ViewModelのインスタンスをStateObjectとして保持することで、Viewのライフサイクルに合わせて管理する
    @StateObject var viewModel = ViewModel()

    var body: some View {
        // テキストフィールド。入力されたテキストはViewModelのtextプロパティに双方向バインディングされる
        TextField("Type something...", text: $viewModel.text)
            .padding() // テキストフィールドにパディングを追加
    }
}

解説

ViewModelクラスはObservableObjectに準拠しており、@Publishedプロパティラッパーを用いてtextプロパティを宣言しています。これにより、textが変更されると、自動的に更新が通知され、バインディングされたViewが再描画されます。
init内で、$textというPublisherから発行される値に対して、debounceオペレーターを使用しています。debounceは、0.5秒間新たな入力がなければ最後の値を通す仕組みです。これにより、例えばユーザーが素早く入力している場合に不要な処理（この例ではprint文）が連続して呼ばれるのを防ぐことができます。
sinkで値を受け取った後、その購読（subscription）はcancellablesセットに保存されます。これにより、ViewModelが解放される際に購読が自動的にキャンセルされ、メモリリークなどの問題を防ぎます。
ContentViewでは、@StateObjectを用いてViewModelのインスタンスを保持しています。これにより、ViewModelはViewのライフサイクルに沿って正しく管理されます。また、TextFieldはviewModel.textにバインドされており、ユーザーが入力すると即座にViewModelのtextが更新され、その結果、先ほどのCombineパイプラインが動作してデバウンス処理後にprint文が実行されます。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up