EventBus by Kotlin coroutine

EventBus.kt

import kotlinx.coroutines.experimental.channels.BroadcastChannel
import kotlinx.coroutines.experimental.channels.ConflatedBroadcastChannel
import kotlinx.coroutines.experimental.channels.ReceiveChannel
import kotlinx.coroutines.experimental.channels.filter
import kotlinx.coroutines.experimental.channels.map
import kotlinx.coroutines.experimental.launch
import javax.inject.Inject
import javax.inject.Singleton

/**
 * You can use like this.
 * val channel = EventBus().asChannel<ItemChangeAction>()
 * launch (UI){
 *   for(action in channel){
 *     // You can use item
 *     action.item
 *   }
 * }
 */
@Singleton
class EventBus @Inject constructor() {
    val bus: BroadcastChannel<Any> = ConflatedBroadcastChannel<Any>()

    fun send(o: Any) {
        launch {
            bus.send(o)
        }
    }

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

I found my own solution.

Thanks for your idea, it's amazingly simple. I would suggest a slightly manipulated solution, because the documentation of ConflatedBroadcastChannel says:

Every subscriber immediately receives the most recently sent element.

This behaviour is (in my understanding) unwanted in case of the EventBus, because Subscribers would eventually receive an event, that was emitted before the Subscribtion. You can reproduce this with:

@Test
fun `should receive only events occuring after subscription`() = runBlockingTest {
    val publisher = ConflatedBroadcastChannel<Int>()
    val values = mutableListOf<Int>()

    publisher.offer(1)
    publisher.offer(2)

    val job = launch {
        publisher.openSubscription().consumeEach { values.add(it) }
    }

    publisher.offer(3)
    assertEquals(listOf(3), values)
    assertEquals(3, publisher.value)

    job.cancel()
}

This test fails with:

org.opentest4j.AssertionFailedError:
Expected :[3]
Actual :[2, 3]

Creating the Channel with an initial empty event and for every subscriber skipping the first event will avoid this beaviour:

internal class EventBus {
    private val bus = ConflatedBroadcastChannel<Any>(object {})

    suspend fun send(o : Any) { bus.send(o) }

    final inline fun <reified T> on() = bus.asFlow().drop(1).filter { it is T }.map { it as T }
}

this is for kotlin version ="1.4.20"

@ExperimentalCoroutinesApi

object CoroutinesEvent {

private val observerChanner = BroadcastChannel<Any>(Channel.BUFFERED)

suspend fun publish(Response: Any) {

    observerChanner.send(Response)

}

fun <Any> listen(eventType: Class<Any>): ReceiveChannel<Any> =
    observerChanner.openSubscription().filter { it is kotlin.Any }.map { it as Any }

}