axiom-combine-patterns

Combine のパターン

概要

Combine は、UIKit のデリゲート、NotificationCenter のブリッジ、KVO による監視、@Published プロパティなど、巨大な本番コードベースのあらゆる場所に今も組み込まれています。新しいコードでは async/await が優先されますが、既存の Combine パイプラインとの相互運用や保守は日々の作業です。このスキルでは、重要な判断と落とし穴を扱います。Combine と async/await のどちらを使うべきか、メモリーリークを防ぐ方法、そしてこの 2 つのパラダイムを橋渡しする方法について説明します。

基本原則：Combine は廃れていません。成熟しているのです。問題は「Combine を使うべきか」ではなく、「この特定のデータフローには Combine が適切なツールか」です。

このスキルを使う場面

• 既存の Combine パイプラインを扱う場合
• 新しいデータフローに Combine と async/await のどちらを使うか判断する場合
• AnyCancellable のメモリーリークや、パイプラインが何も通知せず失敗する問題をデバッグする場合
• @Published または ObservableObject を使う場合
• Combine のパブリッシャーを async/await のコードと橋渡しする場合
• Subject（PassthroughSubject、CurrentValueSubject）を扱う場合

このスキルを使わない場面

• Timer.publish のパターン → axiom-ios-concurrency 経由で timer-patterns スキルへ（タイマーのライフサイクルを専門に扱います）
• ObservableObject から @Observable への移行 → axiom-swift-concurrency を使用（最新の監視機能）
• UIKit と SwiftUI の橋渡し → axiom-ios-ui 経由で（データフローではなく、ビューのラッピングを扱います）
• 一般的な async/await のパターン → axiom-swift-concurrency を使用

プロンプトの例

• 「ここでは Combine と async/await のどちらを使うべきですか？」
• 「Combine のパイプラインが何も通知せず、値の生成を停止します」
• 「パブリッシャーを非同期シーケンスに変換するにはどうすればよいですか？」
• 「AnyCancellable でリークが発生しています。どこに保存すればよいですか？」
• 「combineLatest と zip の違いは何ですか？」
• 「Combine でテキストフィールドにデバウンスを適用するにはどうすればよいですか？」
• 「@Published プロパティの更新がビューに反映されません」
• 「Combine のパブリッシャーを async/await のコードと橋渡しするにはどうすればよいですか？」

---

パート 1：Combine と async/await の判断フローチャート

ユースケース	Combine	async/await	理由
1 回限りのネットワークリクエスト	いいえ	はい	async/await のほうがシンプルで、キャンセル可能なオブジェクトの管理も不要です
時間の経過とともに値が流れる	はい	AsyncStream	Combine のオペレーター（debounce、combineLatest）のほうが高機能です
ユーザー入力のデバウンス／スロットリング	はい	扱いにくい	Combine には debounce／throttle が組み込まれていますが、AsyncStream では手動で実装する必要があります
複数のソースをマージする	はい	TaskGroup	Combine の merge／combineLatest は、異種ストリームを自然に処理できます
既存の UIKit KVO／Notification	はい	ブリッジ	publisher(for:) と NotificationCenter.default.publisher は、慣用的な Combine の書き方です
iOS 17 以降の新規プロジェクト	いいえ	はい	@Observable と async/await の組み合わせが最新のパターンです
Combine を使っている既存のコードベース	保守する	段階的に移行する	動作しているパイプラインを書き換えず、境界で橋渡ししてください

すばやく判断する

Is it a one-shot operation (network call, file read)?
├─ Yes → async/await (simpler, no cancellable management)
│
Does it need time-based operators (debounce, throttle, delay)?
├─ Yes → Combine (built-in operators, no manual implementation)
│
Are you combining multiple ongoing streams?
├─ Yes → Combine (combineLatest, merge, zip are purpose-built)
│
Is this new code on iOS 17+?
├─ Yes → async/await + @Observable (modern pattern)
│
Is it existing Combine code that works?
└─ Yes → Keep it. Bridge at boundaries when async/await code needs the data.

---

パート 2：パブリッシャー／サブスクライバーのライフサイクル

AnyCancellable の保存ルール

