JavaScript の EventEmitter から繋がる Rx の世界を Swift で見通す

導入

この記事では JavaScript の EventEmitter から出発して、それをバインディングパターンを見据えて改良していく過程で Rx になる話を書くことで、両者が近縁な存在である事を示します。実装は Swift で提示しますが、JavaScript と Swift のどちらかがわかれば理解できるように書きます。コードは特に断りがない限りは Swift　です。

Swift で EventEmitter

JavaScript には EventEmitter という有名なパターンがあります。これはいろいろな場面で便利なので、 Swift でも実装して利用する事が考えられます。

まずは簡単に EventEmitter を説明します。

これの原型はおそらく、 ブラウザ向けで window.addEventListener として提供されているAPI です。

使い方の例を示します。

// JavaScript
window.addEventListener("keydown", function(event) {
    console.log("keydown:", event.key);
    });

このように、 addEventListener メソッドは、第1引数に文字列でイベント名を与えて、第2引数にそのハンドラを書きます。この例では、 "keydown" を与えているので、ハンドラには KeyboardEvent オブジェクトが渡されるため、 event.key で押されたキーを取得できます。

なお、 JS ではハンドラは event listener と呼ばれていますが、この記事では好みの都合でリスナではなくハンドラと呼びます。

その後、Server Side JavaScript 環境の NodeJS にて EventEmitter という標準ライブラリ が提供されました。

先程の addEventListener メソッドが on という名前に変更されています。その他、イベントを送信するための emit メソッド、一度登録したハンドラを登録解除するための removeListener メソッドなどがあります。

さらにその後、これがブラウザでも使えるように逆輸入され、ライブラリ化されました。ここでは EventEmitter3 をあげておきます。

これらの利用時には、あるオブジェクトが発生させるイベントは、この on メソッドで全て拾えるようになっていて、 メソッドの第1引数のイベント名で拾い分けます。

さて、このインターフェースを素直に Swift で書くと以下のようになります。実装部は省略しています。

class EventEmitter {
    func on(eventName: String, handler: @escaping ([Any]) -> Void)
    func emit(eventName: String, args: [Any])

    func removeHandler(eventName: String, handler: @escaping ([Any]) -> Void)
}

利用側のコードは以下のようになります。

window.on(eventName: "click") { _ in 
    print("click") 
}
window.on(eventName: "mousedown") { events in 
    print("mousedown", (events[0] as! MouseEvent).button) 
}
window.on(eventName: "keydown") { events in 
    print("keydown", (events[0] as! KeyboardEvent).key) 
}

これにはいくつかの問題があります。

イベント種類での型分割

メソッドの型定義に Any 型の配列が出現しているのが残念です。利用側のコードでは、イベントが [Any] で渡ってくるので、0番要素へのアクセスと as! によるダウンキャストが必要です。これは記述も面倒ですし、イベント名を間違えたり typo してしまうと、入っているイベントオブジェクトの型が違ったり配列が空になったりして、クラッシュする恐れがあります。

これは複数の種類のイベントをまとめて1つの EventEmitter で扱っているからです。JavaScript の目線で見れば、本来的にはいずれのメソッドでも、 eventName と連動して handler の引数の型は決まるため、ハンドラの型が決定できますが、 Swift の型システムではそのような他の引数によって決まる型を表現できません。

これを解決するために、1つの EventEmitter が複数のイベントを取り扱うのではなく、1つの EventEmitter が 1つの種類のイベントだけを取り扱うように型を分割します。すると、インターフェースが以下のように変更されます。

class EventEmitter<Event> {
    func on(handler: @escaping (Event) -> Void)
    func emit(_ args: Event)

    func removeHandler(_ handler: @escaping (Event) -> Void)
}

ジェネリッククラスにして、イベントの型を型パラメータにしました。

すると、利用側のコードは以下のように変化します。

window.clickEvent.on { _ in
    print("click")
}
window.mouseDownEvent.on { event in
    print("mousedown", event.button)
}
window.keyDownEvent.on { event in
    print("keydown", event.key)
}

イベントが個別のプロパティになっているので、 on から受け取る型が静的に解決していて、ダウンキャストなどが不要になっています。また、プロパティ名は静的にチェックされるので、イベント名をtypoする事もありません。

