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

Flowing in the Deep - Event Streams in Kotlin

Flowing in the Deep - Event Streams in Kotlin

The Kotlin team introduced a new type called Flow which looks similar to RxJava’s Observable or Flowable. Have you ever wondered what’s the difference between Flow and RxJava? How does Flow work under the hood? How is it connected to Coroutines? How do you write your own operator? What about Kotlin Coroutine Channels? Backpressure? Coroutine Scopes?

Join this session to get an introduction into Flow followed by a deep dive into the some core aspects of Flow to ultimately feel confident using Flow inside your android app.

Hannes Dorfmann

July 02, 2019
Tweet

More Decks by Hannes Dorfmann

Other Decks in Technology

Transcript

  1. class MyActivity : Activity() { override fun onCreate(b: Bundle) {

    thread { val x = doHttpRequest() ??? } } }
  2. fun doHttpRequest( callback: (X) -> Unit ){ // A long

    running operation ... callback(X) }
  3. fun doHttpRequest( callback: (X) -> Unit ){ thread { //

    A long running operation ... callback(X) } }
  4. class MyActivity : Activity() { override fun onCreate(b: Bundle) {

    doHttpRequest { result -> // do something with the result } } }
  5. fun doHttpRequest( callback: (X) -> Unit ) { // A

    long running operation ... callback(X) }
  6. launch { val a = doHttpRequest() val b = doHttpRequest()

    val c = doHttpRequest() val d = doHttpRequest() val e = doHttpRequest() val f = doHttpRequest() val g = doHttpRequest() }
  7. fun doHttpRequest( callback: (X) -> Unit ){ // A long

    running operation ... callback(X) }
  8. fun doHttpRequest( observer: (X) -> Unit ){ // A long

    running operation ... observer(X) }
  9. fun doHttpRequest( observer: (X) -> Unit ){ // A long

    running operation ... observer(X) }
  10. fun downloadVideo( observer: (progress) -> Unit ) { // A

    long running operation ... observer(0.1) ... observer(0.2) ... observer(1.0) }
  11. suspend fun downloadVideo() : ??? { // A long running

    operation ... ??? } With Coroutines?
  12. interface Downloader<T> { fun download( observer: (T) -> Unit )

    } interface DownloadObserver<T> { fun emit( value: T ) }
  13. interface Downloader<T> { fun download( o: DownloadObserver<T> ) } interface

    DownloadObserver<T> { fun emit( value: T ) } interface Downloader<T> { fun download( o: DownloadObserver<T> ) } interface DownloadObserver<T> { fun emit( value: T ) }
  14. interface Downloader<T> { fun download( o: DownloadObserver<T> ) } interface

    DownloadObserver<T> { fun emit( value: T ) } interface Downloader<T> { fun collect( o: DownloadObserver<T> ) } interface DownloadObserver<T> { fun emit( value: T ) }
  15. interface Downloader<T> { fun download( o: DownloadObserver<T> ) } interface

    DownloadObserver<T> { fun emit( value: T ) } interface Downloader<T> { fun collect( c: Collector<T> ) } interface Collector<T> { fun emit( value: T ) }
  16. interface Downloader<T> { fun download( o: DownloadObserver<T> ) } interface

    DownloadObserver<T> { fun emit( value: T ) } interface Flow<T> { fun collect( c: Collector<T> ) } interface Collector<T> { fun emit( value: T ) }
  17. interface Downloader<T> { fun download( o: DownloadObserver<T> ) } interface

    DownloadObserver<T> { fun emit( value: T ) } interface Flow<T> { fun collect( c: FlowCollector<T> ) } interface FlowCollector<T> { fun emit( value: T ) }
  18. interface Flow<T> { fun collect( c: FlowCollector<T> ) } interface

    FlowCollector<T> { fun emit( value: T ) } Flow
  19. interface Flow<out T> { suspend fun collect( c: FlowCollector<T> )

    } interface FlowCollector<in T> { suspend fun emit( value: T ) } Flow
  20. fun <T> flow(block: suspend FlowCollector<T>.() -> Unit): Flow<T> fun downloadVideo():

    Flow<Double> = flow { // this: FlowCollector<Double> ... ??? }
  21. fun <T> flow(block: suspend FlowCollector<T>.() -> Unit): Flow<T> fun downloadVideo():

    Flow<Double> = flow { // this: FlowCollector<Double> ... emit(0.1) }
  22. fun <T> flow(block: suspend FlowCollector<T>.() -> Unit): Flow<T> fun downloadVideo():

    Flow<Double> = flow { // this is FlowCollector<Double> ... emit(0.1) ... emit(0.2) ... emit(1.0) }
  23. class MyViewModel : ViewModel() { init { viewModelScope.launch { flow

    { for (i in 0..5) { delay(100) emit(i) } } } } }
  24. class MyViewModel : ViewModel() { init { viewModelScope.launch { flow

    { for (i in 0..5) { delay(100) emit(i) } } } } } Launch new coroutine
  25. class MyViewModel : ViewModel() { init { viewModelScope.launch { flow

    { for (i in 0..5) { delay(100) emit(i) } } } } } Launch new coroutine 2.1.0-beta01
  26. class MyViewModel : ViewModel() { init { viewModelScope.launch { flow

    { for (i in 0..5) { delay(100) emit(i) } } } } } Suspending block
 
 executed in coroutine
  27. launch { flow { for (i in 0..5) { delay(100)

    emit(i) } }.collect { value -> println(value) } }
  28. launch { flow { for (i in 0..5) { delay(100)

    emit(i) } }.collect { value -> println(value) } } Observes flow
  29. launch { flow { for (i in 0..5) { delay(100)

    emit(i) } }.collect { value -> println(value) } } notify observer about value
  30. launch { flow { for (i in 0..5) { delay(100)

    emit(i) } }.collect { value -> println(value) } }
  31. try { flow<Double> { throw Exception("Something went wrong") }.collect {

    println(it) } } catch (e: Exception) { // handle error }
  32. flow<Double> { emit(0.1) throw Exception("Something went wrong") }.collect { println(it)

    } sealed class DownloadStatus { data class Progress(val value: Double) : DownloadStatus() object Success : DownloadStatus() data class Error(val message: String) : DownloadStatus() }
  33. flow<DownloadStatus> { emit(DownloadStatus.Progress(0.1)) try { throw IOException(“Something went wrong”)) }

    catch (e: IOException) { emit(DownloadStatus.Error(e.message)) } }.collect { println(it) } sealed class DownloadStatus { data class Progress(val value: Double) : DownloadStatus() object Success : DownloadStatus() data class Error(val message: String) : DownloadStatus() }
  34. launch { flow { // do stuff }.collect { value

    -> println(value) } } Flow gets instantiated
  35. launch { flow { // do stuff }.collect { value

    -> println(value) } } collect gets called
  36. launch { flow { // do stuff }.collect { value

    -> println(value) } } coroutine suspends here
  37. launch { flow { // do stuff }.collect { value

    -> println(value) } } executed collect suspends here
  38. launch { flow { // do stuff }.collect { value

    -> println(value) } } called for emission collect suspends here
  39. launch { flow { // do stuff }.collect { value

    -> println(value) } } executed collect suspends here
  40. launch { flow { // do stuff }.collect { value

    -> println(value) } } called for emission collect suspends here
  41. launch { flow { // do stuff }.collect { value

    -> println(value) } } collect suspends here
  42. launch { flow { // do stuff }.collect { value

    -> println(value) } handleCompletion() }
  43. flow { // do stuff }.onCompletion { handleCompletion() }.collect {

    value -> println(value) } new in 1.3.0-M 2
  44. flow { for (i in 0..5) { delay(100) emit(i) }

    }.map { value -> value * 2 }.collect { value -> println(value) }
  45. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector } } fun <T> flow(block: suspend FlowCollector<in T>.() -> Unit): Flow<T>
  46. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> } } }
  47. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> } } } Flow<T>.collect
  48. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> } } }
  49. flow { for (i in 0..5) { delay(100) emit(i) }

    }.map { value -> value * 2 }.collect { value -> println(value) }
  50. flow { for (i in 0..5) { delay(100) emit(i) }

    }.map { value -> value * 2 }.collect { value -> println(value) } fun <T, R> Flow<T>.map(…)
  51. flow { for (i in 0..5) { delay(100) emit(i) }

    }.map { value -> value * 2 }.collect { value -> println(value) }
  52. flow { for (i in 0..5) { delay(100) emit(i) }

    }.map { value -> value * 2 }.collect { value -> println(value) }
  53. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> } } }
  54. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> } } }
  55. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> val r = transformer(value) } } }
  56. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> val r = transformer(value) emit(r) } } }
  57. fun <T, R> Flow<T>.map(transformer: suspend (value: T) -> R): Flow<R>

    { return flow { // this: FlowCollector collect { value -> val r = transformer(value) emit(r) } } } FlowCollector.emit(value)
  58. flow { for (i in 0..5) { delay(100) emit(i) }

    }.map { value -> value * 2 }.collect { value println(value) }
  59. flow { for (i in 0..5) { delay(100) emit(i) }

    }.switchMap { value -> flow { emit(“first $value") delay(500) emit(“second $value") } }.collect { value ->
 println(value) }
  60. flow { for (i in 0..5) { delay(100) emit(i) }

    }.switchMap { value -> flow { emit(“first $value") delay(500) emit(“second $value") } }.collect { value ->
 println(value) } Console output first 0 first 1 first 2 first 3 first 4 first 5
  61. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> } } }
  62. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> } } }
  63. flow { for (i in 0..5) { delay(100) emit(i) }

    }.switchMap { value flow { emit(“first $value") delay(500) emit(“second $value") } }.collect { value ->
 println(value) }
  64. flow { for (i in 0..5) { delay(100) emit(i) }

    }.switchMap { value -> flow { emit(“first $value") delay(500) emit(“second $value") } }.collect { value 
 println(value) }
  65. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> } } }
  66. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> val inner : Flow<R> = mapper(outerValue) } } }
  67. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> val inner : Flow<R> = mapper(outerValue) inner.collect { value -> } } } }
  68. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> val inner : Flow<R> = mapper(outerValue) launch { inner.collect { value -> } } } } }
  69. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> val inner : Flow<R> = mapper(outerValue) launch { inner.collect { value -> emit(value) } } } } }
  70. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { // this: FlowCollector collect { outerValue -> val inner : Flow<R> = mapper(outerValue) launch { inner.collect { value -> emit(value) } } } } } FlowCollector.emit(value)
  71. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { collect { outerValue -> val inner : Flow<R> = mapper(outerValue) launch { inner.collect { value -> emit(value) } } } } }
  72. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { var currentJob: Job? = null collect { outerValue -> val inner : Flow<R> = mapper(outerValue) currentJob = launch { inner.collect { value -> emit(value) } } } } }
  73. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { var currentJob: Job? = null collect { outerValue -> val inner : Flow<R> = mapper(outerValue) currentJob?.cancelAndJoin() currentJob = launch { inner.collect { value -> emit(value) } } } } }
  74. fun <T, R> Flow<T>.switchMap( mapper: suspend (value: T) -> Flow<R>

    ): Flow<R> { return flow { coroutineScope { var currentJob: Job? = null collect { outerValue -> val inner : Flow<R> = mapper(outerValue) currentJob?.cancelAndJoin() currentJob = launch { inner.collect { value -> emit(value) } } } } } }
  75. interface CoroutineContext interface Job : CoroutineContext { public fun cancel():

    Unit // a lot more… } abstract class CoroutineDispatcher
  76. interface CoroutineContext interface Job : CoroutineContext { public fun cancel():

    Unit // a lot more… } abstract class CoroutineDispatcher Dispatchers.Main Dispatchers.Default Dispatchers.IO Dispatchers.Unconfined newSingleThreadContext()
  77. interface CoroutineContext interface Job : CoroutineContext { public fun cancel():

    Unit // a lot more… } abstract class CoroutineDispatcher Dispatchers.Main Dispatchers.Default Dispatchers.IO Dispatchers.Unconfined newSingleThreadContext() val context: CoroutineContext = Job() + Dispatchers.IO
  78. public interface CoroutineScope { public val coroutineContext: CoroutineContext } public

    inline fun CoroutineScope.cancel() public fun CoroutineScope.launch(...)
  79. public interface CoroutineScope { public val coroutineContext: CoroutineContext } public

    inline fun CoroutineScope.cancel() public fun CoroutineScope.launch(...) val ViewModel.viewModelScope: CoroutineScope val Lifecycle.coroutineScope: LifecycleCoroutineScope
  80. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 }.map { it.toString() }.collect { println(it) } } main thread
  81. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 }.map { it.toString() }.collect { println(it) } } main thread
  82. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 }.map { it.toString() }.collect { println(it) } } main thread
  83. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 }.map { it.toString() }.collect { println(it) } } main thread
  84. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { GlobalScope.launch { // expensive operation emit(progress)

    // expensive operation emit(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } }
  85. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { withContext(Dispatchers.IO) { // expensive operation emit(progress)

    // expensive operation emit(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } }
  86. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { withContext(Dispatchers.IO) { // expensive operation emit(progress)

    // expensive operation emit(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } } IllegalStateException
  87. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { withContext(Dispatchers.IO) { // expensive operation emit(progress)

    // expensive operation emit(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } }
  88. lifecycle.coroutineScope.launchWhenStarted { channelFlow<Int> { withContext(Dispatchers.IO) { // expensive operation emit(progress)

    // expensive operation emit(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } }
  89. lifecycle.coroutineScope.launchWhenStarted { channelFlow<Int> { // this: ProducerScope<Int> withContext(Dispatchers.IO) { //

    expensive operation emit(progress) // expensive operation emit(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } }
  90. lifecycle.coroutineScope.launchWhenStarted { channelFlow<Int> { // this: ProducerScope<Int> withContext(Dispatchers.IO) { //

    expensive operation send(progress) // expensive operation send(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } }
  91. lifecycle.coroutineScope.launchWhenStarted { channelFlow<Int> { // this: ProducerScope<Int> withContext(Dispatchers.IO) { //

    expensive operation send(progress) // expensive operation send(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } } main thread
  92. lifecycle.coroutineScope.launchWhenStarted { channelFlow<Int> { // this: ProducerScope<Int> withContext(Dispatchers.IO) { //

    expensive operation send(progress) // expensive operation send(result) } }.map { it + 1 }.map { it.toString() }.collect { println(it) } } main thread IO thread
  93. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 } .flowOn(Dispatchers.IO) .map { it.toString() }.collect { println(it) } }
  94. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 } .flowOn(Dispatchers.IO) .map { it.toString() }.collect { println(it) } } main thread IO thread
  95. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 } .flowOn(Dispatchers.IO) .map { it.toString() }.collect { println(it) } } main thread IO thread
  96. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 } .flowOn(Dispatchers.IO) .map { it.toString() }.collect { println(it) } } main thread IO thread
  97. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation }.map { it

    + 1 } .flowOn(Dispatchers.IO) .map { it.toString() }.collect { println(it) } }
  98. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation } .flowOn(Dispatchers.IO) .map

    { it + 1 } .flowOn(Dispatchers.Default) .map { it.toString() }.collect { println(it) } }
  99. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation } .flowOn(Dispatchers.IO) .map

    { it + 1 } .flowOn(Dispatchers.Default) .map { it.toString() }.collect { println(it) } } main thread IO thread Default thread
  100. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation } .flowOn(Dispatchers.IO) .map

    { it + 1 } .flowOn(Dispatchers.Default) .map { it.toString() }.collect { println(it) } } main thread IO thread Default thread
  101. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation } .flowOn(Dispatchers.IO) .map

    { it + 1 } .flowOn(Dispatchers.Default) .map { it.toString() }.collect { println(it) } } main thread IO thread Default thread
  102. lifecycle.coroutineScope.launchWhenStarted { flow<Int> { // expensive operation } .flowOn(Dispatchers.IO) .map

    { it + 1 } .flowOn(Dispatchers.Default) .map { it.toString() }.collect { println(it) } } main thread IO thread Default thread
  103. Threading • collect lambda is always called on the initial

    CoroutineContext • flowOn does only influence the upstream
  104. Threading • collect lambda is always called on the initial

    CoroutineContext • flowOn does only influence the upstream • nothing can change the context of downstream emissions
  105. val someFunction: (() -> Result) = ... someFunction.asFlow() val doHttpRequest:

    (suspend () -> Result) = ... doHttpRequest.asFlow()
  106. RxJava & Flow map() filter() combineLatest() take() scan() retry() debounce()

    distinctUntilChanged() zip() takeUntil() reduce()
  107. flow { emit(userId1) emit(userId2) }.someOperator { userId -> val book

    = myService.getFavoriteBook(userId) book } interface MyService { @GET("{user}/books/favorite") suspend fun getFavoriteBook(@Path("user") userId: Int): Book }
  108. flow { emit(userId1) emit(userId2) }.someOperator { userId -> val book

    = myService.getFavoriteBook(userId) book }.collect { book -> println(book) } interface MyService { @GET("{user}/books/favorite") suspend fun getFavoriteBook(@Path("user") userId: Int): Book }
  109. flow { emit(userId1) emit(userId2) }.map { userId -> val book

    = myService.getFavoriteBook(userId) book }.collect { book -> println(book) } interface MyService { @GET("{user}/books/favorite") suspend fun getFavoriteBook(@Path("user") userId: Int): Book }