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 }
    }
}

Really nice. Thanks for sharing. Do you know if it is possible to do pub/sub using coroutines? Then I don't need rxjava anymore.

Interesting, however I have an issue with this implementation. ConflatedBroadcastChannel keeps the last sent item and delivers it to new subscribers in openSubscription(). I'm looking for an event bus implementation that does not cache items. So if an event is send but there are no active subscriptions the bus should just drop the event. Unfortunately there doesn't seem to be a default implementation of BroadcastChannel with these characteristics.

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 }

}