RxSwiftのソースコードをちょっと読んでみた

# RxSwiftのソースコードを読んでみて

隙間時間を見つけてRxSwiftを勉強し始めて数か月が経つのですが、簡単な実装でも出てくる用語とかオペレータが多く短時間だとなかなか理解が深まりませんでした。

[RxSwift](https://github.com/ReactiveX/RxSwift)のPlaygroundを触ったり、Exampleを触ったり、[マーブルダイアグラム](http://rxmarbles.com/)を見たりして、ようやくある程度理解できるようになってきました。

オペレータについて追ってみたのでまとめてみました。もし間違いなどがあればご指摘いただきたいです。

## RxSwiftのドキュメントについて

RxSwiftのリポジトリには[Documentation](https://github.com/ReactiveX/RxSwift/tree/master/Documentation)が用意されており、いろいろなドキュメントが用意されています。

[HotAndColdObservables.md](https://github.com/ReactiveX/RxSwift/blob/master/Documentation/HotAndColdObservables.md)なんかはHotとColdの違いについて表も用いて説明されているので一度見てみるといいと思います。

RxSwiftで実装されているオペレータについては[API.md](https://github.com/ReactiveX/RxSwift/blob/master/Documentation/API.md)に詳しく書かれています。


### [API.md](https://github.com/ReactiveX/RxSwift/blob/master/Documentation/API.md)に書かれていること

導入部分でさらっと説明が書かれていますが、まとめてみるとこんな感じです。

* いろいろなプラットフォームでいろいろな実装がされており、同じオペレーターでも異なる呼び方をしている
* これは歴史的な理由や言語の予約語が原因だったりする
* オペレーターは状態を持たない
* 目的に応じたAPIのリファレンスが書かれている

ここではオペレータについて記載されています。リンク先はReactiveXのページになっており、マーブルダイアグラムを見ることができます。リンク先のページ下部には言語ごとの実装について記載されていますが、RxSwiftは"TBD"と書かれており、サンプルなどは書かれていません。

[API.md](https://github.com/ReactiveX/RxSwift/blob/master/Documentation/API.md)の[Creating Observables](https://github.com/ReactiveX/RxSwift/blob/master/Documentation/API.md#creating-observables)の`asObservable`のリンク先を見てみると、`asObservable`は[From](http://reactivex.io/documentation/operators/from.html)と言うことが分かったりします。

また、ここに書かれている基本的な関数については、ある程度RxSwiftに含まれている[Playground](https://github.com/ReactiveX/RxSwift/tree/master/Rx.playground/Pages)で動作を確認することができます。

今回は`empty`のソースを追って、`Observable`、`ObserverType`、`AnonymousObserver `などに触れてみました。

## `empty`のソースを追ってみる

`empty`はReactiveXのリンクを見ても["Empty"](http://reactivex.io/documentation/operators/empty-never-throw.html)と紹介されています。

> create an Observable that emits no items but terminates normally

何もアイテムを発行しないが、正常に終了するObservableを作成します。

[Playground](https://github.com/ReactiveX/RxSwift/tree/master/Rx.playground/Pages)では、こんな感じでサンプルが書かれていました。

```swift
let emptySequence = Observable<Int>.empty()
emptySequence
    .subscribe { event in
        print(event) // Completed
}
```

`empty`で作成したObservableに対して`subscribe`を呼び出しています（`subscribe`は後ほど）。

### `Event<Element>`クラス

`empty`は何もアイテムを発行しないので`Completed`と出力されるのですが、これの正体は`Event`というenumになります。

```swift:Event.swift
public enum Event<Element> : CustomDebugStringConvertible {
    /**
    Next element is produced
    */
    case Next(Element)

    /**
    Sequence terminates with error
    */
    case Error(ErrorType)

    /**
    Sequence completes sucessfully
    */
    case Completed
}
```

`Event`は3つのメンバーを持っており、`Next(Element)`、`Error(ErrorType)`、そして`Completed`となっています。`empty`は何もアイテムを発行せずに終了と書かれていたので、`Completed`が完了を意味することが分かります。

### `Observable<Element>`クラス

`empty`の話に戻ります。`empty`の実装は以下のようになっています。`empty`は引数を何も取らずに型`E`の`Observable`を返すことが分かります。

```swift:Observable+Creation.swift
public static func empty() -> Observable<E> {
    return Empty<E>()
}
```
`empty`は`Observable`のextensionです。Observable+Creation.swiftというファイルで定義されており、このファイルには`create`,`just`といった関数も定義されており、ファイル名通り`Observable`を作成するための便利メソッドが定義されています。


`Observable`を見てみましょう（実際はコメントがありますが長くなるので削除しています）。

```swift:Observable.swift
public class Observable<Element> : ObservableType {
    public typealias E = Element

    init() {
#if TRACE_RESOURCES
        OSAtomicIncrement32(&resourceCount)
#endif
    }

    public func subscribe<O: ObserverType where O.E == E>(observer: O) -> Disposable {
        abstractMethod()
    }

    public func asObservable() -> Observable<E> {
        return self
    }

    deinit {
#if TRACE_RESOURCES
        AtomicDecrement(&resourceCount)
#endif
    }

    internal func composeMap<R>(selector: Element throws -> R) -> Observable<R> {
        return Map(source: self, selector: selector)
    }
}
```

`Observable`はジェネリックなクラスになっており、外部公開メソッドとして`subscribe`と`asObservable`を実装しています。

`subscribe`は`abstractMethod()`を実行していますが、`@noreturn`な関数で内部的に`fatalError`を呼び出しています。継承して使うものということでしょうか。

`asObservable`は`self`を返しているだけです。

### `ObservableType`プロトコル

また、`Observable`クラスは`ObservableType`プロトコルに適合していることが分かります。`ObservableType`プロトコルは以下のようになっています（実際はコメントがありますが長くなるので削除しています）。

```swift:ObservableType.swift

public protocol ObservableType : ObservableConvertibleType {

    associatedtype E

    @warn_unused_result(message="http://git.io/rxs.ud")
    func subscribe<O: ObserverType where O.E == E>(observer: O) -> Disposable

}
```
関連型`E`を持ち、`subscribe`メソッドを実装することが、`ObservableType`プロトコルには必要であることが分かります。

`subscribe`できることが`Observable`=観察可能である、という性質を表しているということだと思います。

### `Disposable`プロトコル

`subscribe`は`ObserverType`を引数に取り、`Disposable`を返します。この時`ObserverType`の関連型と`ObservableType`の関連型は同じである必要があります。

`Disposable`はプロトコルです。

```swift:Disposable.swift
/**
Respresents disposable resource.
*/
public protocol Disposable {
    /**
    Dispose resource.
    */
    func dispose()
}
```

`dispose`を実装することでリソースの処分を行うみたいです。

### `ObserverType`プロトコル

`ObserverType`プロトコルも見てみます。`ObserverType`も関連型`E`を持ち、`on`メソッドの実装が求められます。`Event`が再登場しました。

```swift:ObserverType.swift
public protocol ObserverType {

    /**
    The type of elements in sequence that observer can observe.
    */
    associatedtype E

    /**
    Notify observer about sequence event.

    - parameter event: Event that occured.
    */
    func on(event: Event<E>)
}
```

`on`メソッドのはオブザーバー=観察者に対してイベントを通知するためのメソッドのようです。

`ObserverType`には`Event`各種に対する便利メソッドのデフォルト実装が用意されていました。

```swift:ObserverType.swift
public extension ObserverType {

    /**
    Convenience method equivalent to `on(.Next(element: E))`

    - parameter element: Next element to send to observer(s)
    */
    final func onNext(element: E) {
        on(.Next(element))
    }

    /**
    Convenience method equivalent to `on(.Completed)`
    */
    final func onCompleted() {
        on(.Completed)
    }

    /**
    Convenience method equivalent to `on(.Error(error: ErrorType))`
    - parameter error: ErrorType to send to observer(s)
    */
    final func onError(error: ErrorType) {
        on(.Error(error))
    }
}
```

### `Empty`クラス

さて、`Observable`がなんとなくわかったので、`empty`を再度見てみましょう（`Observable`を見るだけだったのですがだいぶ遠回りしました）。

`empty`メソッドは`Observable<E>`を戻り値として返しています。

```swift:Observable+Creation.swift
public static func empty() -> Observable<E> {
    return Empty<E>()
}
```

`Empty`のインスタンスを返していますが、`Empty`は`Observable`のサブクラスなのでしょう。

この`Empty`は以下のように定義されていました。`Empty`クラスは`Producer`クラスを継承したもので、`subscribe`メソッドをオーバーライドしています（`Producer`は`Observable`のサブクラスです）。

```swift:Empty.swift
class Empty<Element> : Producer<Element> {
    override func subscribe<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
        observer.on(.Completed)
        return NopDisposable.instance
    }
}
```

先ほど出てきた`ObserverType`の`on`に`Event`の`.Completed`を渡していますね。`Event`には`.Next`も`.Error`もありますが、どちらも渡していません。

`Empty`は`public`になっていないのでRxSwiftをモジュールとして使う際には使えません。

### `NopDisposable`クラス

`NopDisposable`は何もオペレーションの必要がない`Disposable`でシングルトンになっていました。何もしない`Disposable`みたいです。

```swift:NopDisposable.swift
public struct NopDisposable : Disposable {
 
    /**
    Singleton instance of `NopDisposable`.
    */
    public static let instance: Disposable = NopDisposable()
    
    init() {
        
    }
    
    /**
    Does nothing.
    */
    public func dispose() {
    }
}
```

### ここまでのまとめ

ここまでをまとめると、観察可能な`ObservableType`に適合した`Observable<E>`を継承した`Empty`のインスタンスを`empty`メソッドの戻り値とすることでイベントの監視を可能にしています。

```swift:Observable+Creation.swift
public static func empty() -> Observable<E> {
    return Empty<E>()
}
```

### `subscribe(on: (event: Event<E>) -> Void) -> DisposableはObservable<Element>`メソッド

ここでようやくPlaygroundのサンプルに戻ります。

```swift
let emptySequence = Observable<Int>.empty()
emptySequence
    .subscribe { event in
        print(event) // Completed
}
```
`Observable<Int>.empty()`の流れはここまで書いてきたとおりです。`Observable<Int>`は観察可能なクラスで、`empty`クラスメソッドによって`.Completed`の`Event`のみのシーケンスが生成されます。それが`emptySequence`です。

`Empty`は`subscribe`を実装しているので、`subscribe`を呼び出すことができます。

と思いきや、Playgroundで使用されている`subscribe`は`Empty`クラスに定義されている`subscribe`とは引数が異なります。

`Empty`はこのように実装されていました。`ObserverType`を引数として受け取って`Disposable`を戻り値としていました。

```swift:Empty.swift
class Empty<Element> : Producer<Element> {
    override func subscribe<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
        observer.on(.Completed)
        return NopDisposable.instance
    }
}
```

しかしPlaygroundで使われていたのは以下の`subscribe`です。`Event`を引数に`Void`を返す関数を第一引数とし、戻り値は`Disposable`型です。

```swift:Observable+Extensions.swift
@warn_unused_result(message="http://git.io/rxs.ud")
public func subscribe(on: (event: Event<E>) -> Void)
    -> Disposable {
    let observer = AnonymousObserver { e in
        on(event: e)
    }
    return self.subscribeSafe(observer)
}
```

#### `AnonymousObserver`クラス

引数として渡された`on`は`AnonymousObserver`に渡されています。このクラスは`ObserverBase<ElementType>`のサブクラスとなっており、`ObserverBase`は`ObserverType`と`Disposable`に適合しています。

```swift:AnonymousObserver.swift
class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
    typealias Element = ElementType
    
    typealias EventHandler = Event<Element> -> Void
    
    private let _eventHandler : EventHandler
    
    init(_ eventHandler: EventHandler) {
#if TRACE_RESOURCES
        AtomicIncrement(&resourceCount)
#endif
        _eventHandler = eventHandler
    }

    override func onCore(event: Event<Element>) {
        return _eventHandler(event)
    }
    
#if TRACE_RESOURCES
    deinit {
        AtomicDecrement(&resourceCount)
    }
#endif
}
```

`AnonymousObserver`は`Event<Element> -> Void`型の値をプロパティとして保持しており、オーバーライドしている`onCore`が、スーパークラスである`ObserverBase<ElementType>`の`on`で呼び出されています。

`ObserverBase<ElementType>`の`onCore`はオーバーライドされることが前提で`abstractMethod()`が呼ばれています(いまのところ`AnonymousObserver`以外に`ObserverBase<ElementType>`を継承しているクラスはありませんでした)。

```swift:ObserverBase.swift
class ObserverBase<ElementType> : Disposable, ObserverType {
    typealias E = ElementType

    private var _isStopped: AtomicInt = 0

    func on(event: Event<E>) {
        switch event {
        case .Next:
            if _isStopped == 0 {
                onCore(event)
            }
        case .Error, .Completed:

            if !AtomicCompareAndSwap(0, 1, &_isStopped) {
                return
            }

            onCore(event)
        }
    }

    func onCore(event: Event<E>) {
        abstractMethod()
    }

    func dispose() {
        _isStopped = 1
    }
}
```

イベントを受け取って`ObserverType`の`on`を実行するのが`AnonymousObserver`の役割っぽいです。


```swift:Observable+Extensions.swift
@warn_unused_result(message="http://git.io/rxs.ud")
public func subscribe(on: (event: Event<E>) -> Void)
    -> Disposable {
    let observer = AnonymousObserver { e in
        on(event: e)
    }
    return self.subscribeSafe(observer)
}
```

#### `subscribeSafe`メソッド

`AnonymousObserver`のインスタンスは`subscribeSafe`メソッドの引数として渡されています。

```swift:Observable+Extensions.swift
public extension ObservableType {
    /**
    All internal subscribe calls go through this method
    */
    @warn_unused_result(message="http://git.io/rxs.ud")
    func subscribeSafe<O: ObserverType where O.E == E>(observer: O) -> Disposable {
        return self.asObservable().subscribe(observer)
    }
}
```

> All internal subscribe calls go through this method

インターナルなsubscribeはすべてこのメソッドを通して呼ばれるそうで、`Empty`で定義していた以下の`subscribe`はここで呼び出されているようですね（この"All internal subscribe"がどこまでをカバーしているのかは確認していません）。

```swift:Empty.swift
class Empty<Element> : Producer<Element> {
    override func subscribe<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
        observer.on(.Completed)
        return NopDisposable.instance
    }
}
```

## コードリーディングまとめ

以下のコードを軽くまとめます。

```swift
let emptySequence = Observable<Int>.empty()
emptySequence
    .subscribe { event in
        print(event) // Completed
}
```

整理すると、

1. `Observable<Int>.empty()`で空のシーケンスが作成される
  * `Empty<E>`のインスタンスが生成されている
  * `Empty<E>`は`Observable<Element>`のサブクラス
  * `Observable<Element>`=監察可能`（subscribe`を実装しているので）
2. `subscribe(on: (event: Event<E>) -> Void)
    -> Disposable`で流れてくるイベントを処理する
 * `subscribe(on: (event: Event<E>) -> Void)
    -> Disposable`は`Observable<Element>`のextensionで定義されている
  * 引数として渡しているクロージャは`AnonymousObserver`がプロパティとして保持
  * `AnonymousObserver`インスタンスが`subscribeSafe`メソッドの引数に渡される
  * `subscribeSafe`メソッド内では`Empty`の`subscribe`メソッドが実行される
  * `Empty`の`subscribe`は渡された`observer`の`on`を実行し、引数に`Completed`を渡している

以上のようになるかと思います。

## まとめ

出てくる用語を一つずつ整理すると今回のようにたくさん寄り道をしてソースを読まなければならないので、[API.md](https://github.com/ReactiveX/RxSwift/blob/master/Documentation/API.md)や[マーブルダイアグラム](http://rxmarbles.com/)に目を通してRxの用語を理解しておくともっとスムーズになるかもしれません。

コード的にはclassによる継承が結構使われている印象を受けました。
実装を見るとどのように動くのか分かって良いですね。