removeHandler を on に統合して subscribe にする

現状のインターフェースでは、 on で登録したハンドラを removeHandler で登録解除するようになっていますが、これは正しく動きません。 Swift ではクロージャの同一性が判定できないので、 removeHandler の内部において on で渡されたハンドラを検索することすらできないからです。

これの対応策の一つは、クロージャを自前で boxing することです。以下のようにラップする型を用意すれば、内部で同一性判定ができます。以下に例を示します。

class EventHandler<Event> {
    init(_ f: @escaping (Event) -> Void)
    func send(_ event: Event)
}

class EventEmitter<Event> {
    func on(handler: EventHandler<Event>)
    func emit(_ event: Event)

    func removeHandler(_ handler: EventHandler<Event>)
}

しかしこの方法は on などを呼び出すたびに EventHandler で包む必要があって面倒です。

これを解決する方法として、 on メソッドを呼び出した時点で、その返り値として登録解除用のオブジェクトを返す、という設計があります。これは Rx で用いられているアイデアなので on を subscribe に名前を変更します。登録解除用の型は Disposer とします。すると、 EventEmitter は以下のようになります。

class Disposer {
    func dispose()
}
class EventEmitter<Event> {
    func subscribe(handler: @escaping (Event) -> Void) -> Disposer
    func emit(_ event: Event)
}

EventEmitter とバインディングパターン

上記の EventEmitter でそれなりにイベント通知パターンには使えると思います。 once メソッドなども、簡単に extension で実装できるでしょう。ここからは、イベント通知ではなく、バインディングに利用範囲を広げることを考えていきます。

ある実用上の課題を考えます。あるオブジェクトがあって、その状態変化に応じて動作する別のオブジェクトを作りたいとします。例えば、 OpenGL で球を 3D CG としてレンダリングする事を考えます。球の情報を持つオブジェクトを用意し、それをレンダリングするための頂点メッシュ情報を構築します。球のプロパティが変化した際には、頂点メッシュを更新する事を考えます。このような場合、それぞれ Sphere と SphereRenderer として、以下のように定義できるでしょう。

class Sphere {
    var radius: Float
    var color: Color
}

class SphereRenderer {
    init(sphere: Sphere)
    var sphere: Sphere
}

この場合、 SphereRenderer においては、新しい sphere がセットされた際に GPU リソースを更新する必要があります。また、 sphere オブジェクトの radius や color といったプロパティが変化した際にもそれが必要になります。これらのプロパティの変化に際して、 SphereRenderer は共通の再構築処理を呼び出す場合もあれば、それぞれのプロパティに応じた更新処理を呼び出す場合もあります。

このような、オブジェクトの変化と、それに応じた処理を行うオブジェクトを別々に設計してから繋ぎこむパターンは、一般にバインディングと呼ばれていると思います。バインディングでは、値が変化したときに何か処理をする必要があるので、値が変化した事をイベントとして通知するように EventEmitter を用いて実装できます。そのときにコードがどのようになるか見ていきます。

まず、共通の再構築処理を呼び出す場合を見てみます。

class Sphere {
    init() {}

    var radius: Float = 0 {
        didSet {
            radiusChangeEvent.emit(radius)
        }
    }
    let radiusChangeEvent: EventEmitter<Float> = .init()

    var color: Color = [] {
        didSet {
            colorChangeEvent.emit(color)
        }
    }
    let colorChangeEvent: EventEmitter<Color> = .init()
}

class SphereRenderer {
    init(sphere: Sphere) {
        self.sphere = sphere

        reloadSphere()
    }

    deinit {
        disposers.forEach { $0.dispose() }
        disposers = []
    }

