Introduction to ReactiveX/RxSwift

rxswift.org

Introduction to ReactiveX/RxSwift

RxSwiftは、ReactiveX(Rx)のswift実装である。この投稿ではRxの概要をRxSwiftを用いて解説する。

Rxを簡単に説明すると、下記のものを組み合わせたライブラリである。

Rx = Observable + Operators + Scheduler

基本的にはObserverパターンの拡張であり、非同期データストリームであるObservableに対して LINQライクなクエリを適用して、その結果をイベントとしてObserverで受け取ることが出来る。 また各処理は、Schedulerを使ってどのスレッドで実行するかを指定出来る。

Observable

Observableは観測対象であり、非同期なデータストリームを扱う。データストリームの具体例としては、 画面のタップイベントや、キーボード入力が挙げられる。

先にも述べたように、基本的にはObserverパターンの拡張であるので、Rxでは観測対象(Observable)のデータストリームを 観測者(Observer)が購読(subscribe)するという形式をとる。

// 観測対象
protocol ObservableType {
    typealias E
    func subscribe<O: ObserverType where O.E == E>(observer: O) -> Disposable
}
// 購読解除(unsubscribe)の仕組み
protocol Disposable {
    func dispose()
}
// 観測者
protocol ObserverType {
    typealias E
    func on(event: Event<E>)
}
// 受け取るイベント
enum Event<Element> {
    case Next(Element)
    case Error(ErrorType)
    case Completed
}

ObservableからObserverに通知されるイベントは3種類ある。

• Event.Next
  • Observableに新しいデータが来るたびに呼ばれる
• Event.Completed
  • ストリームが終了する時に呼ばれる
• Event.Error
  • エラーが起きると呼ばれる

CompletedまたはErrorが呼ばれると、それ以降イベントは発生せず、リソースは開放される。 有限時間内にストリームが終了しなかった場合は、明示的にリソースを開放する必要がある。 RxSwiftではDisposeBagオブジェクトを利用してこれを実現している。

import RxSwift

class ViewController: UIViewController {
    let disposeBag = DisposeBag()

    override func viewDidLoad() {
        let observable: Observable<Int> = [1, 2, 3].toObservable
        _ = observable.subscribe(onNext: { (event) in print(event) },
                                 onError: nil,
                                 onCompleted: nil,
                                 onDisposed: nil)
                      // disposeBagのメモリが解放、つまりViewControllerが解放されたタイミングでunsubscribeされる
                      .addDisposableTo(disposeBag)
    }
}

Operators

ObservableTypeには、subscribeの他にmapやflatMap等の高階関数がextensionで多数用意されている。 これらはObservableを操作するためのクエリ演算子で、各演算子は以下の図で表される。これをマーブルダイアグラムと呼ぶ。

マーブルダイアグラムについては、rxmarbles.com のページに学習用としていくつか載っているので、 そちらを参考にすると良い。

Scheduler

ObservableTypeには、Schedulerを指定するメソッドも用意されている。

extension ObservableType {
    // これ以降の処理は指定のSchedulerで動く
    public func observeOn(scheduler: ImmediateSchedulerType) -> RxSwift.Observable<Self.E>
    // これ以前の処理は指定のSchedulerで動く
    public func subscribeOn(scheduler: ImmediateSchedulerType) -> RxSwift.Observable<Self.E>
}

例えば、

func a()

APIを叩いてデータを取得

func b()

データのパース

func c()

パースしたデータ中の画像URLをカスタムImageViewにセットする

とした時に、

observable.observeOn(SerialDispatchQueueScheduler(globalConcurrentQueueQOS: .Default))
          .a()
          .b()
          .observeOn(MainScheduler.instance)
          .c()

とすると、aとbはバックグラウンドで、cはメインスレッドで実行される。

Reference

• 公式サイト
• RxSwiftのレポジトリ