RxJavaを使っている既存アプリに Kotlin Coroutinesを導入しよう

Kotlin FestのLTのネタです。(3分でどこまで話せるか謎です。)

# きっかけ
RxJavaとCoroutinesについてJakeさんが以下のように言っていて、確かにsuspend functionのほうがわかりやすく書けると自分も感じています。

<blockquote class="twitter-tweet" data-lang="ja"><p lang="en" dir="ltr">I don&#39;t use Single, Maybe, or Completable anymore since suspend functions are superior. Channels do not replace Observables.</p>&mdash; Jake Wharton (@JakeWharton) <a href="https://twitter.com/JakeWharton/status/1025429808350720002?ref_src=twsrc%5Etfw">2018年8月3日</a></blockquote>
<script async src="https://platform.twitter.com/widgets.js" charset="utf-8"></script>

また既存のアプリの実装を思い浮かべたときにほとんどのRxJavaを使っているところがSingleまたはCompletableなどだったので、置き換えたほうが良さそうでは？と思いました。
個人的にはRxJavaのObservableで使っているところは、Coroutinesではそこまでスマートにはかけなさそうで、そのままでもいいかなという感じです。

そこですごく簡単なアプリで、移行するサンプルアプリを書いてみました。

# 例となるRxJavaのアプリ

簡単な例で、ViewModelがRxJavaを使ってApiクラスから情報を取得して、LiveDataに流すだけです。

<img src="https://qiita-image-store.s3.amazonaws.com/0/27388/2d0e4ca8-04da-41af-64bf-0d5fda82fa0f.png" width="400" />

```kotlin
class ViewModel(
    private val api: Api = Api(),
    private val scheduler: Scheduler = Schedulers.io()
) {
  private val mutablePersons = MutableLiveData<List<Person>>()
  val persons: LiveData<List<Person>> = mutablePersons

  fun onCreate() {
    api.fetchPersons()
        .subscribeOn(scheduler)
        .subscribe(mutablePersons::postValue)
  }
}
```

# 置き換えする時の問題
大きなアプリで、置き換えていくときApiのクラスの実装を変えてしまうと既存のすべての実装を置き換える必要がありそうです :thinking: 
<img src="https://qiita-image-store.s3.amazonaws.com/0/27388/0797dc70-e03e-e9f2-ad2e-dbf70a2b1503.png" width="400" />

# kotlinx-coroutines-rx2

そこで、[kotlinx-coroutines-rx2](https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-rx2)を使うと、以下のように変換できます。

<img src="https://qiita-image-store.s3.amazonaws.com/0/27388/b9308edc-8874-96fc-b060-b91e6f29d446.png" width="400" />

```kotlin
import kotlinx.coroutines.experimental.rx2.await

class ViewModel(
    private val api: Api = Api(),
    private val coroutineContext: CoroutineContext = CommonPool
) {
  private val mutablePersons = MutableLiveData<List<Person>>()
  val persons: LiveData<List<Person>> = mutablePersons

  fun onCreate() {
    launch(coroutineContext) { // Kotlin Coroutines起動
      // ** ここでRxJavaのSingleを中断関数に変換 **
      val persons = api.fetchPersons().await()
      mutablePersons.postValue(persons)
    }
  }
}
```

kotlinx-coroutines-rx2を使うことで、逆にsuspendメソッドからRxJavaのsingleにも置き換えができます。

# サンプルアプリを用意
置き換えていけるのか検証の意味も込めて、以下のようにサンプルアプリを用意しました。
https://github.com/takahirom/rxjava-2-kotlion-coroutines

