Here’s another question from some experiment I’m doing. I’m trying to write an async EventBus. It simply registers EventListeners associated to a class, and when he receives a notify it finds the right listeners and invokes them asynchronously. Now, I was thinking that it’s a good idea to make this component being initialized with a CoroutineScope, such that someone from outside can actually decide which scope it runs into (ie I want to decide which threadpool will power the eventbus). I came out with something like below and wanted to ask your opinion if that’s good practice or am I missing something.

class AsyncEventBus(
    private val eventBusScope: CoroutineScope
) : EventBus {
    private val eventListeners: MutableMap<Class<*>, MutableList<EventListener<*>>> = mutableMapOf()

    override fun <E : Event> registerListener(aClass: Class<out E>, eventListener: EventListener<E>) {
        val listeners = retrieveListeners(aClass)
        listeners.add(eventListener)
    }

    override fun <E : Event> notify(event: E) {
        eventListeners[event::class.java]?.asSequence()
            ?.filterIsInstance<EventListener<E>>()
            ?.forEach {
                eventBusScope.launch {
                    it.handle(event)
                }
            }
    }

    @Suppress("UNCHECKED_CAST")
    private fun <E : Event> retrieveListeners(aClass: Class<out E>): MutableList<EventListener<E>> =
        eventListeners.getOrPut(aClass) { mutableListOf() } as MutableList<EventListener<E>>
}

My questions on my code (apart from whether I’m not following best practices for some reasons) are: • is there anything that can happen that I’m not considering? • I’m happy that the EventListener handle is not a suspending function since I wanted to make it transparent to who implements handle(). Of course this has the drawback that if someone start to use async blocks into the handle() function (with a different scope) it may find that code to run even though the listener has been killed. Is this something I should be scared about? or is it fair to say that if the handle() goes async, you will take care of what can happen?

will

registerListener

and

notify

ever be called concurrently? you may need to make the map thread-safe

good point, I think the design would be to never call them concurrently, but you’re right it may possibly happen. I have to make the map thread safe you’re right! thanks

here is an alternative model that I think fits more with coroutines https://gist.github.com/octylFractal/7b43cce07b135cbbd246b34e24269075

thanks a lot! When you say

the notifier also now has back-pressure though, which can be more “correct” but that depends on how your code uses the event bus

do you mean that with flows implementation the consumer of the bus can have backpressure? Can you expand it a little bit?

because the BroadcastChannel has only

BUFFERED

size, if a downstream event listener backs up the Flow (say, events are fired every 1 second but it takes 10 seconds to process an event), then eventually

notify

will block waiting for it to process

got it, thanks for the explanation, so you think in general in these cases it’s better to use flows rather than coroutines, right? Any reference you can share to study them?

Flows are a part of coroutines, https://kotlinlang.org/docs/reference/coroutines/flow.html is the basic intro docs

gotcha thanks was reading that

Would be nice if this becomes to a lightway kotlin (with coroutines & flow support) library. 👍

I don't have any other recommendations I think, I usually learn by looking at the implementation and kdocs

Resuming this thread 🙂 @octylFractal I was looking at your gist but I am a bit doubtful, what is the advantage in your mind to use flows instead of simply launch in this specific case of an event bus?

the consumer of the event is able to control the context in which their event listener is run, rather than it being forced by the event bus. this means they can choose to listen in a custom thread pool, on a "main" event thread, or directly on the IO dispatcher if needed

gotcha thanks, actually the idea behind the event bus is that you should be able to run it in a threadpool for the command bus itself. On case you need to run (for some weird reasons) on different threadpool you can still have 2 event buses, but in general that should never happen. My idea was to abstract the 'how' the bus is handling events asynchronously by giving to its consumers just a play interface, hiding all the infrastructure of flows, suspend and channels from the consumers that consume it, if that makes sense?