AnyCancellable は、解放されるとサブスクリプションをキャンセルします。保存しない場合、セットアップの直後にパイプラインがキャンセルされます。

❌ パイプラインが即座に停止する

func setupPipeline() {
    publisher
        .sink { value in
            self.handle(value)  // Never called
        }
    // AnyCancellable returned by sink is discarded → subscription cancelled
}

✅ Set に保存する

private var cancellables = Set<AnyCancellable>()

func setupPipeline() {
    publisher
        .sink { [weak self] value in
            self?.handle(value)
        }
        .store(in: &cancellables)
}

Array ではなく Set を使う理由

Set<AnyCancellable> が慣用的な選択肢である理由は次のとおりです。

• store(in:) は Set と RangeReplaceableCollection（Array を含む）の両方で使えますが、慣例では Set を使います
• サブスクリプションの順序は関係ありません
• セットアップが 2 回実行された場合に、誤って重複するのを防ぎます

4 つのメモリーリークパターン

リーク 1：sink 内で self を強参照する

// ❌ LEAK: sink closure captures self strongly
publisher
    .sink { value in
        self.handle(value)  // Strong capture → retain cycle
    }
    .store(in: &cancellables)

// ✅ FIX: weak self
publisher
    .sink { [weak self] value in
        self?.handle(value)
    }
    .store(in: &cancellables)

リーク 2：store(in:) がない

// ❌ LEAK: cancellable assigned to local var, not stored
let cancellable = publisher.sink { handle($0) }
// cancellable deallocated at end of scope → pipeline cancelled

// ✅ FIX: store in instance property
publisher.sink { [weak self] in self?.handle($0) }
    .store(in: &cancellables)

リーク 3：cancellable を過剰に保持する

// ❌ LEAK: cancellables set never cleared, old pipelines accumulate
func refreshData() {
    // Each call adds another subscription without removing the previous one
    dataPublisher
        .sink { [weak self] in self?.update($0) }
        .store(in: &cancellables)
}

// ✅ FIX: clear before re-subscribing
func refreshData() {
    cancellables.removeAll()  // Cancel previous subscriptions
    dataPublisher
        .sink { [weak self] in self?.update($0) }
        .store(in: &cancellables)
}

リーク 4：assign(to:on:)による強参照

assign(to:on:)はon:パラメーターを強参照します。対象がselfの場合、self → cancellables → subscription → selfという保持サイクルが発生します。

// ❌ LEAK: assign(to:on:) retains self strongly — deinit never called
userPublisher
    .map { $0.name }
    .assign(to: \.displayName, on: self)
    .store(in: &cancellables)

// ✅ FIX: use assign(to:) with @Published projected value (iOS 14+)
userPublisher
    .map { $0.name }
    .assign(to: &$displayName)
// No store(in:) needed — subscription tied to @Published property lifetime

主な違い：assign(to: &$prop)はAnyCancellableを返しません。サブスクリプションは内部で管理され、@Publishedプロパティの所有者が解放されるとキャンセルされます。保持サイクルも、キャンセラブルの保存も必要ありません。

iOS 13をサポートする必要がある場合は、代わりに[weak self]を使ったsinkを使用してください。

---

パート 3：必須オペレーター

各グループにつき、代表的な例を1つ紹介します。これらで実際の使用例の90％をカバーできます。

変換

// map: transform each value
publisher.map { $0.name }

// compactMap: transform + filter nil
publisher.compactMap { Int($0) }

// flatMap: one-to-many (each value produces a new publisher)
searchText
    .flatMap { query in
        api.search(query)  // Returns a publisher
    }

flatMapの注意点：.switchToLatest()またはmaxPublishers: .max(1)を指定しないと、flatMapは上流の値ごとに新しい内部パブリッシャーを作成します。入力中の検索には、代わりにmapとswitchToLatestを組み合わせて使用してください。

searchText
    .map { query in api.search(query) }
    .switchToLatest()  // Cancels previous search when new query arrives

複数のソースを結合する

// combineLatest: latest value from each, fires when ANY changes
Publishers.CombineLatest(namePublisher, agePublisher)
    .map { name, age in "\(name), \(age)" }

// merge: interleave values from same-type publishers
Publishers.Merge(localUpdates, remoteUpdates)
    .sink { update in handle(update) }

