EventBus.kt

package kotlin_playground

import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.channels.sendBlocking
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.filter
import kotlin.reflect.KClass

class EventBus {

    private val publisher: SendChannel<Event>
    private val receiver: Flow<Event>

    init {
        val channel = BroadcastChannel<Event>(Channel.BUFFERED)
        publisher = channel
        receiver = channel.asFlow()
    }

    inline fun <reified E : Event> subscribe(): Flow<E> = subscribe(E::class)

    fun <E : Event> subscribe(eventClass: KClass<E>): Flow<E> {
        @Suppress("UNCHECKED_CAST")
        return receiver.filter { eventClass.isInstance(it) } as Flow<E>
    }

    fun notify(event: Event) {
        // This can cause back-pressure due to downstream needing time to process existing events
        publisher.sendBlocking(event)
    }

}

interface Event