it makes sense, but personally I would not do it, I think coroutines tries to reduce the idea of needing other threadpools for stuff besides I/O. the

Dispatchers.Default

should have all you need for CPU-bound tasks, adding more threads is just extra threading pressure on the OS, and will probably result in very very slightly slower code, depending on how many extra threads you have in that pool, and how many extra tasks they are executing for the events

yeah the interface for EventListener is handle() with not suspending. That’s because I may have different event listeners, sync, async. My idea was to let you configure how the event listener should behave without bringing with you the information all over the places (if possible) and break your implementation that might already be there. For how much I understand your reasons, I think you are referring to which is the dispatcher used to process events and I agree with you. My doubt is more on “is that not a good practice to have a Scope injected into the bus so you can decide how the asynchronicity works (possibly with Dispathers.Default but you will be able to change it in case you need)?

yes, https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-broadcast-channel.html

when capacity positive, but less than UNLIMITED – creates ArrayBroadcastChannel with a buffer of given capacity. Note: this channel looses all items that are send to it until the first subscriber appears;

correct, that’s why I send an event using notify, then I subscribe to those event, but I don’t receive anything 😞

you sent the event first, so it was lost because nothing was subscribed

yeah sorry, my bad, I thought it was a problem of event sending, but even if I invert the order between notify and subscribe, it still hang waiting 😞

oh, right, there's another detail there -- flow doesn't subscribe until

collect

occurs

should this work?

@Test
    fun testingFlow() = runBlocking {
        val bus = EventBus()
        val flow = bus.subscribe<AsyncEvent>()

        launch {
            flow.collect {
                println(it)
            }
        }

        bus.notify(AsyncEvent("hello"))
        delay(Long.MAX_VALUE)
    }

I would doubt it, try doing

delay(100L)

first

you’re right, by doing delay it works

because flows are cold by default, so nothing is subscribed or flows until you invoke a terminal operation like

collect

gotcha, thanks for the reference, would it be easier at that stage to use a receive channel instead of a flow then?

no, I don't think it will help

got it, ok thanks for the help, I’m honestly a bit nervous of using Broadcast channels and flows like this. Apart from the fact that it seems Broadcast channels will be killed soon, I’m not 100% sure of why to use flow in this specific case. Do you believe that avoiding using BroadcastChannel but just using Channel would be better?

no, just Channel is not designed for this sort of shared state. my suggestion was just a prototype that's untested, I think that eventually SharedFlow will be what you want, but for now this solution might be lacking

got it, to have a production ready code, do you think that then is better to launch {} event listeners similarly to my first example, while SharedFlow gets released?

yea, it will work well enough

yeah I do have an interface, so I can provide more Buses depending on the use case, got it. So back on my initial design, do you think something like this a good compromise? or do you see some disadvantages that I can’t see?

class EventBus(
    private val eventBusScope: CoroutineScope
) : EventBus {
    private val eventListeners: ConcurrentHashMap<Class<*>, MutableList<EventListener<*>>> = ConcurrentHashMap()

    override fun <E : Event> registerListener(aClass: Class<out E>, eventListener: EventListener<E>) {
        val listeners = retrieveListeners(aClass)
        listeners.add(eventListener)
    }

    override fun <E : Event> notify(event: E) {
        eventListeners[event::class.java]?.asSequence()
            ?.filterIsInstance<EventListener<E>>()
            ?.forEach {
                eventBusScope.launch {
                    it.handle(event)
                }
            }
    }

    @Suppress("UNCHECKED_CAST")
    private fun <E : Event> retrieveListeners(aClass: Class<out E>): MutableList<EventListener<E>> =
        eventListeners.getOrPut(aClass) { mutableListOf() } as MutableList<EventListener<E>>
}

you may want to try / catch and log instead

you mean instead of a supervisor? just doing something like

forEach {

                  eventBusScope.launch {
                    try{
                      it.handle(event)
                    } catch {log}
                  }
            }

yes