// zip: pairs values 1:1 (waits for both to produce)
Publishers.Zip(requestA, requestB)
    .sink { responseA, responseB in /* both complete */ }

オペレーター	発火するタイミング	用途
combineLatest	いずれかの入力が変化したとき	フォームのバリデーション（すべてのフィールド）
merge	いずれかの入力が値を生成したとき	イベントストリームの結合
zip	すべての入力が1つの値を生成したとき	同時に完了する必要がある並列リクエスト

時間ベース

// debounce: wait until values stop arriving (search-as-you-type)
searchTextPublisher
    .debounce(for: .milliseconds(300), scheduler: RunLoop.main)
    .sink { [weak self] query in self?.search(query) }
    .store(in: &cancellables)

// throttle: emit at most once per interval (scroll position)
scrollOffsetPublisher
    .throttle(for: .milliseconds(100), scheduler: RunLoop.main, latest: true)
    .sink { [weak self] offset in self?.updateHeader(offset) }
    .store(in: &cancellables)

オペレーター	動作	用途
debounce	入力が止まるまで待ち、その後最後の値を出力する	検索フィールド、自動保存
throttle(latest: true)	一定間隔で最新の値を出力する	スクロールの追跡、センサーデータ
throttle(latest: false)	一定間隔で最初の値を出力する	ボタンタップのレート制限

エラー処理

// tryMap: transform that can throw
publisher.tryMap { data in
    try JSONDecoder().decode(Model.self, from: data)
}

// mapError: convert error types
publisher.mapError { error in
    AppError.network(error)
}

// replaceError: provide fallback value (terminates error path)
publisher.replaceError(with: defaultValue)

// retry: re-subscribe on failure
publisher.retry(3)  // Retry up to 3 times before propagating error

エラー処理の順序が重要です：retryはreplaceErrorより前に配置してください。retryは上流のパブリッシャーを再サブスクライブします。replaceErrorはエラーを終了させ、パイプラインをエラーが発生しない状態にします。

api.fetchData()
    .retry(3)                    // Try 3 more times on failure
    .replaceError(with: cached)  // If all retries fail, use cache
    .sink { data in update(data) }
    .store(in: &cancellables)

flatMapの後にreplaceErrorを置くと、外側のパイプラインが終了します：replaceErrorがflatMapの下流にある場合、内部パブリッシャーのエラーが1つ発生しただけで、個別のリクエストだけでなくパイプライン全体が終了します。エラー処理をflatMapの内部に移動し、各内部パブリッシャーでそれぞれのエラーを処理してください。

// ❌ One API error kills the entire pipeline
$searchText
    .flatMap { query in api.search(query) }
    .replaceError(with: [])  // Pipeline completes on first error
    .sink { ... }

// ✅ Each search handles its own errors independently
$searchText
    .flatMap { query in
        api.search(query)
            .replaceError(with: [])  // Only this search affected
    }
    .sink { ... }

---

パート 4：@Published と ObservableObject

willSet のタイミング

@PublishedはdidSetではなくwillSetでパブリッシャーを発火します。つまり、オブジェクト上のプロパティが実際に設定される前に、サブスクライバーが新しい値を受け取ります。

class ViewModel: ObservableObject {
    @Published var count = 0

    init() {
        $count.sink { newValue in
            // 'newValue' is the incoming value
            // BUT self.count is still the OLD value here
            print("New: \(newValue), Current: \(self.count)")
            // Prints "New: 1, Current: 0" when count is set to 1
        }
        .store(in: &cancellables)
    }
}

プロパティの設定後にその値を読み取る必要がある場合は、$countをサブスクライブしないでください。代わりにdidSetオブザーバーを使用するか、少し遅らせてからself.countを読み取ってください。$パブリッシャーは、変更後の状態を読み取るためではなく、これから適用される値に反応するためのものです。

ネストした ObservableObject の落とし穴

SwiftUIはネストしたObservableObjectの変更を監視しません。ビューの更新をトリガーするのは、最上位オブジェクトのobjectWillChangeだけです。

// ❌ View won't update when settings.theme changes
class AppState: ObservableObject {
    @Published var settings = Settings()  // Settings is also ObservableObject
}

