DroidKaigi 2020 アプリでの学び【Kotlin Coroutines Flow 編】

## この記事なに？
DroidKaigi 2020 アプリに commit したので、そこで得た学びや感想を書いた記事です。
いろいろ書きたいことがあったので、分割して書きます。
僕のモチベが続く限り続きます。
ちなみに初めての OSS 活動でした 😎

僕が取り組んだ Pull Request はこの2つになります。

- [Thumbs\-up function 👍 by mkeeda · Pull Request \#626 · DroidKaigi/conference\-app\-2020](https://github.com/DroidKaigi/conference-app-2020/pull/626)
  - セッションへのいいね機能を実装しました
  - medium の crap 機能のように連打できる機能です
- [\[iOS\] Fix error if build directory is already exists by mkeeda · Pull Request \#805 · DroidKaigi/conference\-app\-2020](https://github.com/DroidKaigi/conference-app-2020/pull/805)
  - DroidKaigi は Kotlin Multiplatform Project なので、iOS 用の Kotlin モジュールをビルドしています
  - Xcode から iOS アプリをビルドしたとき、Kotlin のモジュールもビルドするのですが、build ディレクトリがすでにあったときエラーになってたので修正しました


## Kotlin Coroutines Flow 
DroidKaigi 2020 アプリの大きな特徴の1つとして、 Kotlin Coroutines Flow (以下 Flow) の採用があります。
Flow は RxJava を始めとする Reactive Stream の概念にインスパイアされており、Kotlin の標準機能だけでリアクティブプログラミングができるようになります。
[Asynchronous Flow \- Kotlin Programming Language](https://kotlinlang.org/docs/reference/coroutines/flow.html)

僕は業務で RxJava を利用しているため、Flow には以前から興味がありました。
今回取り組んだ PR の中で Flow をがっつり触ったので、RxJava と比較した感想を書いていきます。

### Flow ってなに？
Kotlin は非同期処理の方法としてコルーチンという機能を提供しています。
コルーチンは `suspend` キーワードがついた中断可能な関数を定義し、処理の中断や待ち合わせを書きやすくする仕組みです。
suspend 関数は中断可能であること以外、通常の関数と同じなので、一度の関数呼び出しで一度しか値を得られません。
よって、複数の値を非同期で取得するためにFlow が誕生しました。

Flow は RxJava と同じで publish / subscribe パターンで設計されています。 
`Flow` という Cold なストリーム (RxJava でいう `Observable`) に非同期で値を流せます。
また、値を使う側は `collect` を使って値を購読してよしなに使います。
AndroidX の LiveData とも似てますね。

DroidKaigi 2020 アプリでは、ViewModel - Repository 間や Repository - DB 間の接続で Flow が用いられています。
データ層からデータを購読し、Flow が提供するオペレータでデータを加工して ViewModel へ伝えます。
[conference\-app\-2020/arch1\.png at master · DroidKaigi/conference\-app\-2020](https://github.com/DroidKaigi/conference-app-2020/blob/master/arts/arch1.png)

具体的な例だと、いいね機能では Firestore に格納されている各セッションのいいね数を監視する部分に使用しています。
[conference\-app\-2020/FirestoreImpl\.kt · DroidKaigi/conference\-app\-2020](https://github.com/DroidKaigi/conference-app-2020/blob/67e699a958c3c0c4b7e2f02484540fe64a6ddc6a/data/firestore/src/main/java/io/github/droidkaigi/confsched2020/data/firestore/internal/FirestoreImpl.kt#L71)

## Flow の良いところ: 購読の管理が RxJava より楽
### RxJava の dispose 忘れ問題
業務で RxJava を使っていて大変だなと思うところは、`Disposable` の管理です。
Android のライフサイクルに合わせた dispose 処理をしないと、必要なときにストリームから値が流れてこなかったり、不要になったストリームが動き続けたりします。
そして、dispose の有無はコンパイラではチェックされないので、めちゃ忘れやすいです。

`CompositeDisposable` という複数の dispose 処理をまとめて呼ぶ仕組みもありますが、結局一度は`CompositeDisposable.dispose()` をどこかで呼ばないとダメなので同じです。

```kotlin
fun retrieveDataFromDB(): Observable<Data> {
    // DB を監視してデータを取ってくる処理
}

// disposable.dispose() を呼ばないと、ずっと subscribe が動き続ける
val disposable = retrieveDataFromDB()
    subscribe { data ->
        println(data)
    }
```

### Flow の Structured concurrency によるストリームの購読管理
なんのこっちゃって感じかもしれませんが、要は **CoroutineScope でストリームの購読が動作する範囲を制御できる** 点が良いなと思いました。

コルーチンの一部である Flow は、Structured concurrency という概念のもとに設計されています。
Structured concurrency とは雑に説明すると、並行に実行されるコードブロック (コルーチン) を管理する仕組みのことです。
そして、コルーチンの動作する範囲が `CoroutineScope` です。

`CoroutineScope.cancel()` を呼ぶと Scope 内のコルーチンをすべてキャンセルできます。
つまり、`CoroutineScope.cancel()` が `CompositeDispossable.dispose()` と同じような役割を果たします。

じゃあ RxJava と一緒じゃんと思いきや、Android Architecture Components には `ViewModelScope` や `LifecycleScope` など便利な Scope があります。
これらの Scope を使うと、ViewModel や Activity/Fragment のライフサイクルに合わせてコルーチンをいい感じにキャンセルしてくれます。

すべてのコルーチンは `CoroutineScope` の中で動作させる必要があるため、Flow の購読を始める際は必ず Scope を意識します。
さらに、Android Architecture Components の提供する Scope を使えばライフサイクルを意識しなくて良くなり、購読の管理が楽になります。

と、Dropbox のエンジニアも言ってました😊
[Store grand re\-opening: loading Android data with coroutines \| Dropbox Tech Blog](https://blogs.dropbox.com/tech/2020/01/store-grand-re-opening-loading-android-data-with-coroutines/)


## Flow の良いところ: RxJava 並にオペレータが潤沢
RxJava の強みは豊富なオペレータだと感じています。
オペレータを組み合わせて、様々なストリームの加工ができます。
便利ですね。
(ただ、オペレータが豊富すぎてなかなか使いこなせず、非同期処理のためだけに RxJava を使うのはオーバースペックだの、学習コストが高いだのと言われがちなんですがね。。。)
[ReactiveX \- Operators](http://reactivex.io/documentation/operators.html)

Flow もオペレータはめちゃあります。
`map`, `flatMap` (Flow では `flatMapLatest`) , `filter` など基本的なものから、`debounce`, `distinctUntilChanged` など UI を作るのに便利なオペレータまであります。
[Flow \- kotlinx\-coroutines\-core](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/)

LiveData はオペレータが少なすぎて、 RxJava と同じことをしようとするとオペレータを自作する必要がありました。
Flow は RxJava でやっていたストリーム変換が大体同じようにできました。
さらに、`transform` オペレータを使えば複雑なストリーム変換処理の自作も柔軟にできそうです。
(僕は DroidKaigi アプリでは使ってませんが)

```kotlin
fun Flow<Int>.skipOddAndDuplicateEven(): Flow<Int> = transform { value ->
    if (value % 2 == 0) { // Emit only even values, but twice
        emit(value)
        emit(value)
    } // Do nothing if odd
}
```
[transform \- kotlinx\-coroutines\-core](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform.html)

余談ですが、Flow と RxJava で決定的に違うところは、 null をストリームに流せるというところです。
RxJava では null を流すとエラーが起きます。
どうやら、Reactive Stream で null は流してはいけないと決まっているらしいです。
Flow には、`filterNotNull` など nullable を unwrap するようなオペレータもあるので、null 安全な Kotlin らしく書けて大変便利です。

## Flow の良いところ: Kotlin Multiplatform Project への可能性
Kotlin Multiplatform Project (Kotlin MPP) は Kotlin だけで iOS や Windows などのネイティブコード、Javascript まで書いちゃおう！という技術です。
DroidKaigi も Kotlin MPP を採用しており、iOS アプリで Kotlin のモジュールを使っています。

しかし、Kotlin で書かれたコードを他のプラットフォームで使おうとすると、JVM の資産は使えません。
つまり、Java で書かれた RxJava を使っていては Kotlin MPP 化できません。

残念ながら、コルーチンや Flow もまだ Kotlin MPP には完全に対応しておらず、iOS の Framework や他のネイティブコードへはコンパイルできません。
そのため、DroidKaigi では Swift との接点で Flow は使っていませんでした (たぶん)。
callback 形式のインターフェースで API のレスポンスを返却し、RxSwift で Observable(Single) 化する方法が取られていました。

```SessionDataProvider.swift
final class SessionDataProvider {
    func fetchSessions() -> Single<[Session]> {
        return Single.create { observer -> Disposable in
            ApiComponentKt.generateDroidKaigiApi().getSessions(callback: { response in
                let model = ResponseToModelMapperKt.toModel(__: response)
                observer(.success(model.sessions))
            }, onError: { error in
                observer(.error(KotlinError(localizedDescription: error.description())))
            })
            return Disposables.create()
        }
    }
```
[conference\-app\-2020/SessionDataProvider\.swift · DroidKaigi/conference\-app\-2020](https://github.com/DroidKaigi/conference-app-2020/blob/689c5faf00b283618c6246ecbd2a8f3d063cf2df/ios-base/DroidKaigi%202020/Session/Models/SessionDataProvider.swift#L7)

また、JetBrains が作成した Kotlin Conf のアプリでは、`CFlow` という iOS 用に Flow をラップしたクラスが利用されています。
`CFlow`は callback 形式よりもう少し進んでおり、Swift でも擬似的に Flow の値を購読できます。

```FlowUtils.kt
fun <T> Flow<T>.wrap(): CFlow<T> = CFlow(this)

class CFlow<T>(private val origin: Flow<T>) : Flow<T> by origin {
    fun watch(block: (T) -> Unit): Closeable {
        val job = Job(/*ConferenceService.coroutineContext[Job]*/)

        onEach {
            block(it)
        }.launchIn(CoroutineScope(dispatcher() + job))

        return object : Closeable {
            override fun close() {
                job.cancel()
            }
        }
    }
}
```
[kotlinconf\-app/FlowUtils\.kt at master · JetBrains/kotlinconf\-app](https://github.com/JetBrains/kotlinconf-app/blob/master/common/src/mobileMain/kotlin/org/jetbrains/kotlinconf/FlowUtils.kt)

使い方↓

```HomeController.swift
Conference.votes.watch { votes in
    let count = Float(votes?.count ?? 0)
    let progress = min(1.0, count / Float(Conference.votesCountRequired()))

    let hidden = progress != 1.0
    ...
}
```
[kotlinconf\-app/HomeController\.swift at master · JetBrains/kotlinconf\-app](https://github.com/JetBrains/kotlinconf-app/blob/master/iosApp/iosApp/Controllers/HomeController.swift#L50)

Flow はまだ完全に Kotlin MPP で利用できるとは言えませんが、RxJava の代用にはなりそうです。
何より、サードパーティライブラリではなく 言語の標準機能でリアクティブプログラミングが実現できるという安心感があります。

## Flow の悩み？どころ: CoroutineScope はどこから貰う良いのか
Flow を動作させるのに `CoroutineScope` が必要なことはメリットだと言ったばかりですが、めんどくさいときもありました。

Activity/Fragment や ViewModel などプレゼンテーション層に近い部分では、`ViewModelScope` や `LifecycleScope` が簡単に取得できます。
ですが、Repository より下のデータ層でこれらの Scope を使うベストプラクティスは何なのかよくわかりませんでした。

単純に必要なクラスに DI するのが良いんでしょうか？
それとも `CoroutineScope` 新たに定義するのがいいんでしょうか？
このあたりはまだコルーチンのわかっていない部分だなあ。

## まとめ
RxJava と比較しながら Flow について思ったことをつらつら書きました。
Flow はまだ Experimental な API も多く、まだまだ開発中な機能です。
ですが、想像していたよりずっと RxJava と似ていたので Rx を知っていればそんなに難しくは無いと思います。

まだ分かっていないこととしては、`Observer` に相当する `FlowCollector` や、`Subject` 相当の `Channel` の使い方です。
単純に `Channel` = `Flow` (Observable) + `FlowCollector` (Observer) ではない気がしているので、この辺を探求したいですね💪