Do you even (Kotlin) Flow? The new API for reactive programming

Reactive programming is here to stay.
Most Android devs nowadays are used to sprinkle RxJava throughout their apps, even if they don't actually have use for reactive streams and just want to simplify thread scheduling. However, it's also true that more and more devs are becoming aware that Rx is an overkill for their use case, and long for better options. Kotlin Flow is one of the new additions to the Kotlin coroutines library, and is meant to bring reactive streams to the coroutine world. But is it as powerful as RxJava? In this talk, I'll explore what are Kotlin Flows, how can they be used, and how can they imbue your app with the power of reactive programming.

Transcript

1. Do you even (Kotlin) Flow? The new API for Reactive Programming Ricardo Costeira @rcosteira79 Photo by Riccardo Chiarini on Unsplash
4. Reactive Programming Coroutines Channels
5. Reactive Programming
6. • Focus on the data itself • React to changes in data • No control over where data comes from ◦ helps us avoid asynchronicity issues
7. fun getUsers(): Observable<User> { return api.getAllUsers() // Maybe<List<User>> .filter { it.isNotEmpty() } .flattenAsObservable { it } // Observable<User> } fun getUserDetails(name: String): Maybe<DetailedUser> { return api.getUserDetails(name) } .flatMapMaybe { getUserDetails(it.name) } .toList() // Single<List<DetailedUser>> .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe( { handleDetailedUsers(it) }, { handleErrors(it) } ) getUsers()
8. .flatMapMaybe { getUserDetails(it.name) } .toList() // Single<List<DetailedUser>> .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe( { handleDetailedUsers(it) }, { handleErrors(it) } ) compositeDisposable.add(getUsers() ) fun getUsers(): Observable<User> { return api.getAllUsers() // Maybe<List<User>> .filter { it.isNotEmpty() } .flattenAsObservable { it } // Observable<User> } fun getUserDetails(name: String): Maybe<DetailedUser> { return api.getUserDetails(name) }
9. .flatMapMaybe { getUserDetails(it.name) } .toList() // Single<List<DetailedUser>> .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe( { handleDetailedUsers(it) }, { handleErrors(it) } ) getUsers() .addTo(compositeDisposable) // Extension function fun getUsers(): Observable<User> { return api.getAllUsers() // Maybe<List<User>> .filter { it.isNotEmpty() } .flattenAsObservable { it } // Observable<User> } fun getUserDetails(name: String): Maybe<DetailedUser> { return api.getUserDetails(name) }
10. Coroutines
11. • Lightweight threads ◦ A typical thread on Android: 1 to 2mb ◦ A typical coroutine on Android: a couple of bytes • Provide mechanisms to jump between different thread types • Use suspending functions to pause/continue execution (without blocking the thread) • Perform async computations with sequential looking code
12. suspend fun getUsers(): List<User> { return api.getAllUsers() } suspend fun getUserDetails(name: String): DetailedUser { return api.getUserDetails(name) } val job = scope.launch { // scope tied to the main thread withContext(Dispatchers.IO) { val users = getUsers() // List<User> .map { async { getUserDetails(it.name) } } // Yay parallelism! .map { it.await() } // Wait for the async calls to finish } // back in the main thread if (users.isNotEmpty()) { handleDetailedUsers(users) } else { handleErrors(NoUsersError()) } }
13. Channels
14. • Synchronization primitives • Communicate between sender and receiver through suspending operations • Hot stream - close it, or leak it ◦ channel.close()
15. val job = GlobalScope.launch { print("Everything litty, ") produce<String> { send("I love when it's hot") } } job.invokeOnCompletion { print("Turned up the city, ") } Thread.sleep(1000) println("I broke off the notch") Everything litty, I broke off the notch
16. val channel = Channel<String>() val job = GlobalScope.launch { print("Everything litty, ") channel.send("I love when it's hot") channel.close() } CoroutineScope(Dispatchers.IO).launch { println(channel.receive()) } job.invokeOnCompletion { print("Turned up the city, ") } Thread.sleep(1000) println("I broke off the notch") Everything litty, I love when it's hot Turned up the city, I broke off the notch
18. Kotlin Flow
19. • Asynchronously computed cold reactive streams • Can be built through specific builders ◦ flow { }, flowOf(), .asFlow(), etc • Intermediate operators ◦ map, filter, zip, flatMapMerge, etc • Terminal operators: ◦ collect, toList, toSet, launchIn, etc
20. fun notFoo(): Flow<Int> = flow { for (i in 1..3) { delay(1000) println("Emitting $i") emit(i) } } scope.launch { val flow = notFoo() println("Calling collect!") flow.collect { value -> println("Collecting $value") } println("Again!") flow.collect { value -> println("Collecting $value") } } Emitting 1 Collecting 1 Emitting 2 Collecting 2 Emitting 3 Collecting 3 Emitting 1 Collecting 1 Emitting 2 Collecting 2 Emitting 3 Collecting 3 Calling collect! Again!
21. scope.launch { notFoo() .flowOn(Dispatchers.Default) .collect { value -> println("Collecting $value") } } fun notFoo(): Flow<Int> = flow { for (i in 1..3) { delay(1000) println("Emitting $i") emit(i) } }
22. launch(Dispatchers.Main) { notFoo() } .filter { it % 2 == 0 } // Main // Main
23. launch(Dispatchers.Main) { notFoo() } .filter { it % 2 == 0 } .flowOn(Dispatchers.IO) // IO // IO
24. launch(Dispatchers.Main) { notFoo() } .filter { it % 2 == 0 } .flowOn(Dispatchers.IO) // IO // IO .map { it * 2 } // Main
25. launch(Dispatchers.Main) { notFoo() } .filter { it % 2 == 0 } .flowOn(Dispatchers.IO) // IO // IO .map { it * 2 } // Default .flowOn(Dispatchers.Default)
26. launch(Dispatchers.Main) { notFoo() } .filter { it % 2 == 0 } .flowOn(Dispatchers.IO) // IO // IO .map { it * 2 } // Default .flowOn(Dispatchers.Default) .collect { value -> println("Collecting $value") } // Main
27. Buffer 1 2 3 Collected in 1220 ms fun main() = runBlocking<Unit> { val time = measureTimeMillis { notFoo().collect { value -> delay(300) // “processing” it for 300 ms println(value) } } println("Collected in $time ms") } fun notFoo(): Flow<Int> = flow { for (i in 1..3) { delay(100) // pretend we are waiting 100 ms emit(i) // emit next value } }
28. Buffer fun notFoo(): Flow<Int> = flow { for (i in 1..3) { delay(100) // pretend we are waiting 100 ms emit(i) // emit next value } } 1 2 3 Collected in 1071 ms fun main() = runBlocking<Unit> { val time = measureTimeMillis { notFoo() .buffer() // buffer emissions, don't wait .collect { value -> delay(300) // “processing” it for 300 ms println(value) } } println("Collected in $time ms") }
29. Conflate 1 3 Collected in 766 ms fun notFoo(): Flow<Int> = flow { for (i in 1..3) { delay(100) // pretend we are waiting 100 ms emit(i) // emit next value } } fun main() = runBlocking<Unit> { val time = measureTimeMillis { notFoo() .conflate() // conflate emissions, don't process each one .collect { value -> delay(300) // “processing” it for 300 ms println(value) } } println("Collected in $time ms") }
30. <operator>Latest Collecting 1 Collecting 2 Collecting 3 Done 3 Collected in 741 ms fun notFoo(): Flow<Int> = flow { for (i in 1..3) { delay(100) // pretend we are waiting 100 ms emit(i) // emit next value } } fun main() = runBlocking<Unit> { val time = measureTimeMillis { notFoo() .collectLatest { value -> // cancel & restart on latest println("Collecting $value") delay(300) // pretend we are processing it for 300 ms println("Done $value") } } println("Collected in $time ms") }
31. Size limiting operators fun numbers(): Flow<Int> = flow { try { emit(1) emit(2) println("This line will not execute") emit(3) } catch (e: Exception) { println(e) } finally { println("Finally in numbers") } } fun main() = runBlocking<Unit> { numbers() .take(2) // take only the first two .collect { value -> println(value) } } 1 2 kotlinx.coroutines.flow.internal. AbortFlowException: Flow was aborted, no more elements needed Finally in numbers
32. Size limiting operators fun numbers(): Flow<Int> = flow { emit(1) emit(2) emit(3) } fun main() = runBlocking<Unit> { numbers() .take(2) // take only the first two .catch { e -> println(e)} .onCompletion { println("Finally in numbers") } .collect { value -> println(value) } } 1 2 Finally in numbers
33. Transform suspend fun performRequest(request: Int): String { delay(1000) // imitate long-running asynchronous work return "response $request" } fun main() = runBlocking<Unit> { (1..3).asFlow() // a flow of requests .transform { request -> emit("Making request $request") emit(performRequest(request)) } .collect { response -> println(response) } } Response 1 Making request 2 Making request 1 Response 2 Making request 3 Response 3
34. Flow on Android • User interactions • Database updates • Wherever you need the Observer pattern
35. Flow on Android button.setOnClickListener { // do things }
36. Flow on Android fun View.clicks(): Flow<Unit> = callbackFlow { // ProducerScope<Unit> val listener = View.OnClickListener { offer(Unit) } setOnClickListener(listener) awaitClose { setOnClickListener(null) } } // Usage (RxBinding, anyone?) button.clicks() .map { /* apply operators */ } .collect { /* collect events */ }
37. Flow on Android @Dao interface UsersDao { @Query("SELECT * from Users") fun getAllUsers(): } List<User>
38. Flow on Android @Dao interface UsersDao { @Query("SELECT * from Users") fun getAllUsers(): } Flow<List<User>>
39. Flow on Android sealed class ViewEvent { object UpdateUsers : ViewEvent() object LoadMoreUsers : ViewEvent() }
40. Flow on Android sealed class ViewEvent { object UpdateUsers : ViewEvent() object LoadMoreUsers : ViewEvent() } // SomeActivity fun viewEvents(): Flow<ViewEvent> { val flows = listOf( updateButton.clicks().map { ViewEvent.UpdateUsers }, loadMoreButton.clicks().map { ViewEvent.LoadMoreUsers }, ) return flows.asFlow().flattenMerge(flows.size) }
41. Flow on Android sealed class ViewEvent { object UpdateUsers : ViewEvent() object LoadMoreUsers : ViewEvent() } // SomeActivity fun viewEvents(): Flow<ViewEvent> { val flows = listOf( updateButton.clicks().map { ViewEvent.UpdateUsers }, loadMoreButton.clicks().map { ViewEvent.LoadMoreUsers }, ) return flows.asFlow().flattenMerge(flows.size) } scope.launch { viewEvents() .collect { viewEvent -> viewModel.processEvent(viewEvent) } }
42. Flow on Android sealed class ViewEvent { object UpdateUsers : ViewEvent() object LoadMoreUsers : ViewEvent() } // SomeActivity fun viewEvents(): Flow<ViewEvent> { val flows = listOf( updateButton.clicks().map { ViewEvent.UpdateUsers }, loadMoreButton.clicks().map { ViewEvent.LoadMoreUsers }, ) return flows.asFlow().flattenMerge(flows.size) } viewEvents() .onEach { viewEvent -> viewModel.processEvent(viewEvent) } .launchIn(scope)
43. Flow on Android data class ViewState( val isLoading: Boolean = false, val users: List<User> = emptyList(), val possibleFailure: Failure = Failure.NoFailure )
44. Flow on Android data class ViewState( val isLoading: Boolean = false, val users: List<User> = emptyList(), val possibleFailure: Failure = Failure.NoFailure ) // SomeViewModel private val _state = ConflatedBroadcastChannel<ViewState>() val stateFlow = _state.asFlow()
45. Flow on Android data class ViewState( val isLoading: Boolean = false, val users: List<User> = emptyList(), val possibleFailure: Failure = Failure.NoFailure ) // SomeViewModel private val _state = ConflatedBroadcastChannel<ViewState>() val stateFlow = _state.asFlow() val newState: ViewState = /** Compute new state, according to view event */
46. Flow on Android data class ViewState( val isLoading: Boolean = false, val users: List<User> = emptyList(), val possibleFailure: Failure = Failure.NoFailure ) // SomeViewModel private val _state = ConflatedBroadcastChannel<ViewState>() val stateFlow = _state.asFlow() val newState: ViewState = /** Compute new state, according to view event */ _state.offer(newState)
47. Flow on Android data class ViewState( val isLoading: Boolean = false, val users: List<User> = emptyList(), val possibleFailure: Failure = Failure.NoFailure ) // SomeViewModel private val _state = ConflatedBroadcastChannel<ViewState>() val stateFlow = _state.asFlow() val newState: ViewState = /** Compute new state, according to view event */ _state.offer(newState) // Collect in Activity stateFlow .onEach { state -> updateUI(state) } .launchIn(scope)
48. Resources • Kotlin Flow official documentation - https://kotlinlang.org/docs/reference/coroutines/flow.html • Roman Elizarov’s Medium posts about Flow - https://medium.com/@elizarov/cold-flows-hot-channels-d74769805f9 • Roman Elizarov’s talk at Kotlinconf 2019 - https://www.youtube.com/watch?v=E4F0YU8Jd5g&t=4895s • David Karnok’s Kotlin Flow Extensions - https://github.com/akarnokd/kotlin-flow-extensions • Lessons Learnt using Coroutines Flow in the Android Dev Summit 2019 App - https://medium.com/androiddevelopers/lessons-learnt-using-coroutines-flow-4a6b285c0d06 • Android Dev Summit 2019 App repo - https://github.com/google/iosched/tree/adssched • “Coroutine + Flow = MVI” by Etienne Caron at Droidcon NYC 2019 - https://www.droidcon.com/media-detail?video=362742098 • “Flowing Things, not so strange in the MVI world” by Garima Jain at Droidcon NYC 2019 - https://www.droidcon.com/media-detail?video=362742238
50. Do you even (Kotlin) Flow? The new API for Reactive Programming Ricardo Costeira @rcosteira79 Photo by Riccardo Chiarini on Unsplash Thank you!