class Settings: ObservableObject {
    @Published var theme = "light"  // Changes here don't propagate
}

// ✅ FIX: Forward objectWillChange manually
class AppState: ObservableObject {
    @Published var settings = Settings()
    private var cancellables = Set<AnyCancellable>()

    init() {
        settings.objectWillChange
            .sink { [weak self] _ in
                self?.objectWillChange.send()
            }
            .store(in: &cancellables)
    }
}

iOS 17以降でのより良い修正方法：ネストした変更を自動的に監視する@Observableへ移行してください。移行パターンについてはaxiom-swift-concurrencyを参照してください。

スレッドセーフに関する警告

@Publishedはスレッドセーフではありません。バックグラウンドスレッドから@Publishedプロパティを設定すると、メインスレッド以外でobjectWillChangeが発火し、SwiftUIのビューがクラッシュすることがあります。

// ❌ CRASH: @Published set from background thread
class ViewModel: ObservableObject {
    @Published var data: [Item] = []

    func fetch() {
        Task {
            let items = await api.fetchItems()
            data = items  // Background thread → crash
        }
    }
}

// ✅ FIX: Ensure main thread
@MainActor
class ViewModel: ObservableObject {
    @Published var data: [Item] = []

    func fetch() {
        Task {
            let items = await api.fetchItems()
            data = items  // Safe — @MainActor ensures main thread
        }
    }
}

---

パート 5：Combine と async/await の橋渡し

Publisher → AsyncSequence

.valuesを使用すると、任意のパブリッシャーを非同期シーケンスとして利用できます。

let cancellable = notificationPublisher
    .sink { notification in handle(notification) }

// ✅ Modern equivalent using .values
for await notification in notificationPublisher.values {
    handle(notification)
}

.valuesに関する注意点：

• for awaitループは、パブリッシャーが完了するかTaskがキャンセルされるまで実行され続けます
• パブリッシャーがスローしたエラーによってループは終了します
• コンシューマーは1つだけです。同じ.valuesを2つのfor awaitループで消費した場合の動作は未定義です

async/await → Publisher

非同期関数をFutureでラップすると、Combineで利用できます。

func fetchUser(id: String) async throws -> User { ... }

// Wrap as a Combine publisher
let userPublisher = Future<User, Error> { promise in
    Task {
        do {
            let user = try await fetchUser(id: "123")
            promise(.success(user))
        } catch {
            promise(.failure(error))
        }
    }
}

Futureは即座に実行されます。サブスクライブされたときではなく、作成時にクロージャーを実行します。遅延実行が必要な場合は、Deferredでラップしてください。

let lazyPublisher = Deferred {
    Future<User, Error> { promise in
        Task {
            do {
                let user = try await fetchUser(id: "123")
                promise(.success(user))
            } catch {
                promise(.failure(error))
            }
        }
    }
}

段階的な移行戦略

動作している Combine コードを書き換えないでください。境界で橋渡しします。

Combine pipeline  →  .values  →  async/await code
                     (bridge)

async function    →  Future   →  Combine pipeline
                     (bridge)

移行の優先順位：

1. 新しいコード：async/await で記述します
2. 境界：.values または Future で橋渡しします
3. 既存の Combine：動作しているパイプラインはそのままにします
4. 書き換え：パイプラインに大幅な変更が必要な場合に限ります

---

パート 6：Subject

PassthroughSubject と CurrentValueSubject

特徴	PassthroughSubject	CurrentValueSubject
初期値	なし	必須
後から登録したサブスクライバー	以前の値を受け取りません	現在の値をすぐに受け取ります
.value プロパティ	なし	あり（現在の値を読み取ります）
用途	イベント（ボタンのタップ、通知）	状態（現在の選択、読み込み状況）

// Event-driven: no initial value, late subscribers miss past events
let taps = PassthroughSubject<Void, Never>()
taps.send()

// State-driven: always has a current value
let isLoading = CurrentValueSubject<Bool, Never>(false)
isLoading.value = true  // Direct access
isLoading.send(false)   // Also works

完了後に送信する際の落とし穴

Subject が完了イベントを受け取ると、その後のすべての send() 呼び出しは黙って無視されます。クラッシュもエラーもなく、ただ何も起きません。