    var sphere: Sphere = .init() {
        didSet {
            disposers.forEach { $0.dispose() }
            disposers = []

            rebuildMesh()

            unowned let uself = self

            var disposer = sphere.radiusChangeEvent.subscribe { _ in
                uself.rebuildMesh()
            }
            disposers.append(disposer)

            disposer = sphere.colorChangeEvent.subscribe { _ in
                uself.rebuildMesh()
            }
            disposers.append(disposer)
        }
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    private func reloadSphere() {
        self.sphere = { self.sphere }()
    }

    private var disposers: [Disposer] = []
}

まず、 Sphere はプロパティに加えて、それの変更イベントを通知する EventEmitter を保持します。そして、それぞれのプロパティの didSet でイベントを emit して送出します。 SphereRenderer では、 sphere プロパティの didSet でバインディングを行います。このイベント購読は、 sphere がセットされるたびに登録されるので、 新しい sphere が来たときと、自身が解放される際には解除する必要があります。その購読解除の処理が、 deinit と didSet の冒頭に書いてあります。そして、渡された sphere に応じて即座に rebuildMesh を呼び出します。 さらに sphere の radius と color の変化に対して、これも rebuildMesh を呼び出すようにします。その subscribe で返ってきた disposer は、あとで解除できるよう disposers に入れておきます。変更イベントからは現在値も渡ってきますが、 rebuildMesh の中で sphere を参照すれば良いので読み捨てています。
init では sphere の初期値が渡せるようになっていますが、初期化代入の際には didSet は呼ばれないので、 init の末尾で reloadSphere を呼び出して、その中で didSet を発生させています。

初期化代入時に didSet が呼ばれないのは Swift の問題と思うかもしれませんが、もしオブジェクトの初期化が終わってない状態でのメソッド呼び出しを許してしまうと、不完全な self にアクセスしてしまう可能性が生じて安全性が損なわれてしまうので、それを防ぐための正しい仕様です。
reloadSphere を定義しているのは、 init の実行中でもメソッド経由の代入では didSet が呼び出されるからです。 reloadSphere の内部で右辺をクロージャに包んでいるのは、これがないとコンパイラが無意味な代入としてエラー扱いするので、ワークアラウンドです。

radiusChangeEvent と colorChangeEvent の subscribe メソッドに渡すクロージャでは、 self を unowned な uself という変数を経由してキャプチャしています。これは SphereRenderer を循環参照してしまうのを防ぐためです。 SphereRenderer は deinit で購読を解除するので、 self 解放後にクロージャが呼び出されることは無く、クロージャが呼び出されるときには uself は有効であるため、この unowned は安全です。

やりたいことに対して結構複雑なコードになってしまっています。以下に問題点を挙げます。

Sphere 側のイベント発行部分において、 プロパティとイベントを紐付けているのは名前付けだけです。プロパティの didSet で正しい EventEmitter を指定しなければバグってしまいますし、書き忘れればイベントが発行されません。

SphereRenderer では disposers を解放するための処理が 2箇所に書かれています。プロパティごとに disposer を保持する処理が 2箇所で必要です。 rebuildMesh の呼び出しは 3箇所もあります。

reloadSphere は都合で必要とはいえ煩わしいです。

これらを改善していきます。

DisposerBag

SphereRenderer の例でもみたような、自身が解放されるときに、保持している購読を全て解除するというパターンは頻出なので、ユーティリティを作ります。 Disposer を複数保持できて、 deinit でそれら全ての dispose を呼び出す型、 DisposerBag を作ります。 DisposerBag に Disposer を追加するメソッドに加えて、 Disposer 側に自身を DisposerBag に渡す逆方向のメソッドも作っておくと便利です。 これらは全て Rx にもあって、 DisposeBag と呼ばれています。

インターフェースは以下のようになります。

class DisposerBag {
    func dispose()
    func add(_ disposer: Disposer)
}

extension Disposer {
    func disposed(by disposer: DisposerBag)
}

これを導入すると SphereRenderer は以下のようになります。

class SphereRenderer {
    init(sphere: Sphere) {
        self.sphere = sphere

        reloadSphere()
    }

    var sphere: Sphere = .init() {
        didSet {
            disposerBag.dispose()

            rebuildMesh()

            unowned let uself = self

            sphere.radiusChangeEvent
                .subscribe { _ in
                    uself.rebuildMesh()
                }.disposed(by: disposerBag)

            sphere.colorChangeEvent
                .subscribe { _ in
                    uself.rebuildMesh()
                }.disposed(by: disposerBag)
        }
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    private func reloadSphere() {
        self.sphere = { self.sphere }()
    }

    private let disposerBag: DisposerBag = .init()
}

DisposerBag のおかげで、 deinit を書く必要がなくなりました。これにより解放漏れのミスが防ぎやすくなります。 Disposer を Bag に入れるところも、メソッドチェーンで書けるので中間変数が不要になりました。

Property

Sphere の実装においては、 didSet でイベントを emit するコードを書いています。そして、プロパティごとに対応する EventEmitter を定義しています。このような、値とその変更イベントをペアにしたデータ構造は使いどころが多いため、型にして再利用できるようにします。ここではこれを Property と命名します。インターフェースは以下のようになります。

class Property<T> {
    init(_ value: T)

