Upgrade to Pro — share decks privately, control downloads, hide ads and more …

5分でわかるKotlin Coroutines Flow

yagi
August 01, 2019

5分でわかるKotlin Coroutines Flow

- Kotlin CoroutinesにコールドストリームがなかったのでFlowが登場
- Channelはホットストリーム
- ホットストリームは結構簡単にリークする
- Flowは受信を開始しない限り動作しない
- Flowは実行する時に必ずコルーチンスコープに属するので閉じ忘れがない
- Flowにはオペレータがたくさんある
- ChannelはFlowの中で生き続ける
- FlowはだいたいRxJava
- Androidにおいては、RxJavaでストリームを扱っていたものをリプレースする時に使えそう

yagi

August 01, 2019
Tweet

More Decks by yagi

Other Decks in Technology

Transcript

  1. © 2019 Ubie, Inc. About Me 2 2 • 八木俊広

    • @sys1yagi • Software engineer at 医療機関向け 業務効率化 AI問診 患者向け 病気推測 Dr.Ubie
  2. © 2019 Ubie, Inc. 3 5分でわかる Kotlin Coroutines Flow 3

    • Kotlin Coroutines Flowとは? • ホットストリームとコールドストリーム • ホットストリームとリソースのリーク • 最初のFlow • Flowの仕組み • Flowのオペレータ • Flowを使う(イベントストリーム) • まとめ
  3. © 2019 Ubie, Inc. Kotlin Coroutines Flowとは? 5 5 Kotlin

    Coroutines 1.2.0-alpha-2からfeature previewで登場した、コールドストリーム。
  4. © 2019 Ubie, Inc. Kotlin Coroutinesが提供する機能たち 6 6 ワンショット 戻り値なし

    launch{}, Job ワンショット 戻り値あり async{}, Deferred ホットストリーム (複数の値) Channel
  5. © 2019 Ubie, Inc. Kotlin Coroutinesが提供する機能たち 7 7 ワンショット 戻り値なし

    launch{}, Job ワンショット 戻り値あり async{}, Deferred ホットストリーム (複数の値) Channel コールドストリーム (複数の値) Flow
  6. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 1 11 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で受け取る関数
  7. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 2 12 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
  8. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 3 13 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 にある関数。
  9. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 4 14 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がクローズする時に実行し、 リスナーを解除する
  10. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 5 15 lifecycleScope.launch {

    val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } } } TextViewの値の変化をChannelで受け取る関数
  11. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 6 16 lifecycleScope.launch {

    val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } } } TextViewの値の変化をChannelで受け取る関数 この時点で動き出す
  12. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 7 17 lifecycleScope.launch {

    val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } } } TextViewの値の変化をChannelで受け取る関数 trueでonTextChangeを使わない場合でもリスナーは動き続ける
  13. © 2019 Ubie, Inc. ホットストリームとリソースのリーク 1 8 18 lifecycleScope.launch {

    val onTextChange = edit.textChangeAsChannel() if (!readOnly) { onTextChange.consumeEach { button.isEnabled = !it.isNullOrEmpty() } } } TextViewの値の変化をChannelで受け取る関数 trueでonTextChangeを使わない場合でもリスナーは動き続ける リーク!!
  14. © 2019 Ubie, Inc. Flowの導入 2 0 20 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が残っている
  15. © 2019 Ubie, Inc. 最初のFlow 2 1 21 fun intStream():

    Flow<Int> = flow { repeat (10) { delay(10) emit(it) } }
  16. © 2019 Ubie, Inc. 最初のFlow 2 2 22 fun intStream():

    Flow<Int> = flow { repeat (10) { delay(10) emit(it) } } Flowを作成する関数
  17. © 2019 Ubie, Inc. 最初のFlow 2 3 23 fun intStream():

    Flow<Int> = flow { repeat (10) { delay(10) emit(it) } } ブロック内は suspend FlowCollector<T>.() -> Unit
  18. © 2019 Ubie, Inc. 最初のFlow 2 4 24 fun intStream():

    Flow<Int> = flow { repeat (10) { delay(10) emit(it) } } 任意のsuspend関数を呼び出せる
  19. © 2019 Ubie, Inc. 最初のFlow 2 5 25 fun intStream():

    Flow<Int> = flow { repeat (10) { delay(10) emit(it) } } 値を送信するsuspend関数
  20. © 2019 Ubie, Inc. 最初のFlow 2 6 26 GlobalScope.launch {

    val ints: Flow<Int> = intStream() ints.collect { println(it) } }
  21. © 2019 Ubie, Inc. 最初のFlow 2 7 27 GlobalScope.launch {

    val ints: Flow<Int> = intStream() ints.collect { println(it) } } この時点で動作はしない
  22. © 2019 Ubie, Inc. 最初のFlow 2 8 28 GlobalScope.launch {

    val ints: Flow<Int> = intStream() ints.collect { println(it) } } 受信を開始するsuspend関数
  23. © 2019 Ubie, Inc. 最初のFlow 2 9 29 GlobalScope.launch {

    val ints: Flow<Int> = intStream() ints.collect { println(it) } } 0 1 2 3 4 5 6 7 8 9
  24. © 2019 Ubie, Inc. Flowの仕組み 3 0 30 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) })
  25. © 2019 Ubie, Inc. Flowの仕組み 3 1 31 fun intStream():

    Flow<Int> = flow { repeat (10) { delay(10) emit(it) } }
  26. © 2019 Ubie, Inc. Flowの仕組み 3 2 32 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() } } }
  27. © 2019 Ubie, Inc. Flowの仕組み 3 3 33 suspend Flow<T>.collect()

    suspend Flow#collect() SafeCollector( collector, coroutineContext ).block() flow ( block: suspend FlowCollector<T>.() -> Unit ) fun intStream(): Flow<Int> = flow { repeat (10) { delay(10) emit(it) } } 受信開始
  28. © 2019 Ubie, Inc. Flowの仕組み 3 4 34 • Flow<T>.collect関数をトリガーにして、値を発行するsuspend関数

    を実行する • Flow<T>.collect関数はsuspend関数なので必ずコルーチンスコー プに属する。そのため閉じ忘れがない • 受信者毎に独立して動作する
  29. © 2019 Ubie, Inc. Flowのオペレータ 3 5 35 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") } }
  30. © 2019 Ubie, Inc. Flowのオペレータ 3 6 36 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関数
  31. © 2019 Ubie, Inc. Flowのオペレータ 3 7 37 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") } } 4 8 12
  32. © 2019 Ubie, Inc. Flowの末端オペレータ 3 8 38 GlobalScope.launch {

    val result = processed.first() } 最初の1件を取り出す
  33. © 2019 Ubie, Inc. Flowの末端オペレータ 3 9 39 GlobalScope.launch {

    val result = processed.first() } collect(): Unit single(): T singleOrNULL(): T? toList(): List<T> count(): Int fold(): T reduce(): T ...色々ある
  34. © 2019 Ubie, Inc. Flowを使ってイベントストリームを作る 4 0 40 public fun

    <T> channelFlow( @BuilderInference block: suspend ProducerScope<T>.() -> Unit ): Flow<T> = ChannelFlowBuilder(block) 値の送信にChannelを利用するFlowを作る関数
  35. © 2019 Ubie, Inc. Flowを使ってイベントストリームを作る 4 1 41 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()
  36. © 2019 Ubie, Inc. Flowを使ってイベントストリームを作る 4 2 42 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 をもっている
  37. © 2019 Ubie, Inc. Flowを使ってイベントストリームを作る 4 3 43 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() 値を送信
  38. © 2019 Ubie, Inc. Flowを使ってイベントストリームを作る 4 4 44 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が終了する ので、受信者に閉じられるまで待つ
  39. © 2019 Ubie, Inc. Flowを使ってイベントストリームを作る 4 5 45 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() 常に最新の値を流す
  40. © 2019 Ubie, Inc. Flowを使ってイベントストリームを作る 4 6 46 lifecycleScope.launch {

    val onTextChange = edit.textChangeAsFlow() if (!readOnly) { onTextChange.collect { button.isEnabled = !it.isNullOrEmpty() } } } ここで初めて動き出す
  41. © 2019 Ubie, Inc. まとめ 4 7 47 • Kotlin

    CoroutinesにコールドストリームがなかったのでFlowが登場 • Channelはホットストリーム • ホットストリームは結構簡単にリークする • Flowは受信を開始しない限り動作しない • Flowは実行する時に必ずコルーチンスコープに属するので閉じ忘れがない • Flowにはオペレータがたくさんある • ChannelはFlowの中で生き続ける • FlowはだいたいRxJava • Androidにおいては、RxJavaでストリームを扱っていたものをリプレースする時に使えそう