Combine入門: Publisher, Subscriberについてまとめてみた

本記事は、フューチャーアドベントカレンダー2023の15日目です。

前日の14日目は、@yamat2667さんで【Flutter】AR × GPT-V API で食べ物のカロリーに応じたARモデルを表示してみたでした。

はじめに

HealthCare Innovation Group(HIG)¹所属の橋本です。

今回は、iOSアプリ開発でリアクティブプログラミングを実現する方法の一つであるCombineフレームワークの基本的な使い方についてまとめていきます。具体的には、PublisherとSubscriberの定義について、公式ドキュメントを参照しながら、どのような処理をされているか確認していきます。

この記事を読めば、WWDC2019で発表されたCombineフレームワークの基本的な知識を身につけることができるかなと思います。この記事と私が参照した公式ドキュメントも一緒に見ていただけると幸いです。

目次

• Swiftを用いたiOSアプリ開発におけるリアクティブプログラミングの現状
• Appleが提供するCombine Frameworkとは
• Publisherの役割と使い方
  • subject
  • PassthroughSubject
  • CurrentValueSubject
• Subscriberの役割と使い方
  • sink()メソッド
  • assign()メソッド
  • store()メソッド
• さいごに

Swiftを用いたiOSアプリ開発におけるリアクティブプログラミングの現状

まず、リアクティブプログラミングとは、非同期なイベント処理とデータストリームの概念に基づく宣言的なプログラミング手法のことを指します。
現在、Swiftを用いたリアクティブプログラミングでは、以下の３つのライブラリで実装されていることが多いです。

• Combine | Apple標準ライブラリ

• RxSwift | サードパーティ製ライブラリ

• ReactiveSwift | サードパーティ製ライブラリ

Appleが提供するCombine Frameworkとは

Apple公式ドキュメントを見ると、概要は以下の通り。

Combineフレームワークは、時間をかけて値を処理するための宣言的なSwift APIを提供します。これらの値は多くの種類の非同期イベントを表すことができます。Combine は時間と共に変化する値を公開するPublishersと、Publishersからそれらの値を受け取るSubscribersを宣言します。(以下、Appleの公式ドキュメントとより引用。)

The Combine framework provides a declarative Swift API for processing values over time. These values can represent many kinds of asynchronous events. Combine declares publishers to expose values that can change over time, and subscribers to receive those values from the publishers.

つまり、さきほど説明したリアクティブプログラミングを実現するフレームワークの一種であることがわかります。
まず、Combineで重要な項目として、以下の３つがあります。

• Publisher
• Subscriber
• Operator

Publisherとは、名前の通り発行者ということで値などをPublish(発行/生成)する者のことを指します。PublisherがPublishした値などはSubscriberに渡すことができます。つまり、Subscriberは、Publisherの値を受信し、何らかの処理を行う者のことを指します。

文字だけだと分かりづらいので、PublisherとSubscriberを使用した基本的な例を紹介します。ここでは、[1,2,3]という配列を生成し、Subscriberのsink()メソッドでPublisherからの値を受信しています。

基本的な例

import Combine // Combineフレームワークを使うため。

// Publisherを定義
let publisher = [1,2,3].publisher

// sink()メソッドがSubscriberで、Publisherから渡された値を受信する。
publisher.sink(receiveCompletion: { completion in
    print("Received completion:", completion)

}, receiveValue: { value in
    print("Received value:", value)
})

上記のコードの出力結果は以下のようになります。配列の値を一つずつ、sink()メソッドで受信し、receiveValueのクロージャで処理が実行されています。このように、Publisherで値を発行し、Subscriberで値を受信と処理を行っていることがわかりました。

出力結果

Received value: 1
Received value: 2
Received value: 3
Received completion: finished

次の節からは、PublisherとSubscriberについて、もう少し深掘りをしていきます。

Publisherクラスの役割と使い方

Publisherとは、特定の型の値を時間とともに連続的に、Subscriberという値を受信するものに対して、値を送る(Publish/発行する)機能を持つものです。Publishする者なので、Pubilsher のようなイメージです。
Publisherクラスの定義は次の通りです。

Publisherクラスの定義

protocol Publisher {
    associatedtype Output
    associatedtype Failure: Error
    
    func subscribe<S: Subscriber>(_ subscriber: S)
        where S.Input == Output, S.Failure == Failure
}