    var value: T

    func subscribe(handler: @escaping (Event) -> Void) -> Disposer
}

EventEmitter と同様の subscribe メソッドを提供します。 value プロパティの setter でイベントが発行されます。

実装は EventEmitter を内部で使用すれば簡単で、以下のように書けます。

class Property<Value> {
    init(_ value: Value) {
        self._value = value
        self.emitter = EventEmitter<Value>()
    }

    public var value: Value {
        get {
            return _value
        }
        set {
            _value = newValue
            emitter.emit(_value)
        }
    }

    public func subscribe(_ handler: @escaping (Value) -> Void) -> Disposer {
        return emitter.subscribe(handler: handler)
    }

    private var _value: Value
    private let emitter: EventEmitter<Value>
}

これを使って Sphere の実装は以下のように更新されます。

class Sphere {
    init() {}

    var radius: Float {
        get { return radiusProperty.value }
        set { radiusProperty.value = newValue }
    }
    let radiusProperty: Property<Float> = .init(0)

    var color: Color {
        get { return colorProperty.value }
        set { colorProperty.value = newValue }
    }
    let colorProperty: Property<Color> = .init([])
}

radiusProperty と colorProperty が主な機能で、 radius と color はそれらのプロパティの value プロパティに素通しするだけのプロパティです。 これを作っておくと、従来の通常のプロパティの利便性も残すことができます。

通常のプロパティと EventEmitter のペアだった際には、 didSet で emit の呼び出しの実装を忘れるリスクがありましたが、 Property が先にあって追加で簡易プロパティを作る場合には、つなぎこまない限りコンパイルできないので、実装を忘れにくい形になっています。

この Sphere の変更を受けて、 SphereRenderer は以下のようになります。これまで EventEmitter を参照していた場所が、 Property になるだけです。

class SphereRenderer1 {
    init(sphere: Sphere) {
        self.sphere = sphere

        reloadSphere()
    }

    var sphere: Sphere = .init() {
        didSet {
            disposerBag.dispose()

            rebuildMesh()

            unowned let uself = self

            sphere.radiusProperty
                .subscribe { _ in
                    uself.rebuildMesh()
                }.disposed(by: disposerBag)

            sphere.colorProperty
                .subscribe { _ in
                    uself.rebuildMesh()
                }.disposed(by: disposerBag)
        }
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    private func reloadSphere() {
        self.sphere = { self.sphere }()
    }

    private let disposerBag: DisposerBag = .init()
}

didSet 代わりの Property

Sphere の radius と color の didSet は、 Property にしたことによって不要になりました。これを SphereRenderer の内部にも適用することができます。つまり、 SphereRenderer の sphere プロパティと didSet で書いていた処理を、 Property 型の sphereProperty プロパティと、そこから間接的に構成する computed property としての sphere プロパティに書き換えることができます。

これ自体にはあまりメリットはありませんが、後の変更の前処理として導入します。

class SphereRenderer {
    init(sphere: Sphere) {
        self.sphereProperty = .init(sphere)

        unowned let uself = self

        sphereProperty
            .subscribe { sphere in
                uself.disposerBag.dispose()

                uself.rebuildMesh()

                sphere.radiusProperty
                    .subscribe { _ in
                        uself.rebuildMesh()
                    }.disposed(by: uself.disposerBag)

                sphere.colorProperty
                    .subscribe { _ in
                        uself.rebuildMesh()
                    }.disposed(by: uself.disposerBag)
            }.disposed(by: disposerBag)

        reloadSphere()
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    var sphere: Sphere {
        get { return sphereProperty.value }
        set { sphereProperty.value = newValue }
    }

    let sphereProperty: Property<Sphere>

    private func reloadSphere() {
        self.sphere = { self.sphere }()
    }

    private let disposerBag: DisposerBag = .init()
}

sphereProperty プロパティを作って sphere プロパティはそれの簡易アクセスにしました。 init で sphereProperty を初期値付きで構築したあと、もともと sphere の didSet に書いていた内容を、 sphereProperty の subscribe のハンドラとして書いています。最後に呼んでいる reloadSphere の内容はそのままです。 computed property として書き直した sphere プロパティがこれまでどおりに振る舞うからです。

Property の初回 subscribe

さて、 Property では value のセッターでイベントを emit するようにしましたが、ここで一工夫するとより便利になります。それは、 subscribe において、その渡されたハンドラに即座に現在の値を流し込むことです。 Property はバインディング用途で使われることが想定されるため、値が流れてきたらそれに連動して何か同期処理をする事が多いです。ということは、購読を登録する初期化のタイミングでも、即座に現在値で同期を取ることになります。 現在の SphereRenderer においては、 sphereProperty の subscribe の中に書いてある uself.rebuildMesh の呼び出しと、 init の末尾に書いてある reloadSphere の呼び出しがそれに該当します。

ここで subscribe のタイミングで現在値を流してやれば、購読側の同期処理が subscribe のハンドラの挙動として一本化できます。 これは EventEmitter ではできなかったことです。　EventEmitter は emit メソッドでイベントを流すだけで、現在値の概念が無いからです。 Property には常に現在値が何かしら入っているのでこのような事ができます。

具体的に Property の実装としては、 subscribe メソッドが下記のように変更されます。

class Property<T> {
    ...

