coroutines-con-kotlin.updates.sharedflow-02.kt

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.*

fun log(message: String) {
    println("[${Thread.currentThread().name}] : $message")
}

fun main() {
    log("Start")

    val myFlow = MutableSharedFlow<String>(1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
    myFlow.tryEmit("Initial Value")

    runBlocking {

        val job = launch(Dispatchers.Default) {
            myFlow
                .distinctUntilChanged()
                .collect {
                log("Collected => $it")
            }
        }

        launch {
            listOf(1, 2, 3, 3, 3, 4, 5, 6).forEach {
                log("Updating => $it")
                myFlow.emit("$it")
                delay(300)
            }
            delay(1500)
            job.cancel()
        }
    }

    log("End")
}