Publisherには、Outputという出力する値の型とErrorが起こったときに返すError型に準拠したFailure型の２つの連想型が用意されています。次に、subscribe()メソッドのSelf.Failure == S.FailureおよびSelf.Output == S.Inputの条件は、PublisherとSubscriberで型を一致させる必要があることを示しています。基本的には、subscribe()メソッドは明示的に実行する必要はありません。実行タイミングとしては、後ほど紹介するSubscriberのsink()メソッドなどを呼んだときに暗黙的に実行されます。

Publisherが実際に出力する値は、

1. Publisherが出力する値: Output
2. イベント完了を知らせるcompletion: .finished
3. イベントの失敗を知らせるcompletion: .failure(Failure)

の3つがあります。
1の出力値と2.のイベント完了を表す.finishedが呼ばれるパターンのサンプルコードです。

イベント完了を表す.finishedが呼ばれるサンプルコード

// Publisherを定義
let publisher = [1,2,3].publisher　// 配列をpublishする

let subscription = publisher.sink(receiveCompletion: { completion in
    print("Received completion:", completion)

}, receiveValue: { value in
    print("Received value:", value)
})

出力結果

Received value: 1
Received value: 2
Received value: 3
Received completion: finished

配列[1,2,3]の値が一つずつ.sinkメソッドで受信処理が行われていることが、Received value: 1、Received value: 2、Received value: 3と順に出力されていることからわかります。また、すべての値の出力が終えたあとに、receiveCompletionのクロージャが呼ばれ、.finishedが出力されていることが確認できます。

続いて、3.のイベントの失敗を知らせるcompletionが呼ばれるパターンのサンプルコードです。

イベントの失敗を知らせるcompletionが呼ばれるサンプルコード

// Publisherに使用できるように、Error型に準拠したMyErrorを定義
enum MyError: Error {
    case failure
}

// エラーを発生させるPublisherのFail型
let publisher = Fail<Int, Error>(error: MyError.failure)

let subscription = publisher.sink(receiveCompletion: { completion in
    print("Received completion:", completion)

}, receiveValue: { value in
    print("Received value:", value)
})

出力結果

Received completion: failure(__lldb_expr_59.MyError.failure)

この例では、Pubisherにエラーを発生させるFail型のPublisherを定義させることで、receiveCompletionのクロージャでエラーが起こったことを伝える.failureが出力されています。
以上より、Publisherで出力する値は、

1. Publisherが出力する値: Output
2. イベント完了を知らせるcompletion: .finished
3. イベントの失敗を知らせるcompletion: .failure(Failure)

の3つであることが確認できました。

subject

続いて、subjectです。
subjectは、データを発行することとデータを受信ことができるPublisherの一種のオブジェクトです。

Combineでは、

1. PassthroughSubject
2. CurrentValueSubject

の2つのsubjectが提供されています。

PassthroughSubject

PassthroughSubjectとは
データを中継するクラスで、値を保持せずに受け取った要素をSubscriberに受け渡すsubjectです。

PassthroughSubjectの定義は、以下のようになっており、

PassthroughSubjectの定義

final class PassthroughSubject<Output, Failure> where Failure : Error

主な特徴は、

• 初期値を持たない。
• イベントやデータを下流のサブスクライバーにブロードキャストする。
• 複数のサブスクライバーに同時に要素を送信でき、ブロードキャストの機能を提供する。
• パススルー（透過）の特性を持ち、要素を透過的にサブスクライバーに伝える。

です。よく使われる場面としては、イベントを伝える場合などがあります。例えば、ボタンがタップされたイベントをsubscriberに伝えて何らかの処理を実行するときに使えます。

PassthroughSubjectを使ったサンプルコード

// PassthroughSubjectの作成
let subject = PassthroughSubject<Int, Never>()

// PassthroughSubjectに対して購読を行う
subject.sink { value in
    print("Received value: \(value)")
}
    
// PassthroughSubjectに新しい値を発行する
subject.send(1)
subject.send(2)
subject.send(3)

出力結果

Received value: 1
Received value: 2
Received value: 3

PassthroughSubjectの持つ、send()メソッドによって新しい値を発行し、その発行した値をSubscriberのsink()メソッドで受け取ることでクロージャ内の処理が走っていることがわかります。

CurrentValueSubject

CurrentValueSubjectとは
一つの値を保持し、値に変更があったときに新しい値をpublishするSubjectです。