let subject = PassthroughSubject<Int, Never>()

subject.send(1)           // Delivered
subject.send(completion: .finished)
subject.send(2)           // Silently ignored — no crash, no warning

// This is the most common cause of "my pipeline stopped working"

診断：パイプラインが値を生成しなくなった場合、上流から .finished または .failure の完了イベントが送られていないか確認してください。完了すると、パイプラインは機能しなくなります。

---

パート 7：コールドとホットの Publisher（share/multicast）

Combine の Publisher の多くはコールドです。サブスクライブされたときに処理を開始し、各サブスクライバーはそれぞれ独立した処理を実行します。URLSession.dataTaskPublisher はサブスクライバーごとに新しい HTTP リクエストを送信します。

// ❌ Two subscribers = two network requests
let publisher = URLSession.shared
    .dataTaskPublisher(for: url)
    .map(\.data)
    .eraseToAnyPublisher()

publisher.sink { cache.store($0) }.store(in: &cancellables)  // Request 1
publisher.sink { display($0) }.store(in: &cancellables)       // Request 2

share()

.share() はコールドな Publisher をホットにします。最初のサブスクライバーが処理を開始し、後続のサブスクライバーは出力を共有します。

// ✅ One request, shared result
let publisher = URLSession.shared
    .dataTaskPublisher(for: url)
    .map(\.data)
    .share()
    .eraseToAnyPublisher()

publisher.sink { cache.store($0) }.store(in: &cancellables)  // Triggers request
publisher.sink { display($0) }.store(in: &cancellables)       // Shares result

share() の落とし穴

落とし穴	影響	対処法
後から登録したサブスクライバーは値を受け取れません	share() は PassthroughSubject を使用するため、値は再送されません	最初の値が届く前にすべてのサブスクライバーを登録するか、CurrentValueSubject を使った multicast を使用します
サブスクライバーの登録前に上流が完了しました	後から登録したサブスクライバーは値を受け取らず、すぐに .finished を受け取ります	サブスクリプションの順序を確保するか、Combine の外部に結果をキャッシュします
すべてのサブスクライバーがキャンセルされると上流もキャンセルされます	その後に新しいサブスクライバーが登録されると、上流の処理が新たに実行されます	想定された動作ですが、結果がキャッシュされると思い込んでいると意外に感じることがあります

share() を使う場面

Multiple subscribers to the same expensive publisher?
├─ No → Don't use share() (unnecessary complexity)
│
├─ Yes, all subscribe at the same time?
│   └─ Yes → share() works
│
└─ Yes, subscribers attach at different times?
    └─ Use multicast(subject:) with CurrentValueSubject, or cache the result in a property

---

誤った正当化を避ける

考え	実際
「Combine はもう終わりだから、async/await を使えばいい」	Combine に廃止予定の告知はありません。何千もの本番アプリが利用しています。動作しているパイプラインを書き換えると時間を無駄にし、バグを招きます。代わりに段階的に橋渡ししてください。
「どこでも .sink を使えばいい」	[weak self] と適切な store(in:) がなければ、すべての sink がメモリリークの原因になり得ます。パート 2 のライフサイクルルールで、よくある上位 4 つのリークパターンを防げます。
「assign(to:on:) で問題ない。標準 API だから」	on: を強参照するため、ターゲットが self の場合は循環参照になります。代わりに assign(to: &$prop) を使用してください（パート 2、リーク 4）。
「debounce と throttle は同じものだ」	debounce は入力が途切れるまで待ち、throttle は一定間隔で値を出力します。間違った方を使うと、応答が遅れたりイベントを取りこぼしたりします。パート 3 に選択表があります。
「@Published の仕組みはわかっている」	@Published は didSet ではなく willSet で通知します。ネストされた ObservableObject の変更は伝播しません。バックグラウンドスレッドからアクセスするとクラッシュします。パート 4 では、この 3 つの落とし穴をすべて解説します。
「すべてを一度に async/await に移行しよう」	動作している Combine コードを全面的に書き換えると、バグを招き、時間を無駄にします。境界で橋渡ししてください（パート 5）。パイプラインに大幅な変更が必要な場合に限り、書き換えてください。

