Upgrade to Pro
— share decks privately, control downloads, hide ads and more …
Speaker Deck
Features
Speaker Deck
PRO
Sign in
Sign up for free
Search
Search
Kotlin State & Shared Flows in Action
Search
Mohit S
July 27, 2021
Programming
3
1.1k
Kotlin State & Shared Flows in Action
Shared Flow APIs
Mohit S
July 27, 2021
Tweet
Share
More Decks by Mohit S
See All by Mohit S
Building Shared UIs across Platforms with Compose
heyitsmohit
1
380
Building Multiplatform Apps with Compose
heyitsmohit
2
310
Building StateFlows with Jetpack Compose
heyitsmohit
6
1.6k
Building Android Testing Infrastructure
heyitsmohit
1
320
Migrating to Kotlin State & Shared Flows
heyitsmohit
1
560
Using Square Workflow for Android & iOS
heyitsmohit
1
320
Building Android Infrastructure Teams at Scale
heyitsmohit
2
230
Strategies for Migrating to Jetpack Compose
heyitsmohit
2
440
Challenges of Building Kotlin Multiplatform Libraries
heyitsmohit
1
370
Other Decks in Programming
See All in Programming
Findy - エンジニア向け会社紹介 / Findy Letter for Engineers
findyinc
2
74k
RailsConf 2024: Riffing on Rails: sketch your way to better designed code
kaspth
1
220
HonoのRPCで真の型安全が欲しかった
kosei28
1
160
FoodGram
iseruuuuu
0
230
mb_trim関数を作りました
youkidearitai
PRO
1
240
GoでParserを書く
karupanerura
3
440
Deep Dive into React Stream/Serialize
mugi_uno
4
870
Using "modern" Ruby to build a better, faster Homebrew
mikemcquaid
2
290
ソースコードを美しくたもつために ~コードレビューの認知限界を突破し、年間400リリースを達成する~
kotauchisunsun
1
790
Dev ContainersとTestcontainers
bells17
3
100
TypeScriptの型とパフォーマンス (TSKaigi 2024)
ypresto
14
5.1k
TypeScriptで使いやすいOpenAPIの書き方
yukimochi_dwango
1
950
Featured
See All Featured
A Tale of Four Properties
chriscoyier
153
22k
Sharpening the Axe: The Primacy of Toolmaking
bcantrill
22
1.4k
How to Ace a Technical Interview
jacobian
273
22k
Happy Clients
brianwarren
92
6.4k
Imperfection Machines: The Place of Print at Facebook
scottboms
261
12k
Thoughts on Productivity
jonyablonski
60
3.9k
Designing Experiences People Love
moore
136
23k
CoffeeScript is Beautiful & I Never Want to Write Plain JavaScript Again
sstephenson
155
14k
RailsConf 2023
tenderlove
9
590
Creating an realtime collaboration tool: Agile Flush - .NET Oxford
marcduiker
14
1.5k
Statistics for Hackers
jakevdp
790
220k
It's Worth the Effort
3n
180
27k
Transcript
Mohit Sarveiya Kotlin State & Shared Flows in Action @heyitsmohit
Kotlin State & Shared Flows in Action • State Flow
• Shared Flow • Broadcast Channel vs shared flow • Convert cold streams to shared flows • Buffer Overflow Strategies
Cold vs Hot Flows
What is a cold stream? A cold stream is a
flow that triggers the same code every time it is collected.
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 }
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 } flow.collect { ... }
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 } flow.collect { ... }
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 } flow.collect { ... }
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 } flow.collect { // 2, 3, 4 }
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 } flow.collect { ... } flow.collect { ... }
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 } flow.collect { ... } flow.collect { ... }
Cold Flows val flow = flowOf(1, 2, 3) .map
{ it + 1 } flow.collect { ... } flow.collect { // 2, 3, 4 }
What is a hot stream? A hot stream is a
flow whose active instance exists independently of the presence of collectors.
Hot Streams State Flow Shared Flow
State Flow View View Model
State Flow View View Model
State Flow View View Model State
State Flow sealed class UiState { data class Error(
val exception: Throwable ): UiState() }
State Flow sealed class UiState { data class Success(
val data: Data ): UiState() data class Error( val exception: Throwable ): UiState() }
State Flow sealed class UiState { data class Success(
val data: Data ): UiState() data class Error( val exception: Throwable ): UiState() }
State Flow val uiState = MutableStateFlow()
val uiState = MutableStateFlow( UiState.Success(Data()) ) State Flow
State Flow val uiState = MutableStateFlow( . .. ) uiState.emit(
UIState.Success(Data()) )
State Flow val uiState = MutableStateFlow( . .. ) uiState.value
= UIState.Success(Data())
State Flow val uiState = MutableStateFlow( . .. ) uiState.collect
{ ... } Latest value is received
State Flow val uiState = MutableStateFlow( . .. ) uiState.collect
{ ... } uiState.collect { ... } Latest value is received
State Flow Conflation val uiState = MutableStateFlow( . .. )
uiState.value = UIState.Success( .. . ) uiState.value = UIState.Error( .. . ) Conflate
State Flow Conflation val uiState = MutableStateFlow( . .. )
uiState.value = UIState.Success( .. . ) uiState.value = UIState.Error( .. . ) uiState.collect { ... } Error
State Flow vs Live Data State Flow Live Data Default
Value Unsubscribe (Stopped State)
State Flow vs Live Data State Flow Live Data Default
Value Unsubscribe (Stopped State)
State Flow vs Live Data State Flow Live Data Default
Value Unsubscribe (Stopped State)
State Flow Summary • How to setup state flow •
Emit and collect • State Flow vs Live Data
Shared Flow
Shared Flow
Shared Flow Consumer 1 Consumer 2
Shared Flow Consumer 1 Consumer 2
Shared Flow Consumer 1 Consumer 2 Event Event
Shared Flow Consumer 1 Consumer 2 Replay Replay
Shared Flow Buffer
Shared Flow val flow = MutableSharedFlow < > ()
Shared Flow val flow = MutableSharedFlow<String>()
Shared Flow val flow = MutableSharedFlow<String>()
Shared Flow val flow = MutableSharedFlow<String>() launch { flow.collect {
} }
Shared Flow val flow = MutableSharedFlow<String>() launch { flow.emit("Event
1") } launch { flow.collect { } }
Shared Flow val flow = MutableSharedFlow<String>() launch { flow.emit("Event
1") } launch { flow.collect { } } Event 1
Shared Flow val flow = MutableSharedFlow<String>() launch { flow.emit("Event
1”) } launch { delay(2000); flow.collect { } }
Shared Flow val flow = MutableSharedFlow<String>() launch { delay(2000);
flow.collect { } } launch { flow.emit("Event 1”) }
Shared Flow val flow = MutableSharedFlow<String>() launch { delay(2000);
flow.collect { } } launch { flow.emit("Event 1”) }
Shared Flow val flow = MutableSharedFlow<String>() launch { delay(2000);
flow.collect { } } No value is received launch { flow.emit("Event 1”) }
Shared Flow val flow = MutableSharedFlow<String>() launch { flow.emit("Event
1”) } launch { delay(2000); flow.collect { } } Replay
Shared Flow val flow = MutableSharedFlow<String>(replay = 1) launch
{ flow.emit("Event 1”) } launch { delay(2000); flow.collect { } }
Shared Flow val flow = MutableSharedFlow<String>(replay = 1) launch
{ delay(2000); flow.collect { } } launch { flow.emit("Event 1”) } Event 1
val flow = MutableSharedFlow<String>(replay = 1) Shared Flow State
Flow launch { flow.collect { } } launch { flow.subscriptionCount.value }
Shared Flow val flow = MutableSharedFlow<String>(replay = 1) 1 Subscriber
launch { flow.collect { } } launch { flow.subscriptionCount.value }
Shared Flow val flow = MutableSharedFlow<String>(replay = 1) Shared Flow
does not complete normally launch { flow.collect { } }
Cold Flows val flow = flowOf(1, 2, 3) flow
.onCompletion { } .collect { ... } Flow completes normally
Shared Flow val flow = MutableSharedFlow<String>(replay = 1) val job
= launch { flow.collect { } } job.cancel()
Shared Flow val flow = MutableSharedFlow<String>(replay = 1) val job
= launch { flow.onCompletion { }.collect { } } job.cancel() Flow completes exceptionally
Shared Flow Summary • Setup • Replay and emit •
Cancellation
Broadcast Channel vs Shared Flow
Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
val channel = BroadcastChannel<Int>(10)
val channel = BroadcastChannel<Int>(10) channel.send( ... )
val channel = BroadcastChannel<Int>(10) channel.send( ... ) channel.close()
val stateFlow = MutableSharedFlow() stateFlow.emit( ... )
Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
val stateFlow = MutableSharedFlow(replay = 2) stateFlow.emit( ... )
Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
val channel = BroadcastChannel<Int>(capacity = 10) channel.send( ... )
val stateFlow = MutableSharedFlow( replay = 2, extraBufferCapacity = 10
) stateFlow.emit( ... )
Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
val channel = BroadcastChannel<Int>(10) channel.send( ... ) channel.close()
val stateFlow = MutableSharedFlow( replay = 2, extraBufferCapacity = 10
) stateFlow.emit( ... )
Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
Broadcast Channel Shared Flow
val channel = BroadcastChannel<Int>(capacity)
val channel = BroadcastChannel<Int>(capacity) val flow = MutableSharedFlow<String>(extraBufferCapacity)
channel.send( ... ) channel.trySend( ... ) flow.emit( ... ) flow.tryEmit(
... )
Broadcast Channel Shared Flow Channel APIs Replay Buffer Closed
Flow Sharing Strategies
Cold Flow Hot Flow Convert
Sharing Policies • While Subscribed • Eagerly • Lazily
Sharing Policies flow
Sharing Policies flow.shareIn( )
Sharing Policies flow.shareIn( externalScope, )
Sharing Policies flow.shareIn( externalScope, replay = 1, )
Sharing Policies flow.shareIn( externalScope, replay = 1, started = SharingStarted.WhileSubscribed()
)
Sharing Policies val sharedFlow = flow.shareIn( externalScope, replay = 1,
started = SharingStarted.WhileSubscribed() )
While Subscribed • Active as long as external scope is
alive • Remains as long as there are collectors.
Properties Active as long as external scope is alive
Properties Active as long as external scope is alive sharedFlow.collect
{ } Subscriber
flow.shareIn( externalScope, replay = 1, started = SharingStarted.WhileSubscribed() ) Properties
Active as long as external scope is alive
Properties Active as long as external scope is alive sharedFlow.collect
{ } externalScope.cancel()
Properties Active as long as external scope is alive sharedFlow.collect
{ } externalScope.cancel() Complete Exceptionally
Properties Remains as long as there are collectors.
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…)
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } }
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } } job.cancel()
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } } job.cancel()
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…) val job = launch { sharedFlow.onCompletion { }.collect { } } job.cancel()
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…) val job1 = launch { sharedFlow.collect { } } val job2 = launch { sharedFlow.collect { } }
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…) job1.cancel() val job2 = launch { sharedFlow.collect { } }
Properties Remains as long as there are collectors. val sharedFlow
= flow.onCompletion { }.shareIn(…) job1.cancel() val job2 = launch { sharedFlow.collect { } } Remain Active
Properties • Active as long as external scope is alive
• Remains as long as there are collectors.
Sharing Policies • While Subscribed • Eagerly • Lazily
Eagerly flow.shareIn( externalScope, replay = 1, started = SharingStarted.Eagerly() )
Eagerly Start producer eagerly and never stop flow .onStart {
println("ON START") } .shareIn( ... started = SharingStarted.Eagerly)
Eagerly Start producer eagerly and never stop flow .onStart {
println("ON START") } .shareIn( ... started = SharingStarted.Eagerly)
Eagerly Start producer eagerly and never stop flow .onStart {
println("ON START") } .shareIn( ... started = SharingStarted.Eagerly) ON START
Eagerly Start producer eagerly and never stop flow .onComplete {
println("ON COMPLETE”) } .shareIn( ... started = SharingStarted.Eagerly)
Eagerly Start producer eagerly and never stop flow .onComplete {
println("ON COMPLETE”) } .shareIn( ... started = SharingStarted.Eagerly) externalScope.cancel()
Eagerly Start producer eagerly and never stop flow .onComplete {
println("ON COMPLETE”) } .shareIn( ... started = SharingStarted.Eagerly) Never stops externalScope.cancel()
Eagerly Start producer eagerly and never stop
Sharing Policies • While Subscribed • Eagerly • Lazily
Lazily Start sharing after the first subscriber appears and never
stop
Lazily flow.shareIn( externalScope, replay = 1, started = SharingStarted.Lazily )
Lazily flow .onStart { println("ON START") } .shareIn(…,started = SharingStarted.Lazily)
Lazily flow .onStart { println("ON START") } .shareIn(…,started = SharingStarted.Lazily)
launch { sharedFlow.collect { } }
Lazily flow .onStart { println("ON START") } .shareIn(…,started = SharingStarted.Lazily)
launch { sharedFlow.collect { } } "ON START"
Lazily flow .onCompletion { println("COMPLETE") } .shareIn(…,started = SharingStarted.Lazily)
flow .onCompletion { println("COMPLETE") } .shareIn(externalScope,…,started = SharingStarted.Lazily) Lazily
flow .onCompletion { println("COMPLETE") } .shareIn(externalScope,…,started = SharingStarted.Lazily) Lazily externalScope.cancel()
flow .onCompletion { println("COMPLETE") } .shareIn(externalScope,…,started = SharingStarted.Lazily) Lazily Never
stops externalScope.cancel()
Lazily Start sharing after the first subscriber appears and never
stop
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
Buffer Overflow Strategies
Shared Flow Buffer
Shared Flow Producer Consumer
Shared Flow Producer Consumer Generating events fast
Shared Flow Producer Consumer Listening to events with
delay
Shared Flow Producer Consumer
Shared Flow Producer Consumer
Shared Flow Producer Consumer What happens when it is
full?
Buffering Overflow Strategies • Suspend • Drop oldest • Drop
latest
Shared Flow Producer Consumer Suspend
Buffering Overflow Strategies val flow = MutableSharedFlow<String>( extraBufferCapacity = 2,
onBufferOverflow = BufferOverflow.SUSPEND ) Buffer + Replay Count
Buffering Overflow Strategies val flow = MutableSharedFlow<String>( extraBufferCapacity = 2,
onBufferOverflow = BufferOverflow.SUSPEND )
Buffering Overflow Strategies launch { flow.emit("Event 1") flow.emit("Event 2") flow.emit("Event
3") } Suspend
Buffering Overflow Strategies • Suspend • Drop oldest • Drop
latest
Shared Flow Producer Consumer Drop Oldest
Shared Flow Producer Consumer Drop latest
Buffering Overflow Strategies • Suspend • Drop oldest • Drop
latest
Kotlin State & Shared Flows in Action • State Flow
• Shared Flow • Broadcast Channel vs shared flow • Convert cold streams to shared flows • Buffer Overflow Strategies
https: // codingwithmohit.com/coroutines/learning-shared-and-state-flows-with-tests/ Coding with Mohit
Thank You! www.codingwithmohit.com @heyitsmohit