RxSwift入門(2) 非同期処理してみる

はじめに

「オブザーバーパターンから始めるRxSwift入門」の続きです。

前の記事では Rx の Hot な Observable のみを使った利用方法を説明しました。今回は Cold な Observable の生成方法や、非同期処理を実現するためのスレッド指定方法などを解説します。

HotなObservableって何？

前の記事で説明した Subject を使ったものが Hot な Observable です。Subject そのものでなくても、例えば Variable のように内部で Subject を使っているものもあります。

• どこからも subscribe されていなくても動作する
• 複数から subscribe されると同じ動作結果を通知する

という特徴があります。オブサーバーパターンを置き換えるなら、この動作は当たり前ですよね？

上図の PublishSubject はどこからも subscribe されていなくても動作していてイベントを発行しています。それを subscribe すると、その時点以降のイベントを通知してもらえるようになります。複数から subscribe しても、同じタイミングで同じイベントが通知されます。

ColdなObservableって何？

逆に Subject を使わないで作られる Observable は全て Cold です。Rx は Cold の方が基本で、むしろ Hot の方が特殊と言えます。

• subscribe されないと動作せず、subscribe されると動作を開始する
• 複数から subscribe されるとそれぞれ別々に動作する

という特徴があります。なんのこっちゃと思うかもしれませんが、例えば配列の forEach を呼び出す場合を考えてみてください。

[ 1, 2, 3, 4, 5].forEach { value in
  // ...
}

forEach を呼び出すと配列の最初から値を渡してきます。しかも複数が forEach を呼び出すとそれぞれ別々に動作して最初から値を渡してきます。

RxSwift で同じ動作をさせるには以下のようにします。

_ = Observable.of(1, 2, 3, 4, 5).subscribeNext { value in
  // ...
}

Observable.of(1, 2, 3, 4, 5) で作っているのが Cold な Observable になります。subscribe するまでは動作せず、複数から subsribe されるとそれぞれ別々に動作して、1, 2, 3, 4, 5 を順番に渡します。

HotなObservableを使って非同期通信結果を通知してみる

前の記事で紹介した Subject を使う方法で、サーバーからデータを取得するクラスを作ってみます。

class ServerDataLoader {
  private let resultSubject = PublishSubject<NSData>()
  private let request: NSURLRequest
  private var task: NSURLSessionDataTask?

  var result: Observable<NSData> { return resultSubject }

  func start() {
    let configuration = NSURLSessionConfiguration.defaultSessionConfiguration()
    let session = NSURLSession(configuration: configuration)
    let task = session.dataTaskWithRequest(request) { // 完了時の処理
      [resultSubject] data, response, error in
      dispatch_async(dispatch_get_main_queue()) {
        if let data = data {
          resultSubject.onNext(data)
          resultSubject.onCompleted()
        } else {
          resultSubject.onError(error ?? MyError.FailedToFetchServerData)
        }
      }
    }
    task.resume() // 通信開始（非同期でバックグラウンド実行される）
    self.task = task
  }

  func cancel() {
    task?.cancel()
  }

  init(request: NSURLRequest) {
    self.request = request
  }
}

こんな感じで使います。

let loader = ServerDataLoader(request: request)
loader.result.subscribe {
  onNext: { data in
    // データ受信時の処理
  },
  onError: { error in
    // エラー時の処理
  }
}
loader.start()

NSURLSessionが通信をバックグラウンドスレッドで非同期で実行してくれます。そして dataTaskWithRequest に渡す完了時の処理もバックグラウンドスレッドで実行されます。ここではGCDを使って通知をメインスレッドで行うように変更しています。

注意点としては、このクラスは１回限りの使い捨てということです。onError/onCompleted が発生したらその後はイベントが通知されないので、もう一度 start すると、通信は開始しますが結果が通知されません。

Observable.create

今度は同様の処理を Cold な Observable でやってみます。Observable.create は Cold な Observable を作る汎用的な方法です。「subscribe されるとサーバーからデータを取得する処理を開始する」Observable を作って返しています。

func fetchServerDataWithRequest(request: NSURLRequest) -> Observable<NSData> {
  return Observable.create { observer in
    let configuration = NSURLSessionConfiguration.defaultSessionConfiguration()
    let session = NSURLSession(configuration: configuration)
    let task = session.dataTaskWithRequest(request) { data, response, error in
      if let data = data {
        observer.onNext(data)
        observer.onCompleted()
      } else {
        observer.onError(error ?? MyError.FailedToFetchServerData)
      }
    }
    task.resume()
    return AnonymousDisposable { task.cancel() }
  }
}

