EventBus.kt

object EventBus {
    @OptIn(ExperimentalCoroutinesApi::class)
    val bus: BroadcastChannel<Any> = ConflatedBroadcastChannel(2) // receive only most recent data

    @ExperimentalCoroutinesApi
    fun send(o: Any) = runBlocking {
        launch {
            bus.send(o)
        }
    }

    @ExperimentalCoroutinesApi
    inline fun <reified T> asFlow(): Flow<T> {
        return bus.openSubscription().filter { it is T }.map { it as T }.consumeAsFlow()
    }

    inline fun <reified T> asChannel(): ReceiveChannel<T> {
        return bus.openSubscription().filter { it is T }.map { it as T }
    }
}

// updated as of 6/22/2021

@OptIn(ExperimentalCoroutinesApi::class)
object GameEventManager {
    private val _events = MutableSharedFlow<Any>() // no buffer, rendezvous with subscribers
    val events: SharedFlow<Any> get() = _events // expose as a plain flow

    inline fun <reified T> asChannel(crossinline action: (value: T) -> Unit) {
        KtxAsync.launch {
            events.filterIsInstance<T>().collect { event-> action.invoke(event) }
        }
    }

    fun send(e: Any) {
        KtxAsync.launch {
            produceEvent(e)
        }
    }

    private suspend fun produceEvent(event: Any) {
        _events.emit(event) // suspends until all subscribers receive it
    }
}