この記事は、2017/09/15〜17 に早稲田大学 理工学部 西早稲田キャンパスで開催される iOSDC Japan 2017 で行われるセッション「RxSwiftのObservableとは何か」の発表原稿、およびその補足資料です。
- スライドはこちらです ➡︎ https://www.slideshare.net/gomi_ningen/rxswiftobservable-iosdc-japan-2017
- あわせて読みたい ➡︎ http://qiita.com/gomi_ningen/items/dc08a8a5514be9aa0eb2
なお、本文に先立ち注意事項を掲載しておきます。
注意事項
- 以下の内容を理解しなくても RxSwift は十分使えるライブラリです
- 記事の内容的には Rx 系ライブラリの利用経験がなくても分かるように書いたつもりです
- 以下の実装は RxSwift のものであり、他言語の Rx ライブラリとは実装が異なる場合があります
1.RxSwiftとは?
ReactiveExtensions(Rx) とは一体何をするライブラリなのでしょうか? ReactiveX のWebページ冒頭や RxSwiftの README にはこのように書かれています
- An API for asynchronous programming with observable streams.(Observable を用いた非同期プログラミングのためのAPI)
- Rx is a generic abstraction of computation expressed through Observable interface. (Rxは Observable を用いて計算を抽象化します)
どちらにも Observable
というキーワードが登場しています。このことから、Rx を理解するためには、Observable
について理解する必要がありそうだということがお分りいただけるかと思います。
実際に Rx は、主に以下の3つの要素から構成されていると言ってよいでしょう。
-
Observable (
Observable
,Observer
,Disposable
, etc...) -
Operator (
map
,flatMap
,filter
, etc...) -
Scheduler (
MainScheduler
,ConcurrentDispatchQueueScheduler
, etc...)
本セッションでは、このうち Observable
にスポットライトをあて、その実装を俯瞰していくことによって Rx への理解を深めることを目的としてお話を進めていきます。
しばしば、Rx のストリームは「川である」といったように様々なものの例えで表現されることがあります。またリアクティブプログラミングとは〜のようなものであるといった記事がネット上にはたくさん存在します。こうした、あいまいでつかみどころのない例え話や解説記事から一歩足を踏み入れて、コードレベルで振る舞いを理解することにチャレンジしてみませんか。
実装を知ることは、 RxSwift をはじめとした Rx 系ライブラリを活用する手助けになるだけでなく、ソフトウェアを設計する上でのひとつの大きな指針を手にいれることにも繋がると、私は考えています。
2. Observerパターンとそのバリエーション
2.1. Observer パターンの問題意識
Rx の Observable
の実装を見ていく前に、ひとつの問題について考えてみましょう。
Q1. 「A に変化が生じたら B に伝えたい」とき、どうすればよいでしょうか?
現実世界に置き換えると、以下のように例えられると思います。
- 【例1】 自分の予定が変わったので(状態変化)、お店に予約キャンセルの連絡をする(通知)
- 【例2】 従業員は体調が悪くなったので(状態変化)、上司に会社を休む連絡をする(通知)
実際にこのようなケースの場合、自分が相手の連絡先を保持して、連絡をとる(相手に通知する)ことになるかと思います。【例2】のケースを考えると、普通従業員は上司の連絡先をあらかじめ保持していて、適切なタイミングでお休みの連絡を入れるのではないでしょうか。これはプログラミングの世界でも一緒で、「A に変化が生じたら B に伝えたい」ときには、「A が B の参照を保持しておき、状態変化時に通知する」という構造になります。
【例2】のパターンをコードで表現すると以下のようになると思います。Playgroundで動作するコードですのでお試しください。
class Member {
private let boss: Boss
public var isFine: Bool = true {
didSet(value) { boss.notify() }
}
public init(boss: Boss) { self.boss = boss }
}
class Boss {
public func notify() {
NSLog("誰かから通知が来たよ")
}
}
var boss1 = Boss()
var member1 = Member(boss: boss1)
member1.isFine = true //=> ログ出力される
member1.isFine = false //=> ログ出力される
この実装は、問題を解決できており、完全に正しい実装です。与えられた前提条件に対して設計上の問題点はありません。さて、これを踏まえて考える問題をもう少し複雑にしてみましょう。
Q2. 「通知元(A)の状態変化を、複数の通知先(B, C, D...)に伝えたい」とき、どうすればよいでしょうか?
基本的に A が B, C, D の参照を保持するという点については変わりませんが、B, C, D に通知をする際のインターフェースが同じであるとは限りません。これは現実に例えれば、友人数人と遊ぶ約束をしていて、遅刻しそうなとき、B さんは twitter、C さんは 電話、D さんはメールで連絡しなければいけないといったような状況です。
class A {
private let b: B
private let c: C
private let d: D
public var isFine: Bool = true {
didSet(value) {
if value {
b.notifyByTwitter()
c.notifyByTelephone()
d.notifyByEmail()
}
}
}
public init(b: B, c: C, d: D) {
self.b = b; self.c = c; self.d = d;
}
}
class B { func notifyByTwitter() { NSLog("連絡がきた") } }
class C { func notifyByTelephone() { NSLog("連絡がきた") } }
class D { func notifyByEmail() { NSLog("連絡がきた") } }
2.2. pull 型 Observer パターン
先にみた構造には、以下のような問題点があります。
- 通知先のオブジェクトが変更されたとき、通知元の実装も変更しなければならない
- 通知先のオブジェクトの種類が増えるとき、通知元の実装も追加しなければならない
2つの問題は通知元のオブジェクトが、通知先のオブジェクトの詳細を知りすぎているということに起因しています。通知元は、通知先の参照をどうあがいても保持する必要がありますが、通知先の詳細を知ったまま保持する必要性はありません。
つまり、通知元は、通知先の状態遷移を伝えるためのインターフェースだけを知っていればよいことになります。友人一般を表す Friend
という名前で protocol を切り、通知を受け入れるメソッド notify
を定義しておくと、B, C, D はそれぞれ下図のような構造にできるでしょう。
この形にすることにより、通知元の A のほうも 友人への参照を個別の型で保持するのではなく、[Friend]
という配列で保持することができるようになりました。配列で保持しているので Friend
プロトコルを実装した新たなインスタンスを容易に受け入れることができる構造となりました。また配列から要素の削除も容易に行えるため、通知を取りやめることも実現できる構造になっています。
A が新たな Friend のインスタンスを受け入れるメソッドを subscribe
、逆に取り除くメソッドを unsubscribe
と命名すると以下の図のように単純に構造を表現できます
ここまでの検討をまとめれば「通知元の状態変化を、複数の通知先に伝えたい」場合に、通知元と通知先が持っていてほしいインターフェース仕様が導けるでしょう。それぞれ次のようになります。
- 通知元(Observable=観測可能なオブジェクト)は次の2つのメソッドを持っていてほしい
- 通知先(Observer=観測者)への通知を開始するためのメソッド(subscribe)
- 通知先(Observer=観測者)への通知を終了するためのメソッド(unsubscribe)
- 通知先(Obseerver=観測者)は次の1つのメソッドを持っていてほしい
- 通知を受け付けるメソッド(notify)
クラス図にまとめると以下のような形になります。
この構造は、pull型 Observer パターン1と呼ばれています。Swiftで表現すると、やや制約のあるコードですが以下のように表現できます。
public protocol Observable {
func subscribe(obs: Observer)
func unsubscribe(obs: Observer)
}
public class ConcreteObservable {
private var observers: [Observer] = []
public var isHoge: Bool = false {
didSet { observers.forEach { x in x.notify() } }
}
public func subscribe(obs: Observer) { observers += [obs] }
pubilc func unsubscribe(obs: Observer) {
observers = observers.filter { x in
// this means reference equality
ObjectIdentifier(x) != ObjectIdentifier(obs)
}
}
}
public protocol Observer: class {
func notify()
}
public class ConcreteObserver: Observer {
public func notify() { NSLog("通知を受けた") }
}
// 以下のように playground でお試しできます
let v1 = ConcreteObservable()
let obs1 = ConcreteObserver()
v1.subscribe(obs: obs1)
v1.isHoge = false //=> ログ出力される
v1.isHoge = true //=> ログ出力される
v1.unsubscribe(obs: obs1)
v1.isHoge = false //=> ログ出力されない
C# で表現すると以下のようなコードになります。C#を読んだことがない方でも、言語の違いを飛び越えてほぼ同じような表現が可能であることがわかるのではないでしょうか。
namespace DotNetObservable
{
public interface IObservable
{
void Subscribe(IObserver obs);
void Unsubscribe(IObserver obs);
}
public class ConcreteObservable : IObservable
{
readonly IList<IObserver> observers = new List<IObserver>();
bool isHoge;
bool IsHoge
{
get => isHoge;
set
{
isHoge = value;
foreach (var obs in observers) obs.Notify();
}
}
public void Subscribe(IObserver obs) => observers.Add(obs);
public void Unsubscribe(IObserver obs) => observers.Remove(obs);
}
public interface IObserver
{
void Notify();
}
public class ConcreteObserver : IObserver
{
public void Notify() => Console.WriteLine("通知された");
}
}
pull 型 Observer パターンについてのまとめ
やや長くなったので、ここで pull 型 Observer パターンについてまとめておきましょう。pull 型Observerパターンは、「複数の異なる通知先に状態変化を通知したい」という問題の解決策として以下のような構造や特徴を持ちます。
- 通知元(Observable)は、通知先(Observer)に共通したプロトコルを持つインスタンスをコレクションとして持ち、状態遷移時にそれぞれに通知する
-
Observer
は、Observable
からなんらかの変化が発生したという情報だけをnotify()
メソッド経由で受け取れる -
Observer
は、必要に応じてObservable
の値を問い合わせる(このあたりが pull 型と呼ばれる所以)
2.3. push 型 Observer パターン
pull 型の Observer パターンを用いた場合、Observer
は Observable
の状態が更新されたという事実を知ることができますが、どのような値に更新されたのかは、 Observable
のプロパティなどを参照しにいかなければ知ることができません。したがって、現実的には更新された値を参照するために、Observer
も Observable
の参照を何らかの方法で取得できる状況にする必要があり、相互参照する構造になってしまいます。
この問題を解決するために、notify
時に更新後の値を渡してしまう構造にしたものが push 型 Observer パターン になります。構造は以下のようになります。
図中で interface が型パラメータ(generic type parameter)を持つ箇所がありますが、Swift の protocol は型パラメータを持てず、関連型(associated type, abstract type member) しかもてない仕様があり2、この図をストレートにコードに落とし込むことができません。したがって、ひとまず C# で表現するならば以下のようなコードになります。
using System.Collections.Generic;
namespace DotNetObservable
{
public interface IObservable<T>
{
void Subscribe(IObserver<T> obs);
void Unsubscribe(IObserver<T> obs);
}
public class BooleanObservable : IObservable<bool>
{
readonly IList<IObserver<bool>> observers = new List<IObserver<bool>>();
bool t;
bool TValue
{
get => t;
set
{
t = value;
foreach (var obs in observers) obs.Notify(value);
}
}
public void Subscribe(IObserver<bool> obs) => observers.Add(obs);
public void Unsubscribe(IObserver<bool> obs) => observers.Remove(obs);
}
public interface IObserver<T>
{
void Notify(T t);
}
public class BooleanObserver : IObserver<bool>
{
public void Notify(bool t) => Console.WriteLine("通知された");
}
}
Swift で表現するならば、ちょっと迂回して以下の図のような構造となるでしょう。
Swiftには抽象クラスというものは存在しませんが、擬似的に abstract method 内 で fatalError()
を返すことを抽象クラスである印として、次のようにコードに落とし込めます。この技法は RxSwift のコード内部でも使われています3。
public protocol ObserverType: class {
associatedtype E
func notify(value: E)
}
public protocol ObseravbleType {
associatedtype E
func subscribe<O: ObserverType>(obs: O) where O.E == E
func unsubscribe<O: ObserverType>(obs: O) where O.E == E
}
public class Observable<Element>: ObseravbleType {
public typealias E = Element
public func subscribe<O: ObserverType>(obs: O) where O.E == E {
fatalError("not implemented")
}
public func unsubscribe<O: ObserverType>(obs: O) where O.E == E {
fatalError("not implemented")
}
}
public class BooleanObservable: Observable<Bool> {
private var observers: [ObjectIdentifier:AnonymousObserver<Bool>] = [:]
public var isHoge: Bool = false {
didSet { observers.forEach { x in x.value.notify(value: isHoge) } }
}
public override func subscribe<O>(obs: O) where O : ObserverType, O.E == Bool {
observers[ObjectIdentifier(obs)] = AnonymousObserver(handler: obs.notify)
}
public override func unsubscribe<O>(obs: O) where O : ObserverType, O.E == Bool {
observers[ObjectIdentifier(obs)] = nil
}
}
public class AnonymousObserver<Element>: ObserverType {
public typealias E = Element
public typealias Handler = (E) -> Void
private let handler: Handler
public init(handler: @escaping Handler) {
self.handler = handler
}
public func notify(value: E) { handler(value) }
}
var observable = BooleanObservable()
var observer = AnonymousObserver<Bool>(handler: { x in NSLog(String(x)) })
observable.isHoge = false //=> ログ出力されない
observable.subscribe(obs: observer)
observable.isHoge = true //=> ログ出力される
observable.isHoge = false //=> ログ出力される
observable.unsubscribe(obs: observer)
observable.isHoge = true //=> ログ出力されない
こうして、私たちは「通知元の状態変化を、複数の通知先に伝えたい」という問題を解決する実装パターンである push 型 Observer パターンを手にすることができました。 protocol が型パラメータを持てないという制約を回避するためにやや遠回りで苦しい実装をしましたが、構造的には C# のものと同じです。
3. RxSwift の Observer, Observable の実装
さて、ここからはいよいよ RxSwift の Observable
についてみていきます。基本的に push 型 Observer パターンと同じ構造になりますが、3つ異なる点があります。
- notify 時に値を
.next
,.error
,.completed
という文脈につつむ enumEvent
にラップして渡している -
.next
以外の値が飛んだ場合に、以後イベントは飛ばなくなる -
unsubscribe
の責務をDisposable
に分離している
まずは前者から見ていきましょう。
3.1. Event の実装
RxSwift の Observable
は変化した値(next)に加えて、エラー(error)と完了(completed)という文脈を Observer
に通知することができます。その文脈を表現するのが Event
という enum です。実装が単純なので、コードをみたほうが理解がはやいでしょう。
public enum Event<Element> {
case next(Element)
case error(Error)
case completed
}
また通知に関してもう一つだけ push 型 Observerパターンにはないルールがあります。それは 「next
以外の値が飛んだ場合に、以後イベントは飛ばなくなる」というルールです。これは RxSwift のソースコード内で以下のように表現されています。
extension Event {
public var isStopEvent: Bool {
switch self {
case .next: return false
case .error, .completed: return true
}
}
}
3.2. Disposable プロトコルの定義
続いて Disposable
プロトコルの定義についてみていきましょう。これも先に見た push 型の Observer パターンには存在しないものになります。このプロトコルは単純に「push 型の Observer パターンの unsubscribe
の役割をオブジェクトとして切り出した」ものになります。したがって、unsubscribe
を発火させる単純なメソッド dispose
のみを持つシンプルなインターフェースとなります。
public protocol Disposable {
func dispose()
}
このプロトコルの実装(具体的には SubscriptionDisposable
など)については、ReactiveExtensions の Observable
の本質とは少し離れ、かなり Swift という言語やメモリモデルに依存したものになるため、ひとまず後回しにして他の実装をみていくことにしましょう。
3.3. ObserverType, ObservableType プロトコルの定義
ObserverType, ObservableType プロトコルの定義については、ほとんど push 型の Observer パターンとかわりありません。変化しているのは「値がイベントという文脈付きで通知される」という点だけです。文脈がついているため、いままで notify
と命名していた箇所が on
という名前に変わっています。これは、引数に enum 値として .next(Element)
, .error(Error)
, .completed
が渡ると考えると自然な命名ですね。
public protocol ObservableType {
associatedtype E
func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E
}
public protocol ObserverType {
associatedtype E
func on(_ event: Event<E>)
}
また、ObserverType
には便利な拡張メソッドとして onNext(Element)
, onError(Error)
, onCompleted
が生えています。
extension ObserverType {
public final func onNext(_ element: E) {
on(.next(element))
}
public final func onCompleted() {
on(.completed)
}
public final func onError(_ error: Swift.Error) {
on(.error(error))
}
}
3.4. Observable の実装
つづいて Observable の実装についてみていきましょう。 基本的に ObservableType
の関連型(associated type)を、型パラメータに引き上げる以外のことはやっていません。Swift には抽象クラスや抽象メソッドという言語機能が存在しないため、その意志を込めて Never
型を返すメソッドを呼び出しているのが特徴的です。
public class Observable<Element>: ObservableType {
public typealias E = Element
public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
abstractMethod()
}
}
/// 抽象メソッドを表現するための苦肉の策
func abstractMethod() -> Never {
fatalError("abstract method")
}
3.5. AnonymousObserver の実装
AnonymousObserver
に関しても push 型 Observer パターンをベースに「next
以外の値が飛んだ場合に、以後イベントは飛ばなくなる」というルールが追加した実装に変更します。
class AnonymousObserver<Element>: ObserverType {
typealias E = Element
typealias EventHandler = (Event<Element>) -> Void
private let eventHandler: EventHandler
private var isStopped: Int32 = 0 //=> means AtomicInt
public init(_ eventHandler: @escaping EventHandler) {
self.eventHandler = eventHandler
}
public func on(_ event: Event<Element>) {
switch (event) {
case .next:
if isStopped == 0 { eventHandler(event) }
case .error, .completed:
if !OSAtomicCompareAndSwap32Barrier(0, 1, &isStopped) { return }
eventHandler(event)
}
}
}
Swift では Bool
を参照し、条件次第で異なる値を代入する操作はアトミックではないため、その代わりのフラグとして Int32
が利用されています。また、 OSAtomicCompareAndSwap32Barrier
は、値と変数を比較して等しい場合に、新しい値を代入する操作をアトミックに行うものです。この場合 isStopped
が 0 と等しいか比較をして真であれば 1 に差し替えるという操作をアトミックに行います。このあたりは AnonymousObserver
の本質的な部分でないため、難しければ理解しなくても大丈夫です。
3.6. RxSwift の Observable についてのまとめ
歴史的経緯はどうだったのか知りませんが、Rx の基本的なインターフェースである Observable
, Observer
については以下のように説明できると思います(※筆者は歴史的経緯は知らないので実際の流れは違うかもしれません)。
- push 型 Observer パターンが基本的な出発点
- 通知する値に
.next
,.error
,.completed
という文脈をつけたEvent
が通知の対象物になっているのが特徴的 - また
.next
以外の値が通知されると以降、イベントは送られない(ストリームが閉じる) - 購読解除の仕組みを
Disposable
に分離しているのが特徴的
以上を踏まえると Observable
, Observer
といったインターフェースを自然に導き出すことができると思います。ここまでくれば、Observable
はもはや「川」といったあいまいなものではなく、よりビビッドに振る舞いを捉えることができるようになっているのではないでしょうか。
4. Subject と Disposable を実装する
これまで観測可能な値(Observable
)と、その状態変化を観測するオブジェクト(Observer
)について詳しく見てきましたが、ReactiveExtensions にはそのどちらの役割も持つオブジェクトが存在します。それが、Subject
です。代表的なものに BehaviorSubject
, PublishSubject
などがあります。
Observable
であり、かつ Observer
であるということは単純に以下のような SubjectType
プロトコルに落とし込むことができます。
// 通知元にも通知先にもなりうるオブジェクトを表す
public protocol SubjectType: ObservableType { // 普段は Observable としてふるまう
typealias SubjectObserverType: ObserverType
func asObserver() -> SubjectObserverType // 必要なときに SubjectObserverType に変換できる
}
4.1. PublishSubject のふるまい
Observable
は、 subscribe(Observer)
メソッドによって受け入れた Observer
にイベント発生時に通知をするという働きをします。 Observer
は発生したイベント値の通知を受け入れる on(Event<T>)
というメソッドを持っています。
PublishSubject
は Subject
であるため、 on(Event<T>)
と subscribe(Observer)
双方のメソッドを持っています。ベーシックな使い方としては、 PublishSubject
にあらかじめ Observer
を Subscribe
させておき、on(Event<T>)
を読んだときに通知させるというものになります。これは先にみてきた Observer パターンと全く同じ構図になります。 PublishSubject
というと得体の知れないものに聞こえるかも知れませんが、実態はこれまで散々見てきた Observer パターンのなかの Observable 具象クラスそのものと同じ立ち位置であるといえば、わかりやすいのではないかと思います。実際に動かしたいコードのイメージは以下のようになります。
var isHoge = PublishSubject<Bool>()
var observer = AnonymousObserver<Bool>({ event in
switch(event) {
case .next(var value): NSLog(String(value))
case .error(var error): NSLog(error.localizedDescription)
case .completed: NSLog("completed")
}
})
var disopsable = isHoge.subscribe(observer)
isHoge.on(Event.next(true)) //=> [LOG] `true`
isHoge.on(Event.next(false)) //=> [LOG] `false`
disopsable.dispose()
isHoge.on(Event.next(true)) //=> ログ出力されない
push 型 Observer パターンでもみた風景ですね。これを実現するためには素直に Observable のインターフェースを実装していけば良いはずなので、書き出しは以下のような形になるかと思います。
public class PublishSubject<Element>: Observable<Element>, SubjectType, ObserverType {
public typealias SubjectObserverType = PublishSubject<Element>
var observers: [String:AnonymousObserver<Element>] = [:]
public override func subscribe<O>
(_ observer: O) -> Disposable where O : ObserverType, O.E == Element {
// dictionary にしないと O : ObserverType, O.E == Element が
// Equatable じゃないので削除(unsubscribe)できなくなる
let key = UUID().uuidString
observers[key] = AnonymousObserver(observer.on)
fatalError()
}
public func on(_ event: Event<Element>) { observers.forEach { x in x.value.on(event) } }
public func asObserver() -> PublishSubject<Element> { return self }
}
ほぼ push 型 Observer パターンのときと同様です。しかしながら、まだ Disposable
の具象クラスを実装していないため、subscribe
の戻り値を生成できず、ひとまず Never
型を返す fatalError
をよんでいます。直近の課題は、subscribe
で返すべき SubscriptionDisposable
というクラスを実装するというものになります。
4.2 SubscriptionDisposable の実装
ここでの Disposable
の責務は端的に言えば 「unsubscribe
を呼び出す」というものになります。したがってまず、PublishSubject
自体に unsubscribe
メソッドを生やしてしまいましょう。外部へは Disposable
プロトコルで渡るため、internal スコープ内に unsubscribe
メソッドを持つことを伝えるための UnsubscribeType
というプロトコルも同時に切ると、以下のような形になります。
protocol UnsubscribeType: class {
func unsubscribe(key: String)
}
public class PublishSubject<Element>:
Observable<Element>, SubjectType, ObserverType, UnsubscribeType {
public typealias SubjectObserverType = PublishSubject<Element>
var observers: [String:AnonymousObserver<Element>] = [:]
public override func subscribe<O>
(_ observer: O) -> Disposable where O : ObserverType, O.E == Element {
let key = UUID().uuidString
observers[key] = AnonymousObserver(observer.on)
//=> unsubscribe が以下のように実装されたので、key と 自分自身への弱参照を保持するオブジェクトを返せばよい
fatalError()
}
internal func unsubscribe(key: String) { observers.removeValue(forKey: key) }
public func on(_ event: Event<Element>) { observers.forEach { x in x.value.on(event) } }
public func asObserver() -> PublishSubject<Element> { return self }
}
unsubscribe
がこのように実装されたため、subscribe
ではこのオブジェクトへの弱参照と、引数で受け入れた observer に対応するキーを保持したオブジェクトを作ればよいことになります。
struct SubscriptionDisposable<T>: Disposable {
weak var ref: UnsubscribeType?
let key: String
init(ref: UnsubscribeType, key: String) {
self.ref = ref
self.key = key
}
public func dispose() {
ref?.unsubscribe(key: key)
}
}
こうして購読解除の仕組みを Disposable
に閉じ込めることができました。Disposable
は unsubscribe
を発火させたい Observable
を弱参照で持っているため、Diposable
を扱うクラスと Obseravble
の依存関係を晴れて参照レベルで断ち切ることができました。
4.3. PublishSubject と SubscribeDisposable
最後に PublishSubject
の fatalError()
だった箇所を修正して、ひとまず PublishSubject としてのふるまいを実装できたことになります。
public class PublishSubject<Element>:
Observable<Element>, SubjectType, ObserverType, UnsubscribeType {
public typealias SubjectObserverType = PublishSubject<Element>
var observers: [String:AnonymousObserver<Element>] = [:]
public override func subscribe<O>
(_ observer: O) -> Disposable where O : ObserverType, O.E == Element {
let key = UUID().uuidString
observers[key] = AnonymousObserver(observer.on)
return SubscriptionDisposable<Element>(ref: self, key: key)
}
internal func unsubscribe(key: String) { observers.removeValue(forKey: key) }
public func on(_ event: Event<Element>) { observers.forEach { x in x.value.on(event) } }
public func asObserver() -> PublishSubject<Element> { return self }
}
struct SubscriptionDisposable<T>: Disposable {
weak var ref: UnsubscribeType?
let key: String
init(ref: UnsubscribeType, key: String) {
self.ref = ref
self.key = key
}
public func dispose() { ref?.unsubscribe(key: key) }
}
これは RxSwift の実装と完全には一致しないのですが、本質的にはこのような構造を持つと思います。
4.4. (余談) 停止したストリームの subscribe と NopDisposable
.error
, .completed
が飛んだあとに Observable
を subscribe
しても、以後イベントは飛ばないため意味がありません。このとき subscribe
は observer
コレクションへの追加を行わない実装になっており、コレクションからの削除を実行するためのオブジェクトである Disposable
は何もしなくてもよいことになります。これに対応して、RxSwift には何もしない Disposable
の実装として NopDisposable
が用意されています。コードは非常に単純で次のとおりになります。
struct NopDisposable : Disposable {
init() {}
public func dispose() {}
}
実際の PublishSubject
にはストリームが止まった以後の subscribe
時には SubscriptionDisposable
ではなく NopDisposable
が返されていますので興味のある方はみてみると良いと思います。
4.5. (余談) スレッドセーフにするために
これまで実装した PublishSubject
は subscribe
, dispose
, on
の呼び出しに対してスレッドセーフではありません。RxSwift では RecursiveLock
を使ってスレッドセーフとなるような記述が入っています。使い方は簡単で、以下のようなものです。
let lock = NSRecursiveLock()
lock.lock()
/*
ここに排他したい処理を書く
*/
lock.unlock()
RxSwift 内では特に以下のような書き方が目立ちます。これは排他したい区間(クリティカルセクション)での例外発生時にロックが解除されないことを防ぐ意図があると思われます。
何かのスコープ {
let lock = NSRecursiveLock()
lock.lock()
defer { lock.unlock() }
/*
ここに排他したい処理を書く
*/
}
興味のある方は、このあたりも意識しながら、ソースコードを読んでみると良いと思います。
4.6. (余談) Bag の実装
observer を追加/削除できるコレクションとして今回は UUID#uuidString
をキーとして [String:ObserverType] という dictionary を用いましたが、実際には insert
時にそれに対応するキーを発行してくれる Key-Value ストア Bag
が用いられています。
struct BagKey: Hashable {
fileprivate let rawValue: UInt64
var hashValue: Int { return rawValue.hashValue }
}
func ==(lhs: BagKey, rhs: BagKey) -> Bool { return lhs.rawValue == rhs.rawValue }
struct Bag<T> {
typealias KeyType = BagKey
typealias Entry = (key: BagKey, value: T)
fileprivate var nextKey: BagKey = BagKey(rawValue: 0)
var dictionary: [BagKey:T] = [:]
init() { }
mutating func insert(_ element: T) -> BagKey {
let key = _nextKey
nextKey = BagKey(rawValue: nextKey.rawValue &+ 1)
dictionary[key] = element
return key
}
mutating func removeAll() {
dictionary.removeAll(keepingCapacity: false)
}
mutating func removeKey(_ key: BagKey) -> T? {
return dictionary.removeValue(forKey: key)
}
func forEach(_ action: (T) -> Void) {
for element in dictionary.values {
action(element)
}
}
}
ご覧の通り insert
された順に単純に 0 オリジンで key が発行される仕組みです。&+
は加算に用いるオーバーフロー演算子で、オーバーフローを許容してくれます。実際の RxSwift の Bag
の実装はもう少し工夫をして、要素が少ないときに最適化がかかるようなコードになっているようですが、本質を取り出すとこのような実装になるかと思います。実際のコードを読むときはこのあたりの大枠を知っているほうがコードリーディングしやすいと思いますので、ここに記載しておきます。
5. まとめ
- RxSwift は Observer パターンからの発展を考えることによりインターフェースやその実装を導くことができました
- 一見複雑そうに見える構造も、要求される仕様から自ずと導かれるインターフェースとなっていることがわかったのではないでしょうか
- 設計は基本的に問題を解決するものであり、問題を正しく見抜いたり、設定したりすることが、必要な解を見出す近道になるのではないでしょうか
- 複雑そうに見えるパターンも単純な原理原則に分解できるケースが多いのではないでしょうか
というような偉そうなことをいってまとめとしたいと思います。これからもどのような設計がどのような状況で必要なのかを引き続き考えていきたいと思います。RxSwift はそのヒントを与えてくれる良質なコードとなっていると個人的には思います。
6. ライセンス表記
RxSwift は MIT ライセンスで公開されています。記事内のコードはライセンスに基づき、そのまま掲載している箇所や改変して掲載している箇所があります。
The MIT License Copyright © 2015 Krunoslav Zaher All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-
Erich Gamma, Richard Helm, Ralph Johnson, John Vlissides (1994). Design Patterns: Elements of Reusable Object-Oriented Software. ↩
-
https://developer.apple.com/library/content/documentation/Swift/Conceptual/Swift_Programming_Language/Generics.html ↩
-
https://github.com/ReactiveX/RxSwift/blob/007af77912b39d84857a8e90eecdd02dd20164de/RxSwift/Rx.swift#L36 ↩