Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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")
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.