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
2. Akshay Chordiya Go with the Flow
3. Coroutines Let’s rewind ⏪ a little bit
4. Synchronous code fun showUser() { val user = db.loadUser() // Blocks the thread render(user) }
5. Synchronous code fun showUser() { val user = db.loadUser() // Blocks the thread render(user) } loadUser
6. Synchronous code fun showUser() { val user = db.loadUser() // Blocks the thread render(user) } loadUser render
7. With Coroutines fun showUser() { val user = db.loadUser() render(user) }
8. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread render(user) }
9. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread render(user) }
10. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } }
11. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } } showUser
12. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } } showUser loadUser
13. With Coroutines suspend fun showUser() { val user = db.loadUser() // Background thread withContext(Dispatchers.Main) { render(user) // UI thread } } showUser loadUser render
14. 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
16. Coroutine scope
17. Coroutine scope ~ Lifecycle scope
18. fun main() { lifecycleScope.launch { showUser() } } * available in lifecycle-ktx Jetpack library
19. fun main() { lifecycleScope.launch { showUser() } } CREATED DESTROYED
20. fun main() { lifecycleScope.launch { showUser() } }
21. “A coroutine scope is kinda a subscription management system associated to lifecycle ♻”
22. Structured Concurrency
23. Why Kotlin Flow?
24. Why Kotlin Flow? Kotlin Coroutine
25. RxKotlin or from scratch?
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”
27. So What’s special
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
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
32. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") }
33. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") }
34. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") }
35. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } ❄ ❄ ❄ ❄ ❄
36. fun stream(): Flow<String> = flow { emit("") // Emits the value upstream ☝ emit("⚽") emit("") } Emitter Collector fun main() = runBlocking { stream().collect { value -> println(value) } }
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()
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()
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()
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()
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
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
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
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 ⚽
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 ⚽ ⚽ ..
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 ..
48. Let’s dive deeper
49. public interface Flow<out T> { public suspend fun collect(collector: FlowCollector<T>) } Collector
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
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
52. Flow ❤ Coroutines
53. fun loadFromDatabase(): String { // Heavy operation of fetching from DB return "Very heavy " }
54. suspend fun loadFromDatabase(): String { // Heavy operation of fetching from DB return "Very heavy " }
55. suspend fun loadFromDatabase(): String { // Heavy operation of fetching from DB return "Very heavy " } fun stream() = flow { val value = loadFromDatabase() emit(value) }
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>
57. Principles of Flow
58. fun main() = runBlocking { flow { emit("Context") } .flowOn(Dispatchers.IO) }
59. fun main() = runBlocking { flow { emit("Context") // IO thread } .flowOn(Dispatchers.IO) } ‐
60. fun main() = runBlocking { flow { emit("Context") // IO thread } .flowOn(Dispatchers.IO) .map { "$it Preservation" } }
61. Context Preservation
62. Observable.just("") .map{ /* IO thread */ } .subscribeOn(Schedulers.io()) .map{ /* IO thread */ } .subscribeOn(Schedulers.computation()) .map{ /* IO thread */ } RxJava example
63. Observable.just("") .map{ /* IO thread */ } .subscribeOn(Schedulers.io()) .map{ /* IO thread */ } .subscribeOn(Schedulers.computation()) .map{ /* IO thread */ } .observeOn(Schedulers.computation()) .map { /* */ } RxJava example
64. Exception Transparency ☢
65. flow { val value = loadFromDatabase() emit(value) }
66. flow { val value = loadFromDatabase() emit(value) } .catch { emit("Fallback value") }
67. flow { try { val value = loadFromDatabase() emit(value) } catch (e: Exception) { emit("Fallback value") } }
68. flow { try { val value = loadFromDatabase() emit(value) } catch (e: Exception) { when (e) { is CancellationException -> throw e else -> emit("Fallback value") } } }
69. flow { try { val value = loadFromDatabase() emit(value) } catch (e: IOException) { emit("Fallback value") } }
70. flow { val value = loadFromDatabase() emit(value) } .catch { emit("Fallback value") } .collect { }
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") } } } x Example
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.”
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.”
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.”
75. Channels
76. Channels are communication primitives
77. Channels are communication primitives Source
78. Source Channels are communication primitives
79. Channels are hot in nature
80. Channels are low level primitives
81. private val searchQueryChannel = ConflatedBroadcastChannel<String>()
82. private val searchQueryChannel = ConflatedBroadcastChannel<String>()
83. private val searchQueryChannel = ConflatedBroadcastChannel<String>()
84. // On text change searchQueryChannel.send("Bob the builder ") private val searchQueryChannel = ConflatedBroadcastChannel<String>()
85. // On text change searchQueryChannel.send("Bob the builder ") private val searchQueryChannel = ConflatedBroadcastChannel<String>() searchQueryChannel .asFlow() /* Get the results for the search query */ .distinctUntilChanged() .collect { // Render the result on UI }
86. StateFlow
87. private val searchQueryFlow = MutableStateFlow<String>("")
88. private val searchQueryFlow = MutableStateFlow<String>("") // On text change searchQueryFlow.value = "Bob the builder "
89. private val searchQueryFlow = MutableStateFlow<String>("") // On text change searchQueryFlow.value = "Bob the builder " searchQueryFlow /* Get the results for the search query */ .collect { // Render the result on UI }
90. private val searchQueryFlow = MutableStateFlow<String>("") // On text change searchQueryFlow.value = "Bob the builder " searchQueryFlow /* Get the results for the search query */ .distinctUntilChanged() .collect { // Render the result on UI }
91. Legacy ➡ Flow
92. Setup migration process
93. Discuss with your team
94. Discuss with your team Setup a meeting
95. Discuss with your team Setup a meeting Getting comfortable
96. Discuss with your team Setup a meeting Getting comfortable Evaluate
97. Discuss with your team Setup a meeting Getting comfortable Evaluate Set the common goal
98. Setup foundation ⛩
99. Setup foundation ⛩ Setup guidelines
100. Setup foundation ⛩ Setup guidelines Setup the necessary base classes
101. Setup foundation ⛩ Setup guidelines Setup the necessary base classes Setup the testing framework
102. Setup foundation ⛩ Setup guidelines Setup the necessary base classes Setup the testing framework Expose common features as Flow
103. Expose common features as Flow interface UserRepository { // Legacy function fun observeUser(): Observable<User> }
104. Expose common features as Flow interface UserRepository { // Legacy function fun observeUser(): Observable<User> // Flow equivalent of [observeUser]. fun getUser(): Flow<User> }
105. Expose common features as Flow interface UserRepository { // Legacy function @Deprecated( message = "Use the flow equivalent -> getUser()", replaceWith = ReplaceWith("getUser()", "<package>") ) fun observeUser(): Observable<User> // Flow equivalent of [observeUser]. fun getUser(): Flow<User> } ☠
106. Setup safeguards ☢
107. Setup safeguards ☢ • Ensuring that the new code is written using Kotlin Coroutines while doing the PR reviews • Ensuring when the legacy code is touched, the developer has tried to convert it coroutines or flow • Monthly or quarterly checking of progress of the gradual migration • And other ideas which will pop in your brilliant minds
108. Interoperability
109. fun updateOnDocumentChange(document: Document) { val listener = Document.ChangeListener { updateUI(it) } document.addChangeListener(listener) }
110. fun updateOnDocumentChange(document: Document) { val listener = Document.ChangeListener { updateUI(it) } document.addChangeListener(listener) } removeChangeListener(listener)
111. fun Document.awaitChange(): Flow<Document> = callbackFlow { }
112. fun Document.awaitChange(): Flow<Document> = callbackFlow { val listener = Document.ChangeListener { } addChangeListener(listener) }
113. fun Document.awaitChange(): Flow<Document> = callbackFlow { val listener = Document.ChangeListener { offer(it) } addChangeListener(listener) }
114. fun Document.awaitChange(): Flow<Document> = callbackFlow { val listener = Document.ChangeListener { offer(it) } addChangeListener(listener) awaitClose { removeChangeListener(listener) } }
115. /** * 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) } }
116. Rx ➡ Flow
117. Reactive Stream Specification Flow Rx
118. https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-rx2 fun flow(): Flow<String> { return Observable.just("a") .asFlow() }
119. 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() }
120. Rx Coroutines Single / Completable / Maybe suspend function Observable Flow Subject Channel Schedulers Dispatchers Disposables Scopes subscribe collect Terminology between Rx and Coroutine world
121. Fun part
122. 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 */ } }
123. Rx1 ➡ Flow https://github.com/AkshayChordiya/kotlinx.coroutines-rx1
124. import rx.Observable fun observable(): Observable<String> { return flow { emit("a") }.asObservable() } import rx.Observable fun flow(): Flow<String> { return Observable.just("a") .asFlow() } https://github.com/AkshayChordiya/kotlinx.coroutines-rx1
125. Testing
126. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } }
127. @RunWith(JUnit4::class) class TestingWithFlow { fun getFlow() = flow { emit("Bob the builder ") } @Test fun testFlow() = runBlocking { } }
128. @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) } }
129. @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) } }
130. @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) } }
131. /** * 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()) }
132. @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) } }
133. Conclusion
134. Conclusion • Revisited coroutines • Why Flow? And what makes it special? • Basics of Flow • Principles of Flow • Legacy code -> Flow • Testing • Pro-tips along the way
135. 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
136. Further resources • News App https://github.com/AkshayChordiya/News • Go with the Kotlin Flow https://medium.com/google-developer-experts/go-with-the- kotlin-flow-7067564665a3 • Migrating from legacy code to Kotlin Flow https://medium.com/@aky/migrating-legacy-code-to-kotlin- flow-3f3e1854925
137. Thank You https://bit.ly/2Favsdn