5分でわかるKotlin Coroutines Flow

sys1yagi

Shibuya.apk 2019/08/01

© 2019 Ubie, Inc. 1

© 2019 Ubie, Inc.

About Me

22

● 八木俊広

● @sys1yagi

● Software engineer at

医療機関向け 業務効率化

AI問診

患者向け 病気推測

Dr.Ubie

© 2019 Ubie, Inc. 3

5分でわかる Kotlin Coroutines Flow

3

● Kotlin Coroutines Flowとは？

● ホットストリームとコールドストリーム

● ホットストリームとリソースのリーク

● 最初のFlow

● Flowの仕組み

● Flowのオペレータ

● Flowを使う(イベントストリーム)

● まとめ

© 2019 Ubie, Inc. 44

Kotlin Coroutines Flowとは？

© 2019 Ubie, Inc.

Kotlin Coroutines Flowとは?

55

Kotlin Coroutines 1.2.0-alpha-2からfeature previewで登場した、コールドストリーム。

© 2019 Ubie, Inc.

Kotlin Coroutinesが提供する機能たち

66

ワンショット戻り値なし launch{}, Job

ワンショット戻り値あり async{}, Deferred

ホットストリーム(複数の値) Channel

© 2019 Ubie, Inc.

Kotlin Coroutinesが提供する機能たち

77

ワンショット戻り値なし launch{}, Job

ワンショット戻り値あり async{}, Deferred

ホットストリーム(複数の値) Channel

コールドストリーム(複数の値) Flow

© 2019 Ubie, Inc. 88

● 受信者の有無に関わらず動作を開始する

ホットストリーム

コールドストリーム

● 受信し始めない限り動作しない

© 2019 Ubie, Inc.

ホットストリームとChannel

99

kotlinx.coroutines.channels.Channelはホットストリーム。

© 2019 Ubie, Inc.1010

ホットストリームは

取り扱いに気をつけないと

結構簡単にリークしてしまう

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1111

fun TextView.textChangeAsChannel(): ReceiveChannel<String?> { val channel = Channel<String?>(Channel.CONFLATED) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } channel.invokeOnClose { removeTextChangedListener(textWatcher) } return channel}

TextViewの値の変化をChannelで受け取る関数

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1212

fun TextView.textChangeAsChannel(): ReceiveChannel<String?> { val channel = Channel<String?>(Channel.CONFLATED) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } channel.invokeOnClose { removeTextChangedListener(textWatcher) } return channel}

TextViewの値の変化をChannelで受け取る関数

最新の値だけバッファする Channel

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1313

fun TextView.textChangeAsChannel(): ReceiveChannel<String?> { val channel = Channel<String?>(Channel.CONFLATED) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } channel.invokeOnClose { removeTextChangedListener(textWatcher) } return channel}

TextViewの値の変化をChannelで受け取る関数

androidx.core:core-ktx:1.2.0-alpha02

にある関数。

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1414

fun TextView.textChangeAsChannel(): ReceiveChannel<String?> { val channel = Channel<String?>(Channel.CONFLATED) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } channel.invokeOnClose { removeTextChangedListener(textWatcher) } return channel}

TextViewの値の変化をChannelで受け取る関数

Channelがクローズする時に実行し、

リスナーを解除する

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1515

lifecycleScope.launch { val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } }}

TextViewの値の変化をChannelで受け取る関数

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1616

lifecycleScope.launch { val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } }}

TextViewの値の変化をChannelで受け取る関数

この時点で動き出す

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1717

lifecycleScope.launch { val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } }}

TextViewの値の変化をChannelで受け取る関数

trueでonTextChangeを使わない場合でもリスナーは動き続ける

© 2019 Ubie, Inc.

ホットストリームとリソースのリーク

1818

lifecycleScope.launch { val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } }}

TextViewの値の変化をChannelで受け取る関数

trueでonTextChangeを使わない場合でもリスナーは動き続ける

リーク！！

© 2019 Ubie, Inc.1919

そこでコールドストリーム

Kotlin Coroutines Flow

© 2019 Ubie, Inc.

Flowの導入

2020

implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.0-RC"

implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.0-RC"

APIは1.3.0-RCでStableになった。一部の関数に experimentalが残っている

© 2019 Ubie, Inc.

最初のFlow

2121

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

© 2019 Ubie, Inc.

最初のFlow

2222

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

Flowを作成する関数

© 2019 Ubie, Inc.

最初のFlow

2323

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

ブロック内は

suspend FlowCollector<T>.() -> Unit

© 2019 Ubie, Inc.

最初のFlow

2424

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

任意のsuspend関数を呼び出せる

© 2019 Ubie, Inc.

最初のFlow

2525

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

値を送信するsuspend関数

© 2019 Ubie, Inc.

最初のFlow

2626

GlobalScope.launch { val ints: Flow<Int> = intStream() ints.collect { println(it) }}

© 2019 Ubie, Inc.

最初のFlow

2727

GlobalScope.launch { val ints: Flow<Int> = intStream() ints.collect { println(it) }}

この時点で動作はしない

© 2019 Ubie, Inc.

最初のFlow

2828

GlobalScope.launch { val ints: Flow<Int> = intStream() ints.collect { println(it) }}

受信を開始するsuspend関数

© 2019 Ubie, Inc.

最初のFlow

2929

GlobalScope.launch { val ints: Flow<Int> = intStream() ints.collect { println(it) }}

0123456789

© 2019 Ubie, Inc.

Flowの仕組み

3030

GlobalScope.launch { val ints: Flow<Int> = intStream() ints.collect { println(it) }}

