Flow examples

flow.kt

fun main() {
    runBlocking {
        println("=== Example 1: Cold Flow (default behavior) ===")
        // Cold flow: executes independently for each collector
        val coldFlow = flow {
            println("Cold Flow started")
            emit("A")
            delay(100)
            emit("B")
        }

        coldFlow.collect { println("First collector: $it") }
        coldFlow.collect { println("Second collector: $it") }
        // Output: "Cold Flow started" printed TWICE - each collector triggers execution

        println("\n=== Example 2: shareIn - WRONG WAY (hardcoded delays) ===")
        val sharedScope2 = CoroutineScope(Job() + Dispatchers.Default)
        val sharedFlowWrong = flow {
            println("Shared Flow started...")
            emit("C")
            delay(100)
            emit("D")
        }.shareIn(
            scope = sharedScope2,
            started = SharingStarted.Lazily,
            replay = 0
        )

        val job1 = launch {
            sharedFlowWrong.collect { println("First collector: $it") }
        }
        val job2 = launch {
            sharedFlowWrong.collect { println("Second collector: $it") }
        }
        delay(300) // BAD: Hardcoded delay - fragile and arbitrary
        job1.cancel()
        job2.cancel()
        sharedScope2.cancel() // Clean up this scope

        println("\n=== Example 2a: shareIn - RIGHT WAY with take() ===")
        // Option A: Use take() to collect only specific number of items
        val sharedScope2a = CoroutineScope(Job() + Dispatchers.Default)
        val sharedFlowTake = flow {
            println("Shared Flow started...")
            emit("E")
            delay(100)
            emit("F")
            delay(100)
            emit("G")
        }.shareIn(
            scope = sharedScope2a,
            started = SharingStarted.Lazily,
            replay = 1
        )

        // take(2) automatically completes after 2 emissions - no hardcoded delays!
        launch {
            sharedFlowTake.take(2).collect { println("First collector: $it") }
        }.join() // join() is safe here because take(2) completes automatically
        launch {
            sharedFlowTake.take(2).collect { println("Second collector: $it") }
        }.join() // join() is safe here because take(2) completes automatically
        sharedScope2a.cancel() // Clean up this scope

        println("\n=== Example 2b: shareIn - onCompletion behavior ===")
        // onCompletion on the upstream flow (before shareIn)
        val sharedScope2b = CoroutineScope(Job() + Dispatchers.Default)
        val sharedFlowOnCompletion = flow {
            println("Shared Flow started...")
            repeat(3) { i ->
                delay(100)
                emit("Item $i")
            }
            println("Flow finished emitting (but not necessarily completed)")
        }
            .onCompletion {
                println("onCompletion triggered! (upstream flow completed)")
            }
            .shareIn(
                scope = sharedScope2b,
                started = SharingStarted.Lazily,
                replay = 1
            )

        launch {
            sharedFlowOnCompletion
                .take(2)
                .onCompletion { println("First collector completed (via take)") }
                .collect { println("First collector: $it") }
        }.join()

        launch {
            sharedFlowOnCompletion
                .take(1)
                .onCompletion { println("Second collector completed (via take)") }
                .collect { println("Second collector: $it") }
        }.join()

        // Cancel the scope to complete the upstream flow
        delay(100)
        sharedScope2b.cancel()
        println("Shared scope cancelled")

        println("\n=== Example 3: stateIn - REAL WORLD PATTERN ===")
        // In production: StateFlow is typically a class property with lifecycle-aware scope
        val scope3 = CoroutineScope(Job() + Dispatchers.Default)

        val stateFlow = flow {
            println("State Flow started")
            emit(1)
            delay(100)
            emit(2)
            delay(100)
            emit(3)
        }.stateIn(
            scope = scope3,
            started = SharingStarted.WhileSubscribed(5000), // Stops 5s after last subscriber
            initialValue = 0
        )

        println("Initial value: ${stateFlow.value}")

        // First collector
        val job3 = launch {
            stateFlow.collect { println("StateFlow - Collector 1: $it") }
        }

        delay(150) // Simulate some time passing

        // Late collector gets latest value immediately
        val job4 = launch {
            stateFlow.collect { println("StateFlow - Collector 2 (late): $it") }
        }

        delay(200) // Let emissions complete
        println("Final value: ${stateFlow.value}")

        job3.cancel()
        job4.cancel()
        scope3.cancel() // ✅ Clean up properly

        println("\n=== DELAY CALCULATION STRATEGIES ===")
        println("❌ WRONG: Hardcoded delays like delay(300)")
        println("   - Fragile, depends on knowing flow timing")
        println("   - Breaks if flow logic changes")
        println("")
        println("✅ RIGHT: Use proper patterns:")
        println("1. SharingStarted.WhileSubscribed(stopTimeoutMillis)")
        println("   - Automatically stops when no subscribers")
        println("   - Best for ViewModel/Repository patterns")
        println("")
        println("2. SharingStarted.Eagerly")
        println("   - Starts immediately, never stops")
        println("   - Use when flow should always be active")
        println("")
        println("3. flow.take(n) or flow.first()")
        println("   - Limits emissions, auto-completes")
        println("")
        println("4. Lifecycle-aware scopes (viewModelScope, lifecycleScope)")
        println("   - Automatically cancelled when lifecycle ends")
    }
}