Observer.create には subscribe されたときに実行する処理を渡します。クロージャの引数には ObserverType 型（上記ではobserverという名前）が渡されます。これに onNext/onError/onCompleted メソッドが用意されているので、それらを呼び出してイベントを通知します。クロージャの戻り値として、dispose されたときに購読解除処理を行う Disposable オブジェクトを返します。

Disposable を作る部分は、上の例なら以下のようにも書けますね。

return AnonymousDisposable(task.cancel)

fetchServerDataWithRequest を呼び出した時点では Observable が生成されるだけで、通信は開始しません。この Observable は Cold なので、subscribe して初めて処理を開始します。

また subscribe する度に新しく処理を開始するため、前の Hot を使ったクラスと違って、何度でも利用することができます。

subscribe すると渡したクロージャ（下図の赤色部分）が実行されます。
先ほどと同様に通信は非同期で実行されるので、完了時に実行されるクロージャ（下図の青色部分）はバックグラウンドスレッドで実行されます。

先ほどと違って GCD でメインスレッドに変換していないので、onNext/onError/onCompleted は通信を行ったバックグラウンドスレッドで通知されます。これをメインスレッドで通知させる Rx 流の方法を次に説明します。

補足：
RxCocoa が NSURLSession に rx_data って拡張プロパティを用意してるんで、それ使えば自分でこんな処理を実装する必要はありません。さらに JSON パースまでやってくれる rx_JSON なんてのもあります。

observeOn

先ほどのバックグラウンドスレッドで通知されるイベントを、メインスレッドで通知してもらうように変換するには observeOn を利用します。

_ = fetchServerDataWithRequest(request)
  .observeOn(MainScheduler.instance)
  .subscribe(
    onNext: { data in
      // データ受信時の処理
    },
    onError: { error in
      // エラー時の処理
    }
  )

observeOn に渡すのは Scheduler オブジェクトで、メインスレッドを表す Scheduler は MainScheduler.instance で取得できます。observeOn を呼び出した部分から後は、指定したスレッドで通知されます。

例えば途中で NSData のパースを行うとします。

_ = fetchServerDataWithRequest(request)
  .observeOn(MainScheduler.instance)
  .map { parse($0) }
  .subscribe(
    onNext: { result in
      // パース済みデータ受信時の処理
    },
    onError: { error in
      // エラー時の処理
    }
  )

このようにすると、パース処理もメインスレッドで行われます。observeOn と map を入れ替えると、

  .map { parse($0) }
  .observeOn(MainScheduler.instance)

パース処理までがバックグラウンドスレッドで行われ、その結果の通知がメインスレッドで行われます。

subscribeOn

名前の通り subscibe 処理を実行するスレッドを指定するのですが、勘違いしてはいけません。subscribe に渡すクロージャがそのスレッドで実行されるのではありません。

subscribe 処理は、

• 購読登録処理を行う
• Cold な Observable ならそのときに動作を開始する

ということを行っています。subscribeOn でその実行スレッドを指定できるわけです。

_ = fetchServerDataWithRequest(request)
  .subscribeOn(ConcurrentDispatchQueueScheduler(
    globalConcurrentQueueQOS: .UserInitiated))
  .map { parse($0) }
  .observeOn(MainScheduler.instance)
  .subscribe(
    onNext: { result in
      // パース済みデータ受信時の処理
    },
    onError: { error in
      // エラー時の処理
    }
  )

subscribeOn で QOS_CLASS_USER_INITIATED での実行を指定しています¹。これで fetchServerDataWithRequest が返す Observable の購読登録処理が指定スレッドで行われます。この Observable は Cold なので、subscribe 時点で create に渡したクロージャ（下記の赤色部分）が指定スレッドで実行されます。

このように subscribeOn を使うと Cold な Observable の動作スレッドを指定することができます。

しかしこの例ではNSURLSession が元々バックグラウンド動作しますので、通信処理は指定スレッドでは実行されません。

そして通信完了時の処理（青色の部分）も通信バックグラウンドスレッドで実行されます。通信バックグラウンドスレッドで通知されたイベントは、observeOn 以降はメインスレッドで通知されるようになり、subscribe に渡したクロージャはメインスレッドで呼び出されます。

このようにイベントの通知は subscribeOn で指定したスレッドで行われるとは限りません。以下のように内部でバックグランド実行しないようなものは、subscribeOn で指定したスレッドで通知されますが・・・。

