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

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

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

sealed class MyElement<T> {
    class MyToken<T>(val token: T): MyElement<T>()
    class MyFinalToken<T>: MyElement<T>()
}

fun main() {
    log("Start")

    val myFlow = MutableSharedFlow<MyElement<Int>>(replay = 2)

    runBlocking {

        launch(Dispatchers.Default) {
            val lastValue = myFlow
                .onStart {
                    log("Collector started collecting...")
                }
                .filter { it is MyElement.MyFinalToken }
                .map { (myFlow.replayCache[0] as MyElement.MyToken).token }
                .first()

            log("Last Value => $lastValue")
        }

        launch {
            listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).forEach {
                val element = MyElement.MyToken(it)
                log("Emitting => ${element.token}")
                myFlow.emit(element)
                delay(100)
            }
            myFlow.emit(MyElement.MyFinalToken())
        }
    }

    log("End")
}