RendezvousBroadcastChannel

import kotlinx.atomicfu.AtomicRef
import kotlinx.atomicfu.atomic
import kotlinx.coroutines.experimental.DefaultDispatcher
import kotlinx.coroutines.experimental.channels.ReceiveChannel
import kotlinx.coroutines.experimental.channels.RendezvousChannel
import kotlinx.coroutines.experimental.channels.filter
import kotlinx.coroutines.experimental.channels.map
import kotlinx.coroutines.experimental.launch
import kotlin.coroutines.experimental.CoroutineContext

/**
 * Coroutine-based event bus.
 */
class EventBus {

    private val subscribers = atomic<Iterable<Subscriber>>(emptyList())

    fun send(event: Any, context: CoroutineContext = DefaultDispatcher) {
        launch(context) {
            subscribers.value.forEach { subscriber ->
                subscriber.send(event)
            }
        }
    }

    fun subscribe(): ReceiveChannel<Any> =
        Subscriber(subscribers).also { subscriber ->
            subscribers.value = subscribers.value + subscriber
        }

    inline fun <reified T> subscribeToEvent() =
        subscribe().filter { it is T }.map { it as T }

    private class Subscriber(private val subscribers: AtomicRef<Iterable<Subscriber>>) : RendezvousChannel<Any>() {
        override fun cancel(cause: Throwable?): Boolean =
            close(cause).also {
                subscribers.value = subscribers.value - this
            }
    }
}