แปลง Cold Flow เป็น Hot ด้วย shareIn และ stateIn

โฟลว์เย็นกับโฟลว์ร้อน

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun expensiveFlow() = flow {
    println("Starting upstream work")  // runs once if shared
    repeat(3) { delay(100); emit(it) }
}
fun main() = runBlocking {
    val cold = expensiveFlow()
    // Two collectors = two executions:
    launch { cold.collect { } }
    launch { cold.collect { } }
    delay(500)
    coroutineContext.cancelChildren()
}

พื้นฐานของ shareIn

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val cold = flow {
        println("Upstream started once")
        repeat(5) { delay(100); emit(it) }
    }
    val hot = cold.shareIn(this, SharingStarted.Eagerly, replay = 0)
    launch { hot.collect { println("A: $it") } }
    launch { hot.collect { println("B: $it") } }
    delay(600)
    coroutineContext.cancelChildren()
}

พื้นฐานของ stateIn

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val cold = flow { delay(100); emit(42) }
    val state: StateFlow<Int> = cold.stateIn(
        scope = this,
        started = SharingStarted.Eagerly,
        initialValue = 0
    )
    println(state.value)  // 0 immediately
    delay(200)
    println(state.value)  // 42 after upstream emits
    coroutineContext.cancelChildren()
}

SharingStarted.Eagerly

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val flow = flow {
        println("Eagerly started")
        emit(1)
    }.shareIn(this, SharingStarted.Eagerly)
    // Upstream already running even before any collect
    delay(50)
    launch { flow.collect { println(it) } }
    delay(100)
    coroutineContext.cancelChildren()
}

SharingStarted.Lazily

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val flow = flow {
        println("Lazily started on first subscriber")
        repeat(3) { delay(100); emit(it) }
    }.shareIn(this, SharingStarted.Lazily, replay = 1)
    delay(50)  // no subscriber yet — not started
    launch { flow.collect { println(it) } }  // triggers start
    delay(400)
    coroutineContext.cancelChildren()
}

SharingStarted.WhileSubscribed

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val flow = flow {
        println("Started")
        repeat(10) { delay(100); emit(it) }
    }.shareIn(this, SharingStarted.WhileSubscribed(500))
    val job = launch { flow.collect { print("$it ") } }
    delay(300)
    job.cancel()    // subscriber left
    delay(200)      // within 500ms stop timeout — still running
    launch { flow.collect { print("resume $it ") } }
    delay(500)
    coroutineContext.cancelChildren()
}

รูปแบบทั่วไปของ ViewModel

import kotlinx.coroutines.flow.*
// In ViewModel:
// val uiState: StateFlow<UiState> = repository
//     .dataFlow()
//     .map { UiState.Success(it) }
//     .stateIn(
//         scope = viewModelScope,
//         started = SharingStarted.WhileSubscribed(5_000),
//         initialValue = UiState.Loading
//     )
fun main() { println("WhileSubscribed(5000) is the recommended ViewModel pattern") }

shareIn กับ stateIn

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val cold = (1..3).asFlow().map { it * 10 }

    // SharedFlow — no initial value:
    val shared: SharedFlow<Int> = cold.shareIn(this, SharingStarted.Eagerly, replay = 1)

    // StateFlow — always has a value:
    val state: StateFlow<Int> = cold.stateIn(this, SharingStarted.Eagerly, 0)

    delay(100)
    println("shared cache: ${shared.replayCache}")
    println("state value: ${state.value}")
    coroutineContext.cancelChildren()
}

ข้อแลกเปลี่ยนของแคชการเล่นซ้ำ

import kotlinx.coroutines.flow.*
// replay=0: no history, only future events
// replay=1: last value (equivalent to stateIn)
// replay=N: last N events — use for message feeds, logs

// Trade-off: memory vs subscriber freshness
fun main() { println("Choose replay based on late-subscriber requirements") }

ความเสี่ยงของโฟลว์ที่ไม่ได้แชร์

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun expensiveApi() = flow {
    println("API CALL")  // without sharing: once per collector
    emit("data")
}
fun main() = runBlocking {
    val shared = expensiveApi().shareIn(this, SharingStarted.Lazily, replay = 1)
    launch { shared.collect { } }  // one API call, shared
    launch { shared.collect { } }  // same emission
    delay(200)
    coroutineContext.cancelChildren()
}

การล้างทรัพยากรเมื่อแชร์ข้อมูล

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun resourceFlow() = flow {
    try { repeat(10) { delay(100); emit(it) } }
    finally { println("Upstream cleaned up") }
}
fun main() = runBlocking {
    val scope = CoroutineScope(SupervisorJob())
    val shared = resourceFlow().shareIn(scope, SharingStarted.Eagerly)
    launch { shared.collect { print("$it ") } }
    delay(250)
    scope.cancel()  // upstream cleanup runs
    delay(100)
}

ตรวจสอบอย่างรวดเร็ว

ทบทวน