    func subscribe(_ handler: @escaping (Value) -> Void) -> Disposer {
        let disposer = emitter.subscribe(handler: handler)
        handler(_value)
        return disposer
    }
}

これによって、 SphereRenderer を以下のように改善できます。

class SphereRenderer {
    init(sphere: Sphere) {
        self.sphereProperty = .init(sphere)

        unowned let uself = self

        sphereProperty
            .subscribe { sphere in
                uself.disposerBag.dispose()

                sphere.radiusProperty
                    .subscribe { _ in
                        uself.rebuildMesh()
                    }.disposed(by: uself.disposerBag)

                sphere.colorProperty
                    .subscribe { _ in
                        uself.rebuildMesh()
                    }.disposed(by: uself.disposerBag)
            }.disposed(by: disposerBag)
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    var sphere: Sphere {
        get { return sphereProperty.value }
        set { sphereProperty.value = newValue }
    }

    let sphereProperty: Property<Sphere>

    private let disposerBag: DisposerBag = .init()
}

Property が現在値を即座に流すようになったことで、 sphereProperty の subscribe のハンドラ内部の rebuildMesh と、 init 末尾の reloadSphere が削除できました。 reloadSphere メソッドが不要になったこともあり、まあまあな行数を減らすことができました。

気がついた方もいるかもしれませんが、残念ながらこの変更によって、改悪されてしまっている点があります。それは、 sphere の代入の際に rebuildMesh が 2回走ってしまうようになったことです。 rebuildMesh 自体は 2回呼び出しても 結果は正しいものになる実装を想定しているので、描画結果には問題はありませんが、パフォーマンス上は無駄な処理が増えてしまっています。この問題については後々修正します。

EventSource と flatMapLatest

sphereProperty の subscribe 冒頭で DisposerBag を叩いていますが、これは radius と color に対する購読が、最新の sphere に対してのみ行われるようにするためでした。新しい sphere が来たら古い radius と color の購読を解除して、新たに購読を接続します。このような場面で、ある外側のイベントストリームがあって、そのイベントごとにさらに内側のイベントストリームが生成されているとき、常に最新の外側のイベントから産まれた内側のイベントストリームだけを購読し、古い外側のイベントは解除する、というパターンを共通化することができます。これを flatMapLatest と呼びます。

これはイベントストリームの話なので、 EventEmitter でも Property でも共通です。つまり、これら両方に共通なメソッドとして提供することができます。これを EventSourceProtocol と命名します。

以下に EventSourceProtocol と flatMapLatest のインターフェースを示します。

protocol EventSourceProtocol {
    associatedtype Event

