RxSwift/RxJava入門

* 社内勉強会用の資料です
* 本資料作成時のRxSwift/RxJavaのバージョンは下記です
    * RxSwift 4.0.0
    * RxJava 2.1.6 (with RxKotlin 2.2.0)

---

# 本スライドの対象者

* Rxを聞いたことあるけど、まだ触ってない or 触っては見たけど普段は使ってない
* 下記のどれかの言語を読み書きできる
  * Swift
  * Kotlin

---

# 本スライドの目的

* 「Rxをやってみようかな！」と思って貰える
* 自分で勉強するときのポインタを提供する

---

# Rxとは？

[ReactiveX](http://reactivex.io/) の略

> An API for asynchronous programming
with observable streams

http://reactivex.io/

* 非同期プログラミング用のAPI
* オブザーバーパターンの拡張
* 全てをデータの流れ（ストリーム）として表現する
* ストリームに反応（リアクティブ）して処理をおこなうよう記述する
* 様々な言語向けに提供されいている
    * Java, Kotlin, JavaScript, C#, Scala, Clojure, Swift, etc...

---

## 非同期処理を少ない行数で書ける！

---

# 実例1：インクリメンタルサーチ

![](https://raw.githubusercontent.com/kzaher/rxswiftcontent/master/GithubSearch.gif)

```swift:ViewController.swift
let searchResults = searchBar.rx.text.orEmpty
    .throttle(0.3, scheduler: MainScheduler.instance)
    .distinctUntilChanged()
    .flatMapLatest { query -> Observable<[Repository]> in
        if query.isEmpty {
            return .just([])
        }
        return searchGitHub(query)
            .catchErrorJustReturn([])
    }
    .observeOn(MainScheduler.instance)

searchResults
    .bind(to: tableView.rx.items(cellIdentifier: "Cell")) {
        (index, repository: Repository, cell) in
        cell.textLabel?.text = repository.name
        cell.detailTextLabel?.text = repository.url
    }
    .disposed(by: disposeBag)
```

https://github.com/ReactiveX/RxSwift#usage

---

# 実例2：WebAPI

```kotlin:MainActivity.kt
class MainActivity : AppCompatActivity() {

    lateinit private var binding: ActivityMainBinding
    private val disposable = CompositeDisposable()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        binding = DataBindingUtil.setContentView(this, R.layout.activity_main)

        val retrofit = Retrofit.Builder()
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .baseUrl("https://api.github.com/")
                .build()

        val service = retrofit.create(GitHubService::class.java)

        service.listRepos("tomoya0x00")
                .subscribeOn(Schedulers.io())
                .retry(3)
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext { binding.textView.text = it.joinToString("\n") { it.name } }
                .subscribe()
                .addTo(disposable)        
    }

    override fun onDestroy() {
        super.onDestroy()
        disposable.dispose()
    }
}

interface GitHubService {
    @GET("users/{user}/repos")
    fun listRepos(@Path("user") user: String): Observable<List<Repo>>
}

data class Repo(
        val id: Int,
        val name: String
)
```

[RxJava+RetrofitでAPI通信周りを実装するうえで最低限の知識を30分で詰め込む](https://qiita.com/FumihikoSHIROYAMA/items/201536d9b45ef21b6bc7)

---

# あと嬉しいこと

* iOSでもAndroidでも同じような書き方ができる

---

# 主要な登場人物

* Observable
* Operator
* Subject
* Scheduler
* Disposable

---

# [Observable](http://reactivex.io/documentation/observable.html)
* Observeできる何か
* Subscribeすることでデータが流れてくる

```kotlin:main.kt
// 1 2 3
Observable.create<Int> { e ->
    e.onNext(1)
    e.onNext(2)
    e.onNext(3)
    e.onComplete()
}.subscribe(::println)

// 1 2 3
Observable.fromArray(1, 2, 3)
        .subscribe(::println)

// 2
Observable.just(2)
        .subscribe(::println)

// 2
Observable.fromCallable({ 1 + 1 })
        .subscribe(::println)
```

```swift:main.swift
// 1 2 3
_ = Observable<Int>.create { observer in
    observer.on(.next(1))
    observer.on(.next(2))
    observer.on(.next(3))
    observer.on(.completed)
    return Disposables.create()
    }.subscribe(onNext: { print($0) })

// 1 2 3
_ = Observable.of(1, 2, 3)
    .subscribe(onNext: { print($0) })

// 2
Observable.just(2)
    .subscribe(onNext: { print($0) })  
```
---

# [Operator](http://reactivex.io/documentation/operators.html)
ストリームに流れてきたデータを加工するもの
（Observableを作成するものも含む）

```kotlin:main.kt
// 2 20 4 40 40 6 60 60
Observable.fromArray(1, 2, 3)
        .map { it * 2 }
        .doOnNext(::println) // 2 4 6
        .flatMap { Observable.just(it * 10) }
        .doOnNext(::println) // 20 40 60
        .filter { it >= 30 }
        .doOnNext(::println) // 40 60
        .subscribe()

val a = Observable.fromArray("a1", "a2", "a3", "a4")
val b = Observable.fromArray("b1", "b2", "b3")

// a1 a2 a3 a4 b1 b2 b3
Observable.merge(a, b)
        .doOnNext(::println)
        .subscribe()

// a1b1 a2b2 a3b3
Observable.zip(a, b, BiFunction { t1: String, t2: String -> t1 + t2 })
        .doOnNext(::println)
        .subscribe()

// a4b1 a4b2 a4b3
Observable.combineLatest(a, b, BiFunction { t1: String, t2: String -> t1 + t2 })
        .doOnNext(::println)
        .subscribe()
```

```swift:main.swift
// 2 20 4 40 40 6 60 60
_ = Observable.of(1, 2, 3)
    .map { $0 * 2 } // 2 4 6
    .do(onNext: { print($0) })
    .flatMap { Observable.just($0 * 10) }
    .do(onNext: { print($0) }) // 20 40 60
    .filter { $0 >= 30 }
    .do(onNext: { print($0) }) // 40 60
    .subscribe()

let a = Observable.of("a1", "a2", "a3", "a4")
let b = Observable.of("b1", "b2", "b3")

// a1 b1 a2 b2 a3 b3 a4
_ = Observable.merge(a, b)
    .do(onNext: { print($0) })
    .subscribe()

// a1b1 a2b2 a3b3
_ = Observable.zip(a, b) { $0 + $1 }
    .do(onNext: { print($0) })
    .subscribe()

// a1b1 a2b1 a2b2 a3b2 a3b3 a4b3
_ = Observable.combineLatest(a, b) { $0 + $1 }
    .do(onNext: { print($0) })
    .subscribe()
```

---

# [Subject](http://reactivex.io/documentation/subject.html)
SubscriberとObservableの二つの機能を持つもの

```kotlin:main.kt
val ps = PublishSubject.create<Int>()

ps.onNext(1)
ps.onNext(2)

ps.subscribe(::println)

ps.onNext(3) // 3
ps.onComplete()

val bs = BehaviorSubject.create<Int>()

bs.onNext(1)
bs.onNext(2)

bs.subscribe(::println) // 2

bs.onNext(3) // 3
bs.onComplete()

val rs = ReplaySubject.create<Int>()

rs.onNext(1)
rs.onNext(2)

rs.subscribe(::println) // 1 2

rs.onNext(3) // 3
rs.onComplete()
```

```swift:main.swift
let ps = PublishSubject<Int>()

ps.on(.next(1))
ps.on(.next(2))

ps.subscribe(onNext: { print($0) })

ps.on(.next(3)) // 3
ps.on(.completed)

let bs = BehaviorSubject<Int>(value: 1)

bs.on(.next(2))

bs.subscribe(onNext: { print($0) }) // 2

bs.on(.next(3)) // 3
bs.on(.completed)

let rs = ReplaySubject<Int>.create(bufferSize: 10)

rs.on(.next(1))
rs.on(.next(2))

rs.subscribe(onNext: { print($0) }) // 1 2

rs.on(.next(3)) // 3
rs.on(.completed)
```

---

# [Scheduler](http://reactivex.io/documentation/scheduler.html)
どのスレッドでSubscribe/Observeするか指定するもの
（別スレッドでネットワーク通信して、メインスレッドでUIに反映など）

![](http://reactivex.io/documentation/operators/images/schedulers.png)
http://reactivex.io/documentation/scheduler.html

```kotlin:main.kt
Observable.just("hoge") // on io Scheduler
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.computation())
        .map { "$it fuga" } // on computation Scheduler
        .observeOn(AndroidSchedulers.mainThread())
        .doOnNext { // on mainThread 
            binding.textView.text = it
        }.subscribe()
```

```swift:ViewController.swift
let scheduler1 = SerialDispatchQueueScheduler(qos: .default)
let scheduler2 = ConcurrentDispatchQueueScheduler(qos: .default)

_ = Observable.just("hoge") // on scheduler1
    .subscribeOn(scheduler1)
    .observeOn(scheduler2)
    .map { "\($0) fuga" } // on scheduler2
    .observeOn(MainScheduler.instance)
    .do(onNext: { [unowned self] in // MainScheduler
        self.textLabel.text = $0
    }).subscribe()

// with RxCocoa
_ = Observable.just("hoge")  // on scheduler1
    .subscribeOn(scheduler1)
    .observeOn(scheduler2)
    .map { "\($0) fuga" }  // on scheduler1
    .asDriver(onErrorDriveWith: Driver.empty())
    .drive(textLabel.rx.text) // MainScheduler
```

---

# Disposable
リソース開放用のインターフェース
subscribeの戻り値で、これをdisposeするとリソースが開放される
（メモリリーク防止のため、Activity/Fragment/ViewControllerのライフサイクルに合わせて開放するなど）

```kotlin:main.kt
val hello = Observable.just("hello").delay(10, TimeUnit.SECONDS)
val disposable = hello.subscribe()

println(disposable.isDisposed) // false
disposable.dispose()
println(disposable.isDisposed) // true
```

```swift:main.swift
let hello = Observable.just("hello").delay(10, scheduler: MainScheduler.instance)
let disposable: Disposable = hello.subscribe()

disposable.dispose()
```

---

# Disposableの実用例(iOS)

[DisposeBag](https://github.com/ReactiveX/RxSwift/blob/master/Documentation/GettingStarted.md#dispose-bags)でまとめて管理する
> Dispose bags are used to return ARC like behavior to RX.
> When a DisposeBag is deallocated, it will call dispose on each of the added disposables.

```swift:ViewController.swift
class ViewController: UIViewController {

    @IBOutlet weak var textLabel: UILabel!
    
    private let disposeBag = DisposeBag()
    
    override func viewDidLoad() {
        super.viewDidLoad()
        
        disposeBag.insert(showAfter(seconds: 5))
        disposeBag.insert(showAfter(seconds: 10))
        disposeBag.insert(showAfter(seconds: 15))
    }

    private func showAfter(seconds: RxTimeInterval) -> Disposable {
        return Observable.just("\(seconds) seconds")
            .delay(seconds, scheduler: MainScheduler.instance)
            .asDriver(onErrorDriveWith: Driver.empty())
            .drive(self.textLabel.rx.text)
    }
}
```

---

# Disposableの実用例(Android)

[CompositeDisposable](http://reactivex.io/RxJava/javadoc/io/reactivex/disposables/CompositeDisposable.html)でまとめて管理する
> A disposable container that can hold onto multiple other disposables and offers O(1) add and removal complexity.

```kotlin:MainActivity.kt
class MainActivity : AppCompatActivity() {

    lateinit private var binding: ActivityMainBinding
    private val disposable = CompositeDisposable()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        binding = DataBindingUtil.setContentView(this, R.layout.activity_main)

        disposable.add(
                showAfter(5)
        )

        disposable.addAll(
                showAfter(10),
                showAfter(15)
        )
    }

    override fun onDestroy() {
        super.onDestroy()
        disposable.dispose()
    }

    private fun showAfter(seconds: Long): Disposable {
        return Observable.just("$seconds seconds")
                .delay(seconds, TimeUnit.SECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext {
                    binding.textView.text = it
                }.subscribe()
    }
}
```

---

# エラーハンドリング

* subscribe(onError: {})でハンドリングするか、別の値かストリームに変換する
* リトライも可能（retryWhen()で細かい条件も実現可）

```kotlin:main.kt
val o = Observable.create<Int> { e ->
    e.onNext(1)
    e.onNext(2)
    e.onError(Throwable("hoge"))
    e.onNext(3)
    e.onComplete()
}

// 1 2 io.reactivex.exceptions.OnErrorNotImplementedException: hoge
o.subscribe(::println)

// 1 2 error!(java.lang.Throwable: hoge)
o.subscribe(::println, // onNext
        { println("error!($it)") }) // onError

// 1 2 error!(java.lang.Throwable: hoge) io.reactivex.exceptions.OnErrorNotImplementedException: hoge
o.doOnError { println("error!($it)") }.subscribe(::println)

// 1 2 1 2 io.reactivex.exceptions.OnErrorNotImplementedException: hoge
o.retry(2).subscribe(::println)

// 1 2 0
o.onErrorReturn { 0 }.subscribe(::println)

// 1 2 10 20
o.onErrorResumeNext(Observable.fromArray(10, 20)).subscribe(::println)

// 1 2
o.onErrorResumeNext(Observable.empty()).subscribe(::println)
```

```swift:main.swift
let o = Observable<Int>.create { observer in
    observer.on(.next(1))
    observer.on(.next(2))
    observer.on(.error(NSError(domain: "hoge", code: 0)))
    observer.on(.next(3))
    observer.on(.completed)
    return Disposables.create()
}

// 1 2 Unhandled error happened: Error Domain=hoge Code=0 "(null)"
_ = o.subscribe(onNext: { print($0) })

// 1 2 error!(Error Domain=hoge Code=0 "(null)")
_ = o.subscribe(onNext: { print($0) },
                onError: { print("error!(\($0))") })
// 1 2 error!(Error Domain=hoge Code=0 "(null)") Unhandled error happened: Error Domain=hoge Code=0 "(null)"
_ = o.do(onError: { print("error!(\($0))") }).subscribe(onNext: { print($0) })

// 1 2 1 2 Unhandled error happened: Error Domain=hoge Code=0 "(null)"
_ = o.retry(2).subscribe(onNext: { print($0) })

// 1 2 0
_ = o.catchErrorJustReturn(0).subscribe(onNext: { print($0) })

// 1 2 10 20
_ = o.catchError { _ in Observable.of(10, 20) } .subscribe(onNext: { print($0) })

// 1 2
_ = o.catchError { _ in Observable.empty() } .subscribe(onNext: { print($0) })
```

---

# 関連用語

* [Backpressure](https://github.com/ReactiveX/RxJava/wiki/Backpressure)
    * 流速制御の仕組み
    * 参考：[詳解RxJava2：Backpressureで流速制御](https://qiita.com/yuya_presto/items/0e95271bc85efe7f768e)
* [Hot / Cold observables](https://github.com/Froussios/Intro-To-RxJava/blob/master/Part%203%20-%20Taming%20the%20sequence/6.%20Hot%20and%20Cold%20observables.md)
    * Observableの性質
        * Coldはsubscribeされるまで動作しない＆subscribeした回数分、ストリーム生成される
            * 大体のOperatorはこちら
        * Hotはsubscribeに関係なく動作する＆ストリームは一つのみ
            * Subjectはこちら
        * Cold->Hot変換できる
    * 参考：[今日こそ理解するHot / Cold @社内RxSwift勉強会](https://www.slideshare.net/yukitakahashi3139241/hot-cold)
* Single
    * 結果は成功(with an item)か失敗
    * JavaScriptのPromise的なもの
* Completable
    * 結果は完了(with no item)か失敗
    * reactive runnable
* Maybe
    * 結果は成功(with an item)か完了(with no item)か失敗
    * reactive optional
* Flowable
    * BackpressureありのObservable
* Processor 
    * BackpressureありのSubject


---

# 次に見るべき資料

* 公式Webサイト
    * [ReactiveX](http://reactivex.io/)
    * [RxJava](https://github.com/ReactiveX/RxJava)
        * WikiはRxJava1と2の記述が混在しているので注意
    * [RxSwift](https://github.com/ReactiveX/RxSwift)
* ReactiveXのドキュメントにある図の見方の解説
    * [How to read "marble diagram"](https://www.slideshare.net/yanzm/how-to-read-marble-diagram/yanzm/how-to-read-marble-diagram)
* Jake神によるRxJava親切丁寧説明スライド
    * [Managing the Reactive World with RxJava](https://speakerdeck.com/jakewharton/managing-the-reactive-world-with-rxjava-philly-ete-2017)