定義は以下のようになっています。

CurrentValueSubjectの定義

final class CurrentValueSubject<Output, Failure> where Failure : Error

CurrentValueSubjectの特徴は、

• 初期値が必要。
• 現在値を保持している。
• Subscribe後にすぐにPublishされた値を受け取る。
  です。使用場面としては、状態を監視するときなどに使うことができます。具体的には、あるアニメーションが再生状態と再生されていない状態を監視する例などがあります。

CurrentValueSubjectを使ったサンプルコード

// CurrentValueSubjectの作成
let subject = PassthroughSubject<Int, Never>(0)

// CurrentValueSubjectに対して購読を行う
subject.sink { value in
    print("Received value: \(value)")
}

// CurrentValueSubjectに新しい値を発行する
subject.send(1) 
subject.send(2)
subject.send(3)

出力結果

Received value: 0 // 初期値を最初に受け取る。
Received value: 1
Received value: 2
Received value: 3

このように、PassthroughSubjectとは異なり、CurrentValueSubjectでは初期値が初めに出力されます。

Subscriberの役割と使い方

Subscriberは、Publisherが出力した値を受信し、処理を行うものです。
Subscriberクラスの定義は次の通りです。

Subscriberクラスの定義

public protocol Subscriber<Input, Failure> : CustomCombineIdentifierConvertible {

    associatedtype Input
    associatedtype Failure : Error

