RxSwiftのobserveOnとsubscribeOnを理解する

RxSwiftのobserveOnと subscribeOnを理解する 大庭 慎一郎 株式会社メルカリ / 株式会社ソウゾウ 2016/4/14 RxSwift勉強会 ＠ Sansan 1

自己紹介 大庭 慎一郎 ooba / bricklife 株式会社メルカリ 2013年4月入社 現在は株式会社ソウゾウへ出向中 「メルカリ」iOS版の立ち上げ 「メルカリ アッテ」iOS版の立ち上げ 2

メルカリ アッテ 「なんでも募集できる 地域コミュニティアプリ」 2015年10月 開発開始 2016年2月 紹介制で公開 2016年3月 正式オープン RxSwiftを全面採用 3

observeOnとsubscribeOn 4

observeOnに関する巷の説明 observeOnよりあとに書かれているオペ レータの実行スレッドを決めるもの わかりやすい 実際正しい 5

subscribeOnに関する巷の説明 ストリーム全体の実行スレッドを決めるも の？ subscribeOnより前に書かれているオペ レータの実行スレッドをきめるもの？ 何回書いても一番上のsubscribeOnだけが 有効になる？ 6

subscribeOnのファーストインプレッション 普段「subscribe」という単語をみるのは ストリームの購読部分だけ そのため、最初見た時に僕は subscribeNextなどの購読部分の実行ス レッドを決めるものだと思った 大きな間違い 7

公式の説明 8 http://reactivex.io/documentation/scheduler.html

• http://reactivex.io/documentation/scheduler.html

簡単な使用例 コード： Observable.just(1) .subscribeOn(backgroundScheduler) .map { $0 * 2 } .observeOn(MainScheduler.instance) .subscribe { print($0) } 出力： Next(2) Completed 9

簡単な使用例 コード： Observable.just(1) .subscribeOn(backgroundScheduler) .map { $0 * 2 } バックグラウンドスレッドで実行 .observeOn(MainScheduler.instance) .subscribe { print($0) } 出力： Next(2) Completed 10

簡単な使用例 コード： Observable.just(1) .subscribeOn(backgroundScheduler) .map { $0 * 2 } .observeOn(MainScheduler.instance) .subscribe { print($0) } 出力： Next(2) Completed 11 メインスレッドで実行

Observerとは 12

Observerとは on()というメソッドが実装されている Observableがそれを使ってイベントを伝達 する すなわちObserverType 13

ObserverType protocol ObserverType { associatedtype E func on(event: Event<E>) } 14

Observableとは 15

Observableとは subscribe()というメソッドが実装されてい る Observerがそれを使って購読できる すなわちObservableType 16

ObservableType

protocol ObservableType : ObservableConvertibleType {
associatedtype E

}

17

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

オペレータとは 18

オペレータとは Observableをsubscribeして、新しい Observableを作るもの ObserverとObservableの両方の働きをす る RxSwiftでは全てのオペレータにクラスが 定義されている 19

たとえばmap map() = Map + MapSink https://github.com/ReactiveX/RxSwift/blob/2.4/ RxSwift/Observable.swift https://github.com/ReactiveX/RxSwift/blob/2.4/ RxSwift/Observables/Implementations/Map.swift MapはObservable MapSinkはObserver 20

• https://github.com/ReactiveX/RxSwift/blob/2.4/RxSwift/Observable.swift
• https://github.com/ReactiveX/RxSwift/blob/2.4/RxSwift/Observables/Implementations/Map.swift

observeOnやsubscribeOnは？ もちろんobserveOnやsubscribeOnにもク ラスが定義されている https://github.com/ReactiveX/RxSwift/blob/2.4/ RxSwift/Observables/Implementations/ ObserveOn.swift https://github.com/ReactiveX/RxSwift/blob/2.4/ RxSwift/Observables/Implementations/ SubscribeOn.swift 21

• https://github.com/ReactiveX/RxSwift/blob/2.4/RxSwift/Observables/Implementations/ObserveOn.swift
• https://github.com/ReactiveX/RxSwift/blob/2.4/RxSwift/Observables/Implementations/SubscribeOn.swift

オペレータがやっていること

observeOnやsubscribeOnを使わない例 Observable.just(1) .map { $0 * 2 } .subscribe { print($0) } Just MapSink Map AnonymousObserver 23

処理の流れ Just MapSink Map AnonymousObserver 24

処理の流れ 1. AnonymousObserverがMapのsubscribe()を実行 Just subscribe() ↑ MapSink Map AnonymousObserver 25

処理の流れ 1. AnonymousObserverがMapのsubscribe()を実行 2. MapSinkがJustのsubscribe()を実行 subscribe() ↑ subscribe() ↑ Just MapSink Map AnonymousObserver 26

処理の流れ 1. AnonymousObserverがMapのsubscribe()を実行 2. MapSinkがJustのsubscribe()を実行 3. JustがMapSinkのon()を実行 subscribe() ↑ subscribe() ↑ Just MapSink Map AnonymousObserver 27 ↓ on()

