この記事なに？

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

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

Thumbs-up function 👍 by mkeeda · Pull Request #626 · DroidKaigi/conference-app-2020
- セッションへのいいね機能を実装しました
- medium の crap 機能のように連打できる機能です
[iOS] Fix error if build directory is already exists by mkeeda · Pull Request #805 · DroidKaigi/conference-app-2020
- 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

僕は業務で 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

具体的な例だと、いいね機能では Firestore に格納されている各セッションのいいね数を監視する部分に使用しています。
conference-app-2020/FirestoreImpl.kt · DroidKaigi/conference-app-2020

Flow の良いところ: 購読の管理が RxJava より楽

RxJava の dispose 忘れ問題

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

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

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

Flow の良いところ: RxJava 並にオペレータが潤沢

RxJava の強みは豊富なオペレータだと感じています。
オペレータを組み合わせて、様々なストリームの加工ができます。
便利ですね。
(ただ、オペレータが豊富すぎてなかなか使いこなせず、非同期処理のためだけに RxJava を使うのはオーバースペックだの、学習コストが高いだのと言われがちなんですがね。。。)
ReactiveX - Operators

Flow もオペレータはめちゃあります。
map, flatMap (Flow では flatMapLatest) , filter など基本的なものから、debounce, distinctUntilChanged など UI を作るのに便利なオペレータまであります。
Flow - kotlinx-coroutines-core

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

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

余談ですが、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

また、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

使い方↓

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

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) ではない気がしているので、この辺を探求したいですね💪