    func subscribe(handler: @escaping (Event) -> Void) -> Disposer
}

class EventEmitter<Event> : EventSourceProtocol { ... }

class Property<T> : EventSourceProtocol { ... }

class EventSource<T> : EventSourceProtocol { ... }

extension EventSourceProtocol {
    func flatMapLatest<USource: EventSourceProtocol>(_ flatMap: @escaping (Event) -> USource) -> EventSource<USource.Event>
}

新たに出てきた EventSource というクラスは、EventSourceProtocol に定義された subscribe メソッドだけを持っている匿名なクラスです。具体的な EventSourceProtocol のインスタンスをラップして生成します。 EventEmitter の emit メソッドや、 Property の value のように、イベントを流すための機能は持っていません。このような型を type erasure と呼びます。つまり EventSource は EventSourceProtocol の type erasure です。

flatMapLatest は EventSourceProtocol に対する extension なので、 EventEmitter と Property のどちらに対しても呼び出せます。引数としてイベントから EventSourceProtocol を返すクロージャを受け取っていて、これが内側のイベントストリームを生成します。返り値は生成される内側のイベントストリームと同じイベントの型を持った EventSource です。内側のイベントストリームは、外側のイベントに応じて何度も新たに生成されますが、返り値のオブジェクトはこれらをつなげた一本のイベントストリームのように振る舞います。

flatMapLatest を使うと SphereRenderer は以下のように書けます。

class SphereRenderer {
    init(sphere: Sphere) {
        self.sphereProperty = .init(sphere)

        unowned let uself = self

        sphereProperty.flatMapLatest { $0.radiusProperty }
            .subscribe { _ in
                uself.rebuildMesh()
            }.disposed(by: disposerBag)

        sphereProperty.flatMapLatest { $0.colorProperty }
            .subscribe { _ in
                uself.rebuildMesh()
            }.disposed(by: disposerBag)
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    var sphere: Sphere {
        get { return sphereProperty.value }
        set { sphereProperty.value = newValue }
    }

    let sphereProperty: Property<Sphere>

    private let disposerBag: DisposerBag = .init()
}

これまで　sphereProperty の subscribe の中でさらに subscribe していた入れ子の形が、 flatMapLatest で radiusProperty や colorProperty を引き出す形になって、 subscribe の入れ子を除去できました。また、 flatMapLatest 自体が古い内側の購読を解除するため、 disposerBag を明示的に叩くコードも不要になりました。

map, merge オペレータ

SphereRenderer の実装において、 radius と color に関して、後続の処理は同様の内容が重複しています。これを単一化することを考えます。

イベントの型を変換するだけの map と、同じ型をもつ複数のイベントストリームを一本にまとめる merge を導入すると、それらをまとめる事ができます。

extension EventSourceProtocol {
    func map<U>(_ map: @escaping (Event) -> U) -> EventSource<U>
}

func merge<TSource: EventSourceProtocol>(_ sources: [TSource]) -> EventSource<TSource.Event>

map は EventSourceProtocol のメソッド、 merge はグローバル関数として提供します。そもそも今回の場合は、 radius や color の実際の値は読み捨てていて、変化のイベントだけを取り扱っているので、 map を使って Void 型に変換してしまえば、 2つが同じ型になるので merge して 1つのイベントにする事ができます。つまり、 radius か color のどちらかが変化したことを示すイベントを作れます。

こうして、 SphereRenderer を以下のように書けます。

class SphereRenderer {
    init(sphere: Sphere) {
        self.sphereProperty = .init(sphere)

        unowned let uself = self

        merge([sphereProperty.flatMapLatest { $0.radiusProperty }.map { _ in },
               sphereProperty.flatMapLatest { $0.colorProperty }.map { _ in }])
            .subscribe {
                uself.rebuildMesh()
            }.disposed(by: disposerBag)
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    var sphere: Sphere {
        get { return sphereProperty.value }
        set { sphereProperty.value = newValue }
    }

    let sphereProperty: Property<Sphere>

    private let disposerBag: DisposerBag = .init()
}

これで冗長な記述をほぼ全部削除できました。 SphereRenderer が保持している Sphere の radius か color の変化に際して rebuildMesh を呼び出す、という期待する仕様がシンプルに表せています。
また些細な点ですが、 Void の読み捨ては省略できるので、 subscribe のハンドラの _ in が不要になっています。

ここまででてきたような map, flatMap, merge といった、イベントストリームを加工する道具をオペレータと呼びます。メソッドや関数といった言語機能から離れて、抽象的なデータ操作として捉えた呼び方です。

これらのオペレータは意味が直交しているので、 map や merge を flatMap の内側に書くこともできます。

combine 方式

上記の merge 方式でロジックとしては正しく動作しますが、初期化時や sphere の更新時には、 radius と color のイベントが両方流れるため、処理としては radius について生じる rebuildMesh が完全に無駄です。即座に color についての処理がかかって、 radius の結果は破棄されます。

そこで、複数のストリームをまとめて、1つのタプルを流すストリームにする combine オペレータを導入します。 combine は合成するストリームが流したイベントを記憶しておいて、全てのストリームからイベントが得られたら、タプルにして自身のイベントとして流します。それ以降は、どれか一つのストリームがイベントを流すたびに、タプルのその部分だけ変更してイベントを流します。

これを導入すると以下のように書けます。

class SphereRenderer {
    init(sphere: Sphere) {
        self.sphereProperty = .init(sphere)

        unowned let uself = self

        sphereProperty
            .flatMapLatest {
                combine($0.radiusProperty, $0.colorProperty)
            }.subscribe { _ in
                uself.rebuildMesh()
            }.disposed(by: disposerBag)
    }