処理の流れ 1. AnonymousObserverがMapのsubscribe()を実行 2. MapSinkがJustのsubscribe()を実行 3. JustがMapSinkのon()を実行 4. MapがAnonymousObserverのon()を実行 ※厳密にはちょっと違う subscribe() ↑ subscribe() ↑ Just MapSink Map AnonymousObserver 28 ↓ on() ↓ on()

observeOnやsubscribeOnを使った例 Observable.just(1) .subscribeOn(backgroundScheduler) .map { $0 * 2 } .observeOn(MainScheduler.instance) .subscribe { print($0) } 29

処理の流れ Just SubscribeOnSink SubscribeOn MapSink Map ObserveOnSink ObserveOn AnonymousObserver 30

処理の流れ 1. AnonymousObserverがObserveOnのsubscribe()を実 行 Just SubscribeOnSink SubscribeOn MapSink Map subscribe() ↑ ObserveOnSink ObserveOn AnonymousObserver 31

処理の流れ 2. ObserveOnSinkがMapのsubscribe()を実行 Just SubscribeOnSink SubscribeOn subscribe() ↑ subscribe() ↑ MapSink Map ObserveOnSink ObserveOn AnonymousObserver 32

処理の流れ 3. MapSinkがSubscribeOnのsubscribe()を実行 Just subscribe() ↑ subscribe() ↑ subscribe() ↑ SubscribeOnSink SubscribeOn MapSink Map ObserveOnSink ObserveOn AnonymousObserver 33

処理の流れ 4. SubscribeOnSinkがJustのsubscribe()を backgroundScheduler上で実行 subscribe() ↑ subscribe() ↑ subscribe() ↑ subscribe() ↑ Just SubscribeOnSink SubscribeOn MapSink Map ObserveOnSink ObserveOn AnonymousObserver 34

処理の流れ 5. JustがSubscribeOnSinkのon()をbackgroundScheduler 上で実行 subscribe() ↑ subscribe() ↑ subscribe() ↑ subscribe() ↑ Just SubscribeOnSink SubscribeOn MapSink Map ObserveOnSink ObserveOn AnonymousObserver 35 ↓ on()

処理の流れ 6. SubscribeOnがMapSinkのon()をbackgroundScheduler 上で実行 subscribe() ↑ subscribe() ↑ subscribe() ↑ subscribe() ↑ Just SubscribeOnSink SubscribeOn MapSink Map ObserveOnSink ObserveOn AnonymousObserver 36 ↓ on() ↓ on()

処理の流れ 7. MapがObserveOnSinkのon()をbackgroundScheduler 上で実行 subscribe() ↑ subscribe() ↑ subscribe() ↑ subscribe() ↑ Just SubscribeOnSink SubscribeOn MapSink Map ObserveOnSink ObserveOn AnonymousObserver 37 ↓ on() ↓ on() ↓ on()

処理の流れ 8. ObserveOnがAnonymousObserverのon()を MainScheduler上で実行 subscribe() ↑ subscribe() ↑ subscribe() ↑ subscribe() ↑ Just SubscribeOnSink SubscribeOn MapSink Map ObserveOnSink ObserveOn AnonymousObserver 38 ↓ on() ↓ on() ↓ on() ↓ on()

要するに • subscribeOnは以降(Observable方向)の subscribe()の実行スレッドを決める • observeOnは以降(Observer方向)のon() の実行スレッドを決める というのが正しい理解（のはず） 39

subscribeOnが効かないケース

subscribeOnが効かないケース その１ NSURLSessionのrx_response 41 • 実行スレッドはNSURLSessionの設定で決まる • リクエストを投げるところでは効く

NSURLSessionのrx̲response
func rx_response(request: NSURLRequest) -> Observable<(NSData, NSHTTPURLResponse)> {
return Observable.create { observer in
let task = self.dataTaskWithRequest(request) { (data, response, error) in
guard let response = response, data = data else {
observer.on(.Error(error ?? RxCocoaURLError.Unknown))
return
}
guard let httpResponse = response as? NSHTTPURLResponse else {
observer.on(.Error(RxCocoaURLError.NonHTTPResponse(response: response)))
return
}

}

observer.on(.Next(data, httpResponse))
observer.on(.Completed)

let t = task
t.resume()

}

}

この部分の実⾏スレッドはRxの機能では指定できない

return AnonymousDisposable {
task.cancel()
}

https://github.com/ReactiveX/RxSwift/blob/2.4/
RxCocoa/Common/Observables/NSURLSession+Rx.swift
42

• https://github.com/ReactiveX/RxSwift/blob/2.4/RxCocoa/Common/Observables/NSURLSession Rx.swift

subscribeOnが効かないケース その２ すでに実行スレッドが決まっているもの • Hot Observable全般 - PublishSubjectやBehaviorSubjectなどのSubject系 - shareReplay(1)などのオペレータによってHot化された Observable 43 • UIKit系のRx拡張 • Driver

知見 挙動の理解に自信がなかったらソースを 読むといい（あたりまえ） 44

ありがとうございました！