RxBus.kt

interface RxBusSubscriber<in T> {
    fun onReceived(notification: T)
}

open private class RxBusSubscriberHolder<out T>(t: T) {

    private val ref: WeakReference<T> = WeakReference(t)

    fun get(): T? = ref.get()

    companion object {
        inline fun <reified T> create(instance: T): RxBusSubscriberHolder<T> = object : RxBusSubscriberHolder<T>(instance) {}
    }
}

open class TypeLiteral<T> {
    val type: Type = getSuperclassTypeParameter(javaClass)

    companion object {
        fun getSuperclassTypeParameter(subclass: Class<*>): Type =
                (subclass.genericSuperclass as ParameterizedType).actualTypeArguments[0]
    }
}

object RxBus {
    private var subscribers: MutableList<RxBusSubscriberHolder<*>> = mutableListOf()

    fun <T> subscribe(subscriber: RxBusSubscriber<T>) {
        subscribers.add(RxBusSubscriberHolder.create(subscriber))
    }

    fun <T> publish(event: T) {
        // Notify all the typed subscribers
        subscribers
                .filterIsInstance<RxBusSubscriberHolder<RxBusSubscriber<T>>>()
                .filter { object : TypeLiteral<RxBusSubscriberHolder<RxBusSubscriber<T>>>() {}.type == it.javaClass.genericSuperclass }
                .forEach { it.get()?.onReceived(event) }

        // Remove all the subscribers that have been released.
        subscribers.removeAll { it.get() == null }
    }
}