    func rebuildMesh() {
        // using: sphere.radius, sphere.color
    }

    var sphere: Sphere {
        get { return sphereProperty.value }
        set { sphereProperty.value = newValue }
    }

    let sphereProperty: Property<Sphere>

    private let disposerBag: DisposerBag = .init()
}

combine は flatMapLatest の内側に書いて、 sphere ごとにまとめる必要があります。逆にプロパティの flatMap を combine するように書いてしまうと、2回目以降の sphere では 無駄なイベントが走ってしまいます。 radiusProperty と colorProperty は Property のため、かならず即座に1度はイベントを流す事を保証しているため、 sphere が1つ来て combine が構築されると即座に1つだけタプルを流してくれます。

これで、最初に1回、 sphere の変更時に1回、 radius か color の変更に1回だけ、 rebuildMesh が呼び出される状態になりました。

Rx とバインディング

EventEmitter からはじめて、これをバインディングに適用する場合に、オペレータ等を導入していくことでコードが簡潔化できることをみてきました。こうしてできあがった形は、 Rx でバインディングを行うときのコードとほぼ同じです。どこから Rx と同じだったのかというと、出発点の EventEmitter から見ると、実は、イベント種類ごとに分割し、 subscribe メソッドを定義した時点で、型として Rx の Observable と全く同じものになっていました。 Rx の Observable は 今回定義した EventSource と同様、匿名なインターフェースなので、今回の EventEmitter が Rx の Observable の一種になっていると考えられます。 Rx では今回みたようなオペレータがいろいろ提供されていますが、これは Observable の subscribe メソッドの上で追加的に定義されているものなので、今回の EventSource にも同じようにいろいろと定義できます。

今回の EventSource で Rx と決定的に違うのは、 Rx にはエラーと終了を流す機能があることです。 EventSource では Event の型がそのままイベントでしたが、 Rx においてはイベントの型 T に対して、 next(T), error(Error), completed の 3種類がストリームとして流れるようになっています。 EventSource は本質的に無限ストリームですが、 Rx では 終了があるので有限ストリームも扱えます。

今回見てきたように、バインディングタスクにおいては、エラーと終了は特に欲しくありません。 RxSwift においては、 エラーが流れると購読が解除されるため、ドメインロジックで生じたエラーが表示系のバインディングに流れ込まないように、境界で保護したりします。

非同期

Rx は非同期処理に使われていますが、今回の EventSource もそのまま非同期処理に適用できます。 Rx で行われるように、 subscribe で処理を開始して、非同期な結果を ストリームに流すようにすることもできますし、 Promise を EventSource の一種として実装することもできます。 Promise における非同期処理の then によるチェーンは、ここでは flatMapLatest になります。ただし、エラーを流す仕組みは無いため、だいたいエラーを伴うことになる非同期処理で使う場合は Result を組み合わせるなどの工夫が必要です。

実装

記事では EventEmitter 等のコードはインターフェースだけにとどめて実装は省略しました。今回これらをライブラリとしてまとめたので、興味のある方はソースを見たり使ってみたりしてください。

ReativeEmitter