Kotlin State & Shared Flows in Action

Shared Flow APIs

Transcript

1. Mohit Sarveiya Kotlin State & Shared Flows in Action @heyitsmohit
2. Kotlin State & Shared Flows in Action • State Flow • Shared Flow • Broadcast Channel vs shared flow • Convert cold streams to shared flows • Buffer Overflow Strategies
3. Cold vs Hot Flows
4. What is a cold stream? A cold stream is a flow that triggers the same code every time it is collected.
5. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 }
6. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 } flow.collect { ... }
7. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 } flow.collect { ... }
8. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 } flow.collect { ... }
9. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 } flow.collect { // 2, 3, 4 }
10. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 } flow.collect { ... } flow.collect { ... }
11. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 } flow.collect { ... } flow.collect { ... }
12. Cold Flows val flow = flowOf(1, 2, 3) 
 .map { it + 1 } flow.collect { ... } flow.collect { // 2, 3, 4 }
13. What is a hot stream? A hot stream is a flow whose active instance exists independently 
 of the presence of collectors.
14. Hot Streams State Flow Shared Flow
15. State Flow View View Model
16. State Flow View View Model
17. State Flow View View Model State
18. State Flow sealed class UiState { 
 data class Error( 
 val exception: Throwable 
 ): UiState() }
19. State Flow sealed class UiState { 
 data class Success( val data: Data ): UiState() data class Error( 
 val exception: Throwable 
 ): UiState() }
20. State Flow sealed class UiState { 
 data class Success( val data: Data ): UiState() data class Error( 
 val exception: Throwable 
 ): UiState() }
21. State Flow val uiState = MutableStateFlow()
22. val uiState = MutableStateFlow( UiState.Success(Data()) ) State Flow
23. State Flow val uiState = MutableStateFlow( . .. ) uiState.emit( 
 UIState.Success(Data()) 
 )
24. State Flow val uiState = MutableStateFlow( . .. ) uiState.value = UIState.Success(Data()) 

25. State Flow val uiState = MutableStateFlow( . .. ) uiState.collect { ... } 
 Latest value is received
26. State Flow val uiState = MutableStateFlow( . .. ) uiState.collect { ... } 
 
 uiState.collect { ... } Latest value is received
27. State Flow Conflation val uiState = MutableStateFlow( . .. ) uiState.value = UIState.Success( .. . ) 
 
 uiState.value = UIState.Error( .. . ) Conflate
28. State Flow Conflation val uiState = MutableStateFlow( . .. ) uiState.value = UIState.Success( .. . ) 
 
 uiState.value = UIState.Error( .. . ) 
 
 uiState.collect { ... } Error
29. State Flow vs Live Data State Flow Live Data Default Value Unsubscribe 
 (Stopped State)
30. State Flow vs Live Data State Flow Live Data Default Value Unsubscribe 
 (Stopped State)
31. State Flow vs Live Data State Flow Live Data Default Value Unsubscribe 
 (Stopped State)
32. State Flow Summary • How to setup state flow • Emit and collect • State Flow vs Live Data
33. Shared Flow
34. Shared Flow
35. Shared Flow Consumer 1 Consumer 2
36. Shared Flow Consumer 1 Consumer 2
37. Shared Flow Consumer 1 Consumer 2 Event Event
38. Shared Flow Consumer 1 Consumer 2 Replay Replay
39. 
 Shared Flow Buffer
40. Shared Flow val flow = MutableSharedFlow < > ()
41. Shared Flow val flow = MutableSharedFlow<String>()
42. Shared Flow val flow = MutableSharedFlow<String>()
43. Shared Flow val flow = MutableSharedFlow<String>() launch { flow.collect { } }
44. Shared Flow val flow = MutableSharedFlow<String>() 
 launch { flow.emit("Event 1") } launch { flow.collect { } }
45. Shared Flow val flow = MutableSharedFlow<String>() 
 launch { flow.emit("Event 1") } launch { flow.collect { } } Event 1
46. Shared Flow val flow = MutableSharedFlow<String>() 
 launch { flow.emit("Event 1”) } launch { delay(2000); flow.collect { } }
47. Shared Flow val flow = MutableSharedFlow<String>() 
 launch { delay(2000); flow.collect { } } launch { flow.emit("Event 1”) }
48. Shared Flow val flow = MutableSharedFlow<String>() 
 launch { delay(2000); flow.collect { } } launch { flow.emit("Event 1”) }
49. Shared Flow val flow = MutableSharedFlow<String>() 
 launch { delay(2000); flow.collect { } } No value is received launch { flow.emit("Event 1”) }
50. Shared Flow val flow = MutableSharedFlow<String>() 
 launch { flow.emit("Event 1”) } launch { delay(2000); flow.collect { } } Replay
51. Shared Flow val flow = MutableSharedFlow<String>(replay = 1) 
 launch { flow.emit("Event 1”) } launch { delay(2000); flow.collect { } }
52. Shared Flow val flow = MutableSharedFlow<String>(replay = 1) 
 launch { delay(2000); flow.collect { } } launch { flow.emit("Event 1”) } Event 1
53. val flow = MutableSharedFlow<String>(replay = 1) 
 Shared Flow State Flow launch { flow.collect { } } launch { flow.subscriptionCount.value }
54. Shared Flow val flow = MutableSharedFlow<String>(replay = 1) 1 Subscriber launch { flow.collect { } } launch { flow.subscriptionCount.value }
55. Shared Flow val flow = MutableSharedFlow<String>(replay = 1) Shared Flow does not complete normally launch { flow.collect { } }
56. Cold Flows val flow = flowOf(1, 2, 3) 
 flow .onCompletion { } .collect { ... } Flow completes normally
57. Shared Flow val flow = MutableSharedFlow<String>(replay = 1) val job = launch { flow.collect { } } job.cancel()
58. Shared Flow val flow = MutableSharedFlow<String>(replay = 1) val job = launch { flow.onCompletion { }.collect { } } job.cancel() Flow completes exceptionally
59. Shared Flow Summary • Setup • Replay and emit • Cancellation
60. Broadcast Channel vs Shared Flow
61. Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
62. val channel = BroadcastChannel<Int>(10)
63. val channel = BroadcastChannel<Int>(10) channel.send( ... )
64. val channel = BroadcastChannel<Int>(10) channel.send( ... ) channel.close()
65. val stateFlow = MutableSharedFlow() stateFlow.emit( ... )
66. Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
67. val stateFlow = MutableSharedFlow(replay = 2) stateFlow.emit( ... )
68. Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
69. val channel = BroadcastChannel<Int>(capacity = 10) channel.send( ... )
70. val stateFlow = MutableSharedFlow( replay = 2, extraBufferCapacity = 10 ) stateFlow.emit( ... )
71. Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
72. val channel = BroadcastChannel<Int>(10) channel.send( ... ) channel.close()
73. val stateFlow = MutableSharedFlow( replay = 2, extraBufferCapacity = 10 ) stateFlow.emit( ... )
74. Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
75. Broadcast Channel Shared Flow
76. val channel = BroadcastChannel<Int>(capacity)
77. val channel = BroadcastChannel<Int>(capacity) val flow = MutableSharedFlow<String>(extraBufferCapacity)
78. channel.send( ... ) channel.trySend( ... ) flow.emit( ... ) flow.tryEmit( ... )
79. Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
80. Flow Sharing Strategies
81. Cold Flow Hot Flow Convert
82. Sharing Policies • While Subscribed • Eagerly • Lazily
83. Sharing Policies flow
84. Sharing Policies flow.shareIn( )
85. Sharing Policies flow.shareIn( externalScope, )
86. Sharing Policies flow.shareIn( externalScope, replay = 1, )
87. Sharing Policies flow.shareIn( externalScope, replay = 1, started = SharingStarted.WhileSubscribed() )
88. Sharing Policies val sharedFlow = flow.shareIn( externalScope, replay = 1, started = SharingStarted.WhileSubscribed() )
89. While Subscribed • Active as long as external scope is alive • Remains as long as there are collectors.
90. Properties Active as long as external scope is alive
91. Properties Active as long as external scope is alive sharedFlow.collect { } Subscriber
92. flow.shareIn( externalScope, replay = 1, started = SharingStarted.WhileSubscribed() ) Properties Active as long as external scope is alive
93. Properties Active as long as external scope is alive sharedFlow.collect { } externalScope.cancel()
94. Properties Active as long as external scope is alive sharedFlow.collect { } externalScope.cancel() Complete Exceptionally
95. Properties Remains as long as there are collectors.
96. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…)
97. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } }
98. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } } job.cancel()
99. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } } job.cancel()
100. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } } job.cancel()
101. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…) val job1 = launch { sharedFlow.collect { } } val job2 = launch { sharedFlow.collect { } }
102. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…) job1.cancel() val job2 = launch { sharedFlow.collect { } }
103. Properties Remains as long as there are collectors. val sharedFlow = flow.onCompletion { }.shareIn(…) job1.cancel() val job2 = launch { sharedFlow.collect { } } Remain Active
104. Properties • Active as long as external scope is alive • Remains as long as there are collectors.
105. Sharing Policies • While Subscribed • Eagerly • Lazily
106. Eagerly flow.shareIn( externalScope, replay = 1, started = SharingStarted.Eagerly() )
107. Eagerly Start producer eagerly and never stop flow .onStart { println("ON START") } .shareIn( ... started = SharingStarted.Eagerly)
108. Eagerly Start producer eagerly and never stop flow .onStart { println("ON START") } .shareIn( ... started = SharingStarted.Eagerly)
109. Eagerly Start producer eagerly and never stop flow .onStart { println("ON START") } .shareIn( ... started = SharingStarted.Eagerly) ON START
110. Eagerly Start producer eagerly and never stop flow .onComplete { println("ON COMPLETE”) } .shareIn( ... started = SharingStarted.Eagerly)
111. Eagerly Start producer eagerly and never stop flow .onComplete { println("ON COMPLETE”) } .shareIn( ... started = SharingStarted.Eagerly) externalScope.cancel()
112. Eagerly Start producer eagerly and never stop flow .onComplete { println("ON COMPLETE”) } .shareIn( ... started = SharingStarted.Eagerly) Never stops externalScope.cancel()
113. Eagerly Start producer eagerly and never stop
114. Sharing Policies • While Subscribed • Eagerly • Lazily
115. Lazily Start sharing after the first subscriber appears and never stop
116. Lazily flow.shareIn( externalScope, replay = 1, started = SharingStarted.Lazily )
117. Lazily flow .onStart { println("ON START") } .shareIn(…,started = SharingStarted.Lazily)
118. Lazily flow .onStart { println("ON START") } .shareIn(…,started = SharingStarted.Lazily) launch { sharedFlow.collect { } }
119. Lazily flow .onStart { println("ON START") } .shareIn(…,started = SharingStarted.Lazily) launch { sharedFlow.collect { } } "ON START"
120. Lazily flow .onCompletion { println("COMPLETE") } .shareIn(…,started = SharingStarted.Lazily)
121. flow .onCompletion { println("COMPLETE") } .shareIn(externalScope,…,started = SharingStarted.Lazily) Lazily
122. flow .onCompletion { println("COMPLETE") } .shareIn(externalScope,…,started = SharingStarted.Lazily) Lazily externalScope.cancel()
123. flow .onCompletion { println("COMPLETE") } .shareIn(externalScope,…,started = SharingStarted.Lazily) Lazily Never stops externalScope.cancel()
124. Lazily Start sharing after the first subscriber appears and never stop
125. Sharing Policies • While Subscribed • Active while there are active subscribers. 
 • Eagerly • Start producer eagerly and never stop 
 • Lazily • Start after the first subscriber appears and never stop
126. Buffer Overflow Strategies
127. 
 Shared Flow Buffer
128. 
 Shared Flow Producer Consumer
129. 
 Shared Flow Producer Consumer Generating events fast
130. 
 Shared Flow Producer Consumer Listening to events 
 with delay
131. 
 Shared Flow Producer Consumer
132. 
 Shared Flow Producer Consumer
133. 
 Shared Flow Producer Consumer What happens when it is full?
134. Buffering Overflow Strategies • Suspend • Drop oldest • Drop latest
135. 
 Shared Flow Producer Consumer Suspend
136. Buffering Overflow Strategies val flow = MutableSharedFlow<String>( extraBufferCapacity = 2, onBufferOverflow = BufferOverflow.SUSPEND ) Buffer + Replay Count
137. Buffering Overflow Strategies val flow = MutableSharedFlow<String>( extraBufferCapacity = 2, onBufferOverflow = BufferOverflow.SUSPEND )
138. Buffering Overflow Strategies launch { flow.emit("Event 1") flow.emit("Event 2") flow.emit("Event 3") } Suspend
139. Buffering Overflow Strategies • Suspend • Drop oldest • Drop latest
140. 
 Shared Flow Producer Consumer Drop Oldest
141. 
 Shared Flow Producer Consumer Drop latest
142. Buffering Overflow Strategies • Suspend • Drop oldest • Drop latest
143. Kotlin State & Shared Flows in Action • State Flow • Shared Flow • Broadcast Channel vs shared flow • Convert cold streams to shared flows • Buffer Overflow Strategies
144. https: // codingwithmohit.com/coroutines/learning-shared-and-state-flows-with-tests/ Coding with Mohit
145. Thank You! www.codingwithmohit.com @heyitsmohit