Go with the Kotlin Flow 🌊

Kotlin Flow is yet another implementation of Reactive Stream specification made on top of coroutines for Kotlin.

In this talk, we will look at Kotlin Flow, why it was introduced, some basics, show what makes it special, how can you migrate and most important testing

Agenda
- Why Flow and it's history
- Basics of Kotlin Flow
- How it compares with RxJava and LiveData
- Why and how to migrate from RxJava 1 or 2
- Working with RxJava ❤ Flow in single codebase
- Magic of multi-platform
- Tests
- Conclusion

You’ll walk away with a clear idea of what Kotlin Flow is and how it compares with other reactive implementation particularly RxJava and how to evaluate and how you can gradually migrate from Rx if you want to

Transcript

1. Akshay Chordiya Go with the Flow Hey everyone, it’s super exciting to see people from India some familiar faces and even some people joining from other parts of the world. I hope everyone is coming up with creative ideas while at home and staying safe. Welcome and thank you for joining. I’m Akshay Chordiya, I’m based in Berlin, Germany; originally from Pune, India. I work as Android Developer at Clue which is a female health company. I’m also Google Developer Expert for Android. Today we are going to talk about something which many people are excited about i.e Kotlin Flow [moving to next slide]
2. Akshay Chordiya Go with the Flow The goal of this talk is to tell you why Kotlin Flow was designed, some basics, show what makes Flow special, how can you migrate to Flow, how to architect Flow on Android and most important testing
3. Coroutines Let’s rewind ⏪ a little bit Let’s rewind a little bit and look at what Kotlin gave us - coroutines. They simplify asynchronous programming in general and especially on Android
4. Synchronous code fun showUser() { val user = db.loadUser() // Blocks the thread render(user) } Here is an example of synchronous function called `showUser` which internally loads the user from the database.
5. Synchronous code fun showUser() { val user = db.loadUser() // Blocks the thread render(user) } loadUser When this function calls `loadUser` it blocks the current thread and waits till user is loaded from the database, this blocking is bad because at this moment the whole UI is frozen -> making the user unhappy
6. Synchronous code fun showUser() { val user = db.loadUser() // Blocks the thread render(user) } loadUser render And then once the data is loaded from database the `render` function is called which renders the user data on the UI. There are multiple ways to tackle this. Let’s look at how we can convert this function to use coroutine
7. With Coroutines fun showUser() { val user = db.loadUser() render(user) } Let’s start by taking the original synchronous function
8. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread render(user) } First we need to mark the `showUser` function with suspend keyword, of course we will need to mark the `loadUser` as suspend too since we also want want it to run on background thread. You can see that just by marking it as suspend the code will run in background thread
9. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread render(user) } But there is a problem the `render` function needs to be called from UI thread since its touching the view hierarchy.
10. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } } We can use the `withContext` function which allows to switch the thread inside the suspending function and change the thread to main and call `render` function, and that’s it!
11. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } } showUser This is how it will work, we start by calling the `showUser` function, since it’s marked as suspend it will switch to a background thread or the thread specified while launching the coroutine
12. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } } showUser loadUser and then the user will be loaded from the database on a different thread while the main thread continues to go ahead without waiting for anything i.e not freezing or blocking the user experience
13. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } } showUser loadUser render And then it will switch back to main thread because of `withContext` and then call the `render` function on the Main / UI thread
14. A suspend function can only be called from another suspend function One key thing to remember is a suspend function can only be called from another suspend function
15. A suspend function can only be called from another suspend function or a coroutine scope Or from a coroutine scope,
16. Coroutine scope A coroutine scope represents the scope of the coroutines, each coroutine needs a scope to run in and when the scope is destroyed all the running jobs inside the scope are also stopped or cancelled
17. Coroutine scope ~ Lifecycle scope This scope is generally associated to the lifecycle scope for simplicity. For instance, on Android each activity has a lifecycle. Let’s look at example to help understand this better
18. fun main() { lifecycleScope.launch { showUser() } } * available in lifecycle-ktx Jetpack library For example, in order to call our `showUser` suspend function from a normal function we would need a coroutine scope. On Android these scopes are already implemented for us like `lifecycleScope` which is part of `lifecycle-ktx` library from Jetpack components. it is provided to use coroutine function in Activity or Fragment. It’s a coroutine scope which is tied to the lifecycle of activity or fragment
19. fun main() { lifecycleScope.launch { showUser() } } CREATED DESTROYED This provides the magic of automatically clearing or cancelling all the jobs running in the specified scope when the activity or fragment is destroyed
20. fun main() { lifecycleScope.launch { showUser() } } The launch function is used to start the coroutine, it will immediately start the coroutine That means coroutines are hot in nature, this also one of the reason why coroutines are not replacement for reactive streams
21. “A coroutine scope is kinda a subscription management system associated to lifecycle ♻”
22. Structured Concurrency This concept is called as Structured Concurrency and this is what makes Kotlin coroutines so powerful and also helps developers prevent leaking the job / subscription
23. Why Kotlin Flow? Let’s come to the important question, why Kotlin Flow?
24. Why Kotlin Flow? Kotlin Coroutine With growing usage of Kotlin and the love for Kotlin coroutines, people expressed a lot of interest in having a pure Kotlin implementation of RxJava to leverage all the amazing features which Kotlin provides
25. RxKotlin or from scratch? But the confusion was creating some sort of magic on top of RxJava to leverage Kotlin or to build a whole new library from scratch. In case if you don’t know RxKotlin does exist but it cannot really leverage all the Kotlin features.
26. Me “At the end it’s important that any new library raises the bar compared to existing ones and is not just a replica with slight improvements” To me it’s important that any new library raises the bar compared to any existing libraries out there and is not just a replica with slight improvements. And Kotlin Flow does raise the bar
27. So What’s special So let’s look at what’s special in Kotlin Flow to raise the bar
28. What’s special ✨ Null safety in streams ✨ Interoperability between other reactive streams and coroutines ✨ Supports Kotlin multiplatform ✨ No special handling for back pressure ✨ Fewer and simple operators ✨ Perks of structured concurrency • Flow provides null safety in the stream that means you cannot send null values in the stream, this is not something new; RxJava 2 has it too • This is the key thing, Flow provides easy APIs to convert Flow to any other reactive stream like RxJava Flowable and the reverse. Also you can combine Flow and coroutines, we will see this later • You can actually use Kotlin coroutines and Flow on any of the supported Kotlin platforms like JVM, JS and native + use it the common code or shared code for Kotlin multiplatform projects • There is no special handling for back pressure, there are literally no operators for handling back pressure cause this is magically handled by the suspending nature of coroutines • There are fewer operators which directly correlates to ease of understanding, this is because of 2 reasons • Kotlin’s extension function magic • Suspending nature of coroutines allow having same operator for synchronous / asynchronous code which cuts down the number of operators •
29. What is Flow?
30. What is Flow? Based upon reactive stream specification Similar to a list / collection It has operators like map, switchMap, etc Built on top of coroutines It’s ❄ in nature
31. Basics Let’s look at some of the basics of Flow
32. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Essentially there are 2 components i.e emitter and collector where emitter emits new values and collector receives the value. Here is an example of an emitter.
33. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Which returns Flow of string very similar to how you would return a list of string
34. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } There are various ways of creating a flow using something called as flow builders (they are functions provided by Kotlin coroutines library). It internally creates the flow for us, we will come to it later.
35. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } ❄ ❄ ❄ ❄ ❄ Since Flow is cold, nothing inside the flow builder block will be called until there is a subscriber or a collector observing the flow
36. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } Let’s look at the collector, here we have a main function which gets the flow inside the `runBlocking` function. For people who haven’t used coroutines; `runBlocking` function is a blocking coroutine scope i.e it blocks the current thread until everything is completed. We need the coroutine scope since the `collect` function is a suspend function, so is `emit`. We will see ahead why they are suspend functions
37. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() Let’s understand how things work
38. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() Initially the main function calls the stream function
39. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() Which immediately returns the flow
40. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() Which immediately returns the flow
41. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() collect Then we call the collect function on the flow, which essentially means we are subscribed to the emitter
42. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() collect emit That the emitter starts emitting the value sequentially
43. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() collect emit And the collector receives them and prints in this case
44. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() collect emit collect emit ⚽ And then the same loop happens emitting the next value
45. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() collect emit collect emit ⚽ ⚽
46. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() collect emit collect emit ⚽ ⚽ .. This goes on
47. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } } main() stream() collect emit collect emit ⚽ ⚽ collect .. Till the emitter has no values that’s when the execution moves out of collect and is complete
48. Let’s dive deeper Let’s dive a little deeper and see how this is working internally
49. public interface Flow<out T> { public suspend fun collect(collector: FlowCollector<T>) } Collector As I mentioned before Flow is based upon 2 components a collector and emitter, essentially they are interfaces. The following represents the interface for collector which exposes the `collect` function
50. public interface Flow<out T> { public suspend fun collect(collector: FlowCollector<T>) } Collector public interface FlowCollector<in T> { public suspend fun emit(value: T) } Emitter And the emitter interface which allows emitting values
51. public interface Flow<out T> { public suspend fun collect(collector: FlowCollector<T>) } Collector public interface FlowCollector<in T> { public suspend fun emit(value: T) } Emitter The key thing to note over here is both the functions are suspend which allows the magic of interoperability with coroutines and also allows using the structured concurrency provided by coroutines. The funny thing is the whole internal mechanism of flow is essentially `emit` and `collect` function calling each other.
52. Flow ❤ Coroutines Flow loves coroutine also because it’s built on top of it
53. fun loadFromDatabase(): String { // Heavy operation of fetching from DB return "Very heavy " } Let’s say you have a function which loads some data from database, since this operation takes time to perform you’d ideally want to perform it in the background thread by marking it as suspend
54. suspend fun loadFromDatabase(): String { // Heavy operation of fetching from DB return "Very heavy " } And the funny thing is
55. suspend fun loadFromDatabase(): String { // Heavy operation of fetching from DB return "Very heavy " } fun stream() = flow { val value = loadFromDatabase() emit(value) } you can just directly call any coroutine function inside the `flow` block
56. suspend fun loadFromDatabase(): String { // Heavy operation of fetching from DB return "Very heavy " } fun stream() = flow { val value = loadFromDatabase() emit(value) } public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> That’s because the `flow` block accepts a suspending lambda and also because the both the collector and emitter are suspend functions which is just great!
57. Principles of Flow Let’s take a look at core ideology of Flow and how they make our life easier
58. fun main() = runBlocking { flow { emit("Context") } .flowOn(Dispatchers.IO) } With Flow you can define or switch the thread by using the operator called as `flowOn` which takes in a dispatcher which in this case is IO.
59. fun main() = runBlocking { flow { emit("Context") // IO thread } .flowOn(Dispatchers.IO) } ‐ This operator sets the specified this thread only upstream i.e anything above the `flowOn` operator will run on IO thread.
60. fun main() = runBlocking { flow { emit("Context") // IO thread } .flowOn(Dispatchers.IO) .map { "$it Preservation" } } The map operator below the `flowOn` will run on some different thread and not IO thread
61. Context Preservation This concept of preventing context from leaking is called as “Context Preservation” where the context is encapsulated and it’s never allowed to go downstream ⬇ . There is only one way to change the context of a flow
62. Observable.just("") .map{ /* IO thread */ } .subscribeOn(Schedulers.io()) .map{ /* IO thread */ } .subscribeOn(Schedulers.computation()) .map{ /* IO thread */ } RxJava example If you compare this to RxJava where it’s super confusing especially for beginners to understand which thread their code is going to run on. In the example above you can see that only the first `subscribeOn` is applied and other one is ignored
63. Observable.just("") .map{ /* IO thread */ } .subscribeOn(Schedulers.io()) .map{ /* IO thread */ } .subscribeOn(Schedulers.computation()) .map{ /* IO thread */ } .observeOn(Schedulers.computation()) .map { /* */ } RxJava example Combined with another operator to change thread i.e `observeOn` with different behaviour, it’s a facepalm moment
64. Exception Transparency This is the 2nd principle or ideology of Flow. Flow implementations never catch or handle exceptions that occur in downstream ⬇ flows, they catch only for upstream ⬆ ; this is again due to context preservation. There is only 1 operator to catch error which is the `catch` operators. Let’s look at an example
65. flow { val value = loadFromDatabase() emit(value) } Let’s take the previous example of a flow which tries to get the value from the database and emit. As you can imagine anything can happen while loading from database which might throw some sort of exception
66. flow { val value = loadFromDatabase() emit(value) } .catch { emit("Fallback value") } Very similar to try/catch block, flow has a `catch` operator which will catch the any exceptions thrown above
67. flow { try { val value = loadFromDatabase() emit(value) } catch (e: Exception) { emit("Fallback value") } } You can even use the classic try/catch block, it will behave exactly similar to the catch operator
68. flow { try { val value = loadFromDatabase() emit(value) } catch (e: Exception) { when (e) { is CancellationException -> throw e else -> emit("Fallback value") } } } I’d recommend to use the catch operator but there are can be use-cases. During those time just make sure to ignore the `CancellationException` because you don’t want to emit an error when your flow or coroutine was cancelled when the activity was closed or something. This is pretty how the catch operator works internally
69. flow { try { val value = loadFromDatabase() emit(value) } catch (e: IOException) { emit("Fallback value") } } Or you can specifically handle the exception you expect to see. For example, in case of database operation you’d specifically want to exception related to IO operation.
70. flow { val value = loadFromDatabase() emit(value) } .catch { emit("Fallback value") } .collect { } The value emitted from the catch is received in the collect, that means there is only a single place to handle all types of cases either success or failure which makes it super transparent to handle all the cases
71. sealed class State { object Success: State() object Failure: State() } fun main() = runBlocking { flow<State> { emit(State.Success) } .catch { emit(State.Failure) } .collect { state -> when (state) { State.Success -> println("Success") State.Failure -> println("Failure") } } } Ex Example I’d recommend to toss in sealed classes so you always cover all your cases
72. “- Use suspend for one-shot operations like insert operation or a network operation - Use Flow<T> to get stream of data like getting updates on data changes in the database.” It might be confusing to understand when to use suspend and when to use `Flow`.
73. “- Use suspend for one-shot operations like insert operation or a network operation - Use Flow<T> to get stream of data like getting updates on data changes in the database.” The general rule of thumb is use `suspend` for one-shot operations like insert into database operation or a network operation.
74. “- Use suspend for one-shot operations like insert operation or a network operation - Use Flow<T> to get stream of data like getting updates on data changes in the database.” And use Flow when there is stream of data involved, for example get data from the database and continuously listening for updates.
75. Legacy ➡ Flow This is one of the coolest thing and I love it a lot, Flow provides an amazing API to covert any legacy callback based API to Flow
76. fun updateOnDocumentChange(document: Document) { val listener = Document.ChangeListener { updateUI(it) } document.addChangeListener(listener) } Let’s say you have a function which has a listener which is triggered each time the document is updated.
77. fun updateOnDocumentChange(document: Document) { val listener = Document.ChangeListener { updateUI(it) } document.addChangeListener(listener) } removeChangeListener(listener) You need to make sure to remove this listener when the component is destroyed for example when activity is destroyed
78. fun Document.awaitChange(): Flow<Document> = callbackFlow { } Flow provides an API called as `callbackFlow`
79. fun Document.awaitChange(): Flow<Document> = callbackFlow { val listener = Document.ChangeListener { } addChangeListener(listener) } Where you can write the legacy listener aka callback
80. fun Document.awaitChange(): Flow<Document> = callbackFlow { val listener = Document.ChangeListener { offer(it) } addChangeListener(listener) } And offer which basically means emit this value each time the document is changed. But you also need to remove the listener at some point right?
81. fun Document.awaitChange(): Flow<Document> = callbackFlow { val listener = Document.ChangeListener { offer(it) } addChangeListener(listener) awaitClose { removeChangeListener(listener) } } You can use the function `awaitClose` which is called when the flow or coroutine is cancelled, where we can safely remove the listener
82. /** * Listen for changes in the [Document] and notify the * receivers about changes. */ fun Document.awaitChange(): Flow<Document> = callbackFlow { val listener = Document.ChangeListener { offer(it) } addChangeListener(listener) awaitClose { removeChangeListener(listener) } } Finally with code comments it looks something like this which a lot of people forget to write
83. Rx ➡ Flow Let’s look at how you can migrate from Rx to Flow, the goal here is to not tell you to go next moment and migrate your codebase to Flow [I mean that would be great for your own projects].
84. Reactive Stream Specification Flow Rx Flow and Rx follow the same reactive stream specification so in theory it’s very easy to convert data streams from Rx to Flow and the reverse
85. https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-rx2 fun flow(): Flow<String> { return Observable.just("a") .asFlow() } There are two ways to convert, 1st is like refactor your Rx implementation to use flow but that’s time confusing and sometimes you just want to switch types so you can gradually move to flow. In that cases you can use the `asFlow` extension function to quick switch the type from `Flowable` / `Observable` to `Flow`
86. fun observable(): Observable<String> { return flow { emit("a") }.asObservable() } https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-rx2 fun flow(): Flow<String> { return Observable.just("a") .asFlow() } Same can be done to convert flow to observable or flowable when you need it
87. Rx Coroutines Single / Completable / Maybe suspend function Observable Flow Subject Channel Schedulers Dispatchers Disposables Scopes subscribe collect Terminology between Rx and Coroutine world Here’s a quick comparison of terminology between Rx and coroutine which I stole from Clue’s documentation. In my defence I created this table at Clue Use this chart as a reference when you are new to coroutines world or working on migrating from Rx to coroutines or reverse. Now that see, we pretty much covered all the things mentioned except channel.
88. Fun part Let’s look at something really fun
89. public fun <T: Any> ObservableSource<T>.asFlow(): Flow<T> = callbackFlow { val observer = object : Observer<T> { override fun onComplete() { close() } override fun onSubscribe(d: Disposable) { .. } override fun onNext(t: T) { sendBlocking(t) } override fun onError(e: Throwable) { close(e) } } subscribe(observer) awaitClose { /* Dispose */ } } The `asFlow` function actually uses `callbackFlow` to convert observable to flow which is soooo cool
90. Testing Now to everyone’s favourite topic - testing Testing flow or coroutine is pretty easy
91. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } } Let’s take a simple flow of string
92. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } @Test fun testFlow() = runBlocking { } } And write the test function and use the `runBlocking` coroutine scope because we want our test to wait till the coroutine is completed
93. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } @Test fun testFlow() = runBlocking { val expected = listOf("Bob the builder ") val actualList = getFlow().take(1).toList() // Assert the list assertEquals(expected, actualList) } } Like I said Flow is similar to a list or a collection
94. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } @Test fun testFlow() = runBlocking { val expected = listOf("Bob the builder ") val actualList = getFlow().take(1).toList() // Assert the list assertEquals(expected, actualList) } } So you can use `take(1)` operator to pick the 1st value from the stream
95. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } @Test fun testFlow() = runBlocking { val expected = listOf("Bob the builder ") val actualList = getFlow().take(1).toList() // Assert the list assertEquals(expected, actualList) } } and then convert the flow to list so you can easily assert it
96. /** * Asserts only the [expected] items by just taking that many from the stream */ suspend fun <T> Flow<T>.assertItems(vararg expected: T) { assertEquals(expected.toList(), this.take(expected.size).toList()) } Or use this extension function which I again stole from our Clue code
97. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } @Test fun testFlow() = runBlocking { val expected = "Bob the builder " // Assert the list getFlow().assertItems(expected) } } With the extension function it would be something like this, much more cleaner
98. Conclusion
99. Conclusion • Revisited coroutines • Why Flow? And what makes it special? • Basics of Flow • Principles of Flow • Rx or Legacy code -> Flow • Testing • Pro-tips along the way Here’s a quick summary what you learned today - We revisited coroutines - We saw why Kotlin Flow came into the picture and what makes it special - We saw basics of Flow i.e the emitter and collector and how they work internally - Principles of Flow i.e context preservation and exception transparency and how they make our lives easy - then how to convert Rx or Legacy code to Flow - How you can test flows - And some pro-tips along the way
100. Me “Kotlin Flow raises the bar and introduces easy reactive streaming, that said it’s work in progress and lots of exciting things are coming sooooooon ” Final thoughts To conclude Kotlin Flow raises the bar and makes reactive programming easy, gives us great Kotlin features. That said it’s still work in progress and lots of exciting things are coming soooon.
101. Thank You