---

プレッシャーのかかる場面

シナリオ 1：「Combine のコードをすべて async/await に移行しましょう」

セットアップ：技術リードがコードベースのモダン化を望んでいます。「Combine はレガシーです。取り除きましょう。」

プレッシャー：権威とスコープクリープ。データレイヤー全体で Combine パブリッシャー、@Published プロパティ、オペレーターチェーンが使われています。

このスキルで期待される対応：段階的な移行戦略（パート 5）を提案して、性急な変更を避けます。新しいコードには async/await を使います。境界では .values と Future を使います。既存の正常に動作するパイプラインは、変更が必要になるまでそのままにします。全面的な書き換えは、最もコストが高く、メリットが最も少ない選択肢です。

反論のテンプレート：「Combine は非推奨ではありません。Apple は今もすべての SDK に同梱しています。正常に動作しているパイプラインを全面的に書き換えると、今は存在しないバグを持ち込むことになります。境界で橋渡ししましょう。新しいコードには async/await を使い、既存のパブリッシャーの受信には .values を使います。パイプラインの大幅な変更が必要なときだけ、そのパイプラインを書き換えましょう。」

---

シナリオ 2：「パイプラインがいつの間にか止まりました。作り直しましょう」

セットアップ：リファクタリング後、Combine パイプラインが値を生成しなくなりました。クラッシュもエラーもありません。

プレッシャー：時間的なプレッシャーです。「いったん破棄して、作り直しましょう。」

このスキルで期待される対応：作り直す前に原因を診断します。次の点を確認します：（1）上流に完了イベントが送られていないか（完了後の送信、パート 6）。（2）AnyCancellable がまだ有効か（保持ルール、パート 2）。（3）エラーを処理せずにパブリッシャーがエラー終了していないか（replaceError / catch、パート 3）。この 3 つで、パイプラインがいつの間にか失敗する原因の 90% をカバーできます。

診断チェックリスト：

1. AnyCancellable はまだ保持されていますか？（Set がクリアされておらず、解放されていない）
2. 上流で .finished または .failure を送ったものはありますか？
3. エラー処理のない tryMap や、その他のエラーを投げるオペレーターはありますか？
4. 外側のパブリッシャーが完了する状況で switchToLatest を使っていませんか？

反論のテンプレート：「作り直す前に、4 つの点を確認させてください。Cancellable のライフサイクル、上流の完了イベント、未処理のエラー、そして switchToLatest の完了です。ほとんどの場合、原因はこのうちのどれかです。診断なら 5 分で済みますが、作り直してテストするには 30 分かかります。」

---

シナリオ 3：「設定を変更しても UI が更新されません」

セットアップ：設定画面で、ネストされた ObservableObject を使っています。親の AppState が Settings オブジェクトを保持しています。ユーザーが settings.theme を変更しても、UI が更新されません。

プレッシャー：「バインディング単体では動作しているので、SwiftUI のバグに違いありません。objectWillChange.send() で強制的に再描画させましょう。」

このスキルで期待される対応：ネストされた ObservableObject の落とし穴（パート 4）を見抜きます。SwiftUI はネストされた ObservableObject の変更を監視しません。ビューの更新をトリガーするのは、最上位オブジェクトの objectWillChange だけです。修正方法は、ネストされたオブジェクトからの objectWillChange を転送するか、ネストを自動的に処理する @Observable（iOS 17 以降）に移行することです。

このスキルがない場合のアンチパターン：コードのあちこちに objectWillChange.send() を追加する、すべてのネストされたプロパティに @Published を追加する（効果はありません）、または関心の分離を損なってモデルを再構成し、すべてを 1 つのオブジェクトにまとめる、といった対応です。

反論のテンプレート：「SwiftUI が監視するのは最上位の ObservableObject だけです。ネストされたオブジェクトでは、その objectWillChange を親に転送する必要があります。パート 4 に具体的なパターンがあります。SwiftUI のバグではなく、親の init に 5 行追加するだけで直せます。」

---

リソース

WWDC：2019-722、2019-721、2020-10034

ドキュメント：/combine、/combine/anycancellable、/combine/published

スキル：swift-concurrency、memory-debugging