_ = Observable.of(1, 2, 3, 4, 5)
  .subscribeOn(ConcurrentDispatchQueueScheduler(globalConcurrentQueueQOS: .UserInitiated))
  .subscribeNext { value in
    // 1, 2, 3, 4, 5のループ処理が指定スレッドで行われ、
    // このクロージャも指定スレッドで呼び出される
    // ...
  }

また subscribeOn が subscribe するときの動作スレッドを指定しているのだと理解していれば、subscribe する前から動作している Hot な Observable に subscribeOn を指定してもその動作スレッドには何も影響がないことも理解できるはずです。

リトライ

その名も retry という operator が用意されています。onError が起こると、自動的に subscribe しなおしてくれます。

_ = fetchServerDataWithRequest(request)
  .map { parse($0) }
  .retry(3)
  .observeOn(MainScheduler.instance)
  .subscribe(
    onNext: { result in
      // パース済みデータ受信時の処理
    },
    onError: { error in
      // エラー時の処理
    }
  )

上の例では３回リトライを指定しています。map の後に指定しているので、parse で例外が発生して失敗した場合もリトライします。実は operator による変換で例外が発生した場合、RxSwift は onError で通知してくれるんです。よくできてますね。

ちなみに retry に回数を指定しなかった場合、ずっとリトライし続けます。もっと細かい制御のできる retryWhen もありますが、複雑なのでまた別の機会に。

タイムアウトの指定

timeout という operator があります。第一引数が秒数、第二引数がタイマーを実行するスレッド（Scheduler）の指定です。タイムアウトすると TimeoutError が onError で通知されます。

let lowPriorityScheduler  = ConcurrentDispatchQueueScheduler(
  queue: dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_LOW, 0))

_ = fetchServerDataWithRequest(request)
  .timeout(5, scheduler: lowPriorityScheduler)
  .map { parse($0) }
  .observeOn(MainScheduler.instance)
  .subscribe(
    onNext: { result in
      // パース済みデータ受信時の処理
    },
    onError: { error in
      // エラー時の処理
    }
  )

Scheduler の作成は、先ほど QOS による指定方法を使ったので、今回は dispatch_queue_t による指定にしてみました。

タイムアウトしたら代わりに実行する Observable を指定する other 引数を持つバージョンもあります。

  .timeout(5, other: Observable.just(NSData()), scheduler: lowPriorityScheduler)

Observable.just は指定した値を onNext で１つだけ流して onCompleted になる Observable を生成します。ここではタイムアウトしたら空の NSData を渡して完了するようにしています。

エラーからの復帰

先ほどの other 引数付きの timeout のように、エラーが発生した場合に onError で伝えずに onNext で何かを渡したい場合があります。例えば検索して何かのリストを取得するとします。取得に失敗したら空のリストを渡すようにしたいといったケースです。そんなときは catchErrorJustReturn や catchError が使えます。

catchErrorJustReturn を使って onError が発生したら空の配列を渡すようにしています（ parse の結果が配列で返る想定です）。サーバーからのデータ取得でもパース失敗でも、とくにかく onError になったら空の配列になります。

_ = fetchServerDataWithRequest(request)
  .map { parse($0) }
  .catchErrorJustReturn([])
  .observeOn(MainScheduler.instance)
  .subscribeNext { result in
      // パース済みデータ受信時の処理
  }

catchError の場合はエラー内容によって処理を選択できます。catchError も map と同様に例外が throw されると onError で通知してくれます。

_ = fetchServerDataWithRequest(request)
  .map { parse($0) }
  .catchError { error in
    if error is MyError { throw error }
    return Observable.just([])
  }
  .observeOn(MainScheduler.instance)
  .subscribeNext { result in
      // パース済みデータ受信時の処理
  }

onError のみ発生させる Observable を生成する Observable.error を使って以下のようにも書けます。

  .catchError { error in
    if error is MyError { return Observable.error(error) }
    return Observable.just([])
  }

この先へ

前回はイベント通知に Hot な Observable を使う方法を解説したので、今回は非同期処理に Cold な Observable を使う方法を解説してみました。

Cold な Observable の生成関数はここで紹介した of, create, just, error 以外にも、empty, never, range, deferred などが用意されています。

このようにまだまだ紹介していない operator やクラスが用意されています。しかしここまでの RxSwift の使い方が理解できれば、あとは少しずつ使えるものを増やしていけると思います。

ここまでは RxSwift の使い方を解説してきました。次はさらに使いこなせるように、落とし穴にハマらないように、トラブルにあったときに原因を探れるように、RxSwift の動作をより深く理解することを目的にした

RxSwiftを深く理解する

を読んでみてください。

---

1. iOS8で追加されたQOSによる指定です。 ↩