    func receive(subscription: Subscription)
    func receive(_ input: Self.Input) -> Subscribers.Demand
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

Subscriberのジェネリック型<Input,Failure>には、連想型としてInputという受け取る値とErrorが起こったときに返すError型に準拠したFailure型が用意されています。このとき、SubscriberのInputとFailureの型をPublisherのOutputとFailureの型と一致させる必要があります。

receive(subscription:)メソッド: これはSubscriberがPublisherを正常にサブスクライブしたことを通知します。例えば、subscriptionとして、subjectのPassthroughSubjectを渡した場合、print()メソッドでコンソールに出力してみると以下のように出力されます。

print()でコンソールに出力した例

receive subscription: (PassthroughSubject)

receive(_ input: Self.Input) -> Subscribers.Demand: このメソッドは、Publisherが新たに値を生成したときに呼び出されます。また、このメソッドは、返り値としてSubscribers.Demandを返します。これは、Subscriber側から、Publisherから受け取る値の量を調整しています。Demandは、次の３種類で指定することができます。

指定できる3種類のDemand

unlimited　// Publisherで発行された値をあるだけ受け取る。
none Publisherで発行された値を一つも受け取らない。
max(Int) // Publisherで発行された値をIntの数だけ受け取る。

receive(completion:)メソッド: このメソッドは、Publisherがデータの発行を完了したことを通知します。

ここまで紹介したSubscriberの定義を実際のコードで動きを確認してみます。ここでは、subjectのPassthroughSubjectを渡した場合を見てみます。処理の流れを確認するために、print()メソッドを使用します。

処理の流れを確認するためのサンプルコード

let subject = PassthroughSubject<Int, Never>()

subject
    .print() // コンソールに処理の流れをプリントする。
    .sink { value in
    print("received value:", value)
}

subject.send(1)
subject.send(2)
subject.send(completion: .finished)

出力結果

receive subscription: (PassthroughSubject) // ①
request unlimited // ②
receive value: (1)
received value: 1
receive value: (2)
received value: 2
receive finished //　③

① subscriptionとして、PassthroughSubjectが正常に渡されている。
② Subscribers.Demandの量はunlimitedであることがわかる。
③ Subscribers.Completion<Self.Failure>に、finishedが渡されているのでSubscriberが正常に値を受信できたことを示す。

以上がSubscriberの定義とその使い方です。

sinkメソッド

sink()メソッドは、Subscriberを作成し、Publisherからの値を購読します。

sink()メソッドの定義

func sink(receiveValue: @escaping ((Self.Output) -> Void)) -> AnyCancellable

定義の通り、sink()メソッドはAnyCancellable型を返します。このAnyCancellable型は、Cancellableプロトコルに準拠しているため、購読を任意のタイミングでキャンセルさせることができます。

sink()メソッドを使用したサンプルコード

let subject = PassthroughSubject<Int, Never>()
let subscription: Cancellable = subject.sink { value in
    print("Received value: \(value)")
}

subject.send(1)
subject.send(2)
subscription.cancel() // このタイミングでsubscriptionがキャンセルされる。
subject.send(3) 

出力結果

Received value: 1
Received value: 2

出力結果を見ると、subscriptionがキャンセルされたあとに、subject.send(3)の値を受信できていないことが確認できます。このような仕組みを使うことよって、購読が不要になった後に、不要な処理を回避できるようになります。これはメモリ管理やパフォーマンスの向上に繋がります。

AnyCancellableのドキュメント：

assignメソッド

https://developer.apple.com/documentation/combine/fail/assign(to)

Subscriberの一種であるassign()メソッドは、Publisherが値を生成するたびに、指定されたプロパティを設定したい場合に使用します。assign(to:on:)の使い方は次のとおりです。

assignメソッドの使い方

// to: にはKeyPath式でプロパティのvalueを渡す
// on: にはto:で指定したプロパティを持つオブジェクトを渡す
publisher.assign(to: \.value, on: SomeClass())

実際に、ViewModelクラスを作成し、インスタンスプロパティを更新する例を見ていきます。

assignメソッドを使用したサンプルコード

class ViewModel {
    var value: Int = 0 {
        didSet {
            print("value updated： \(value)")
        }
    }
}
let viewModel = ViewModel()
let subject = PassthroughSubject<Int, Never>()

let cancellable = subject.assign(to: \.value, on: viewModel)

subject.send(1)
subject.send(2)
print("value:", viewModel.value) // valueが更新されていることを確認する。
cancellable.cancel() // このタイミングでsubscriptionがキャンセルされる。
subject.send(3)

出力結果

value updated： 1
value updated： 2
value: 2

このように、assign()メソッドをつかうことでインスタンスプロパティ（ここで言うvalue）の値を更新することができます。

storeメソッド

複数のsubscriptionをまとめて保持したいときに、set型としてひとつにまとめる方法があります。

①insertでset型に挿入する方法

let subject = PassthroughSubject<Int, Never>()
var subscriptions = Set<AnyCancellable>()

// ①insertでsetに挿入する
subscriptions.insert(
    subject.sink { value in
        print("[1]received value:", value)
    }
)
subscriptions.insert(
    subject.sink { value in
        print("[2]received value:", value)
    }
)

subject.send(1)
subject.send(2)
subscriptions
subscriptions.forEach {
    $0.cancel()
}
subject.send(3)

出力結果

[1]received value: 1
[2]received value: 1
[1]received value: 2
[2]received value: 2

set型に直接insertするのではなく、store()メソッドを使うことでも同様のことができます。

②store()を使って挿入する方法

let subject = PassthroughSubject<Int, Never>()
var subscriptions = Set<AnyCancellable>()

let anyCancellable = subject.sink { value in
    print("[1]received value:", value)
}

- // ①insertでset型に挿入する方法
- subscriptions.insert(
-     subject.sink { value in
-         print("[1]received value:", value)
-     }
- )
- subscriptions.insert(
-     subject.sink { value in
-         print("[2]received value:", value)
-     }
- )
+ /// ②store()を使って挿入する方法
+ subject
+     .sink { value in
+         print("[1]received value:", value)
+     }
+     .store(in: &subscriptions)
+ subject
+     .sink { value in
+         print("[2]received value:", value)
+     }
+     .store(in: &subscriptions)

subject.send(1)
subject.send(2)
subscriptions
subscriptions.forEach {
    $0.cancel()
}
subject.send(3)

よって、insert()を使わずに、store()メソッドを使うことで直接的にsubscriptionsにルールを設定することができるようになりました。

出力結果

[1]received value: 1
[2]received value: 1
[1]received value: 2
[2]received value: 2

さいごに

今回はCombineフレームワークの中でも、最も重要であるPublisherとSubscriberについて、学習した内容をまとめてみました。今回紹介できていないOpetatorについても近々まとめて記事にしたいと思います。

明日の16日目は @ShimizuJimmy さんの 「LLMと略語の同定」です。

参考文献

1. 医療・ヘルスケア分野での案件や新規ビジネス創出を担う、2020年に誕生した事業部です。設立エピソードは未来報 | 新規事業の立ち上げ　フューチャーの知られざる医療・ヘルスケアへの挑戦の記事をご覧ください。 ↩