public suspend inline fun <T> Flow<T>.collect( crossinline action: suspend (value: T) -> Unit): Unit = collect(object : FlowCollector<T> { override suspend fun emit(value: T) = action(value) })

© 2019 Ubie, Inc.

Flowの仕組み

3131

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

© 2019 Ubie, Inc.

Flowの仕組み

3232

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

public fun <T> flow( @BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> { return object : Flow<T> { override suspend fun collect(collector: FlowCollector<T>) { SafeCollector(collector, coroutineContext).block() } }}

© 2019 Ubie, Inc.

Flowの仕組み

3333

suspendFlow<T>.collect()

suspendFlow#collect()

SafeCollector( collector, coroutineContext).block()

flow ( block: suspend FlowCollector<T>.() -> Unit)

fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) }}

受信開始

© 2019 Ubie, Inc.

Flowの仕組み

3434

● Flow<T>.collect関数をトリガーにして、値を発行するsuspend関数

を実行する

● Flow<T>.collect関数はsuspend関数なので必ずコルーチンスコー

プに属する。そのため閉じ忘れがない

● 受信者毎に独立して動作する

© 2019 Ubie, Inc.

Flowのオペレータ

3535

val ints: Flow<Int> = intStream()val processed = ints .filter { it % 2 == 0 } .map { it * 2 } .drop(1) .take(3)

GlobalScope.launch { processed.collect { println("$it") }}

© 2019 Ubie, Inc.

Flowのオペレータ

3636

val ints: Flow<Int> = intStream()val processed = ints .filter { it % 2 == 0 } .map { it * 2 } .drop(1) .take(3)

GlobalScope.launch { processed.collect { println("$it") }}

値を加工するオペレータ、suspend関数じゃないのでどこでも呼び出せる。関数ブロック内はsuspend関数

© 2019 Ubie, Inc.

Flowのオペレータ

3737

val ints: Flow<Int> = intStream()val processed = ints .filter { it % 2 == 0 } .map { it * 2 } .drop(1) .take(3)

GlobalScope.launch { processed.collect { println("$it") }}

48

12

© 2019 Ubie, Inc.

Flowの末端オペレータ

3838

GlobalScope.launch { val result = processed.first()}

最初の1件を取り出す

© 2019 Ubie, Inc.

Flowの末端オペレータ

3939

GlobalScope.launch { val result = processed.first()}

collect(): Unitsingle(): TsingleOrNULL(): T?toList(): List<T>count(): Intfold(): Treduce(): T...色々ある

© 2019 Ubie, Inc.

Flowを使ってイベントストリームを作る

4040

public fun <T> channelFlow( @BuilderInference block: suspend ProducerScope<T>.() -> Unit): Flow<T> = ChannelFlowBuilder(block)

値の送信にChannelを利用するFlowを作る関数

© 2019 Ubie, Inc.

Flowを使ってイベントストリームを作る

4141

fun TextView.textChangeAsFlow(): Flow<String?> = channelFlow<String?> { this:ProducerScope<T>

channel.offer(text.toString()) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } awaitClose { removeTextChangedListener(textWatcher) } }.conflate()

© 2019 Ubie, Inc.

Flowを使ってイベントストリームを作る

4242

fun TextView.textChangeAsFlow(): Flow<String?> = channelFlow<String?> { this:ProducerScope<T>

channel.offer(text.toString()) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } awaitClose { removeTextChangedListener(textWatcher) } }.conflate()

プロパティにSendChannel

をもっている

© 2019 Ubie, Inc.

Flowを使ってイベントストリームを作る

4343

fun TextView.textChangeAsFlow(): Flow<String?> = channelFlow<String?> { this:ProducerScope<T>

channel.offer(text.toString()) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } awaitClose { removeTextChangedListener(textWatcher) } }.conflate()

値を送信

© 2019 Ubie, Inc.

Flowを使ってイベントストリームを作る

4444

fun TextView.textChangeAsFlow(): Flow<String?> = channelFlow<String?> { this:ProducerScope<T>

channel.offer(text.toString()) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } awaitClose { removeTextChangedListener(textWatcher) } }.conflate()

ブロックを抜けるとChannelが終了する

ので、受信者に閉じられるまで待つ

© 2019 Ubie, Inc.

Flowを使ってイベントストリームを作る

4545

fun TextView.textChangeAsFlow(): Flow<String?> = channelFlow<String?> { this:ProducerScope<T>

channel.offer(text.toString()) val textWatcher = addTextChangedListener { channel.offer(it?.toString()) } awaitClose { removeTextChangedListener(textWatcher) } }.conflate() 常に最新の値を流す

© 2019 Ubie, Inc.

Flowを使ってイベントストリームを作る

4646

lifecycleScope.launch { val onTextChange = edit.textChangeAsFlow() if (!readOnly) { onTextChange.collect { button.isEnabled = !it.isNullOrEmpty() } }}

ここで初めて動き出す

© 2019 Ubie, Inc.

まとめ

4747

● Kotlin CoroutinesにコールドストリームがなかったのでFlowが登場

● Channelはホットストリーム

● ホットストリームは結構簡単にリークする

● Flowは受信を開始しない限り動作しない

● Flowは実行する時に必ずコルーチンスコープに属するので閉じ忘れがない

● Flowにはオペレータがたくさんある

● ChannelはFlowの中で生き続ける

● FlowはだいたいRxJava

● Androidにおいては、RxJavaでストリームを扱っていたものをリプレースする時に使えそう

© 2019 Ubie, Inc.

細かいところはテクブC96本で...!

4848

https://techbooster.booth.pm/items/1485567