| | Source code | test | image | 
|---|---|---|---|
| RxJavaのアプリ | [Source code](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/1_rxjava/src/main/java/com/github/takahirom/rxjava2kotlincoroutines)  | [Test](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/1_rxjava/src/test/java/com/github/takahirom/rxjava2kotlincoroutines) | ![image](https://user-images.githubusercontent.com/1386930/44306464-618aae80-a3ca-11e8-96f2-078051e9a199.png) |
| CoroutinesからRxJavaのSingleを使う例 | [Source code](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/2_coroutines_rxjava/src/main/java/com/github/takahirom/rxjava2kotlincoroutines) | [Test](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/2_coroutines_rxjava/src/test/java/com/github/takahirom/rxjava2kotlincoroutines) | ![image](https://user-images.githubusercontent.com/1386930/44306466-64859f00-a3ca-11e8-9472-e70cd5ec5079.png) |
| RxJavaからCoroutinesのsuspend functionを使う例 | [Source code](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/3_rxjava_coroutines/src/main/java/com/github/takahirom/rxjava2kotlincoroutines) | [Test](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/3_rxjava_coroutines/src/test/java/com/github/takahirom/rxjava2kotlincoroutines) | ![image](https://user-images.githubusercontent.com/1386930/44306467-67808f80-a3ca-11e8-9035-0fbc6599b867.png) |
| Coroutinesを使う例 | [Source code](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/4_coroutines/src/main/java/com/github/takahirom/rxjava2kotlincoroutines) | [Test](https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/4_coroutines/src/test/java/com/github/takahirom/rxjava2kotlincoroutines) | ![image](https://user-images.githubusercontent.com/1386930/44306469-6e0f0700-a3ca-11e8-8a2a-75c420bdae4e.png) |

# テスト、エラーハンドリング、ライフサイクルを考慮した例
実際に書いていくときはいくつかのポイントを意識する必要があるのでそれを含んだサンプルも用意しています。
https://github.com/takahirom/rxjava-2-kotlion-coroutines/tree/master/5_coroutines_advanced/src

ポイント1: CoroutineContextを渡すことで、テスト時に切り替えられるようにして、テスト可能にする。
ポイント2: Jobを親として渡して、それをViewModel#onClearedでcancelすることで、使われない無駄な処理をさせないことができる。(これはなくても問題なく動作し、オプションと言えそう)
ポイント3: RxJavaのonErrorで処理をしていたのであれば、同じように、launch内でクラッシュしたときにハンドリングしてあげる必要がある。
ポイント4: suspend functionを使っているのでawaitを書く必要がない。Kotlinではこれが

他にもViewModelが外からいじれないようにフィールドを分けたり、SingleLiveDataを使って、画面回転時に知らせないようにするなど、工夫をしていますが、Coroutinesとは関係ないので、省略します。

```kotlin
class PersonsViewModel(
    private val api: Api = Api(),
    coroutineContext: CoroutineContext = CommonPool // ポイント1
) : ViewModel() {
  private val compositeJob = Job() // ポイント2
  private val mutablePersons = MutableLiveData<List<Person>>()
  val persons: LiveData<List<Person>> = mutablePersons

  private val singleError: SingleLiveData<Exception> = SingleLiveData()
  val errors: LiveData<Exception> = singleError

  init {
    launch(coroutineContext, parent = compositeJob) {  // ポイント2
      try { // ポイント3
        val persons = api.fetchPersons() // ポイント4
        mutablePersons.postValue(persons)
      } catch (e: Exception) {
        singleError.postValue(e)
      }
    }
  }

  override fun onCleared() {
    compositeJob.cancel()
  }
}
```

# RxJavaのSchedulerとスレッドを共有したい
RxJavaのioスレッドは1スレッドで動きますが、Coroutineでも独自のスレッドを作っていた場合、2スレッドになってしまい、マルチスレッドを考慮したプログラミングが必要になってしまいます。
そこでRxJavaのスケジューラーを`asCoroutineDispatcher()`を使うことでCoroutineからSchedulerのスレッドが利用できます。

AppCoroutineDispatchers

```kotlin
class AppCoroutineDispatchers(
    val io: CoroutineContext = Schedulers.io().asCoroutineDispatcher(),
    val computation: CoroutineContext = Schedulers.computation().asCoroutineDispatcher(),
    val ui: CoroutineContext = UI
)
```

DaggerのProvider

```kotlin
  @Provides @Singleton 
  fun provideDefaultCoroutineContext(): AppCoroutineDispatchers {
    return AppCoroutineDispatchers()
  }
```

Coroutineを使うクラス

```kotlin
class Hoge @Inject constructor(
    private val coroutineDispatchers: AppCoroutineDispatchers
...
launch(coroutineDispatchers.io) { 
```


# まとめ
まだまだ知見やベストプラクティスを探っている状態ですが、見つけていきたいです。
こうしたらいいなどのコメントをTwitterなどでもいいのでお待ちしています :bow: