Skip to content

Instantly share code, notes, and snippets.

@takahirom
Last active June 8, 2024 14:09
Show Gist options
  • Save takahirom/f2dbcc3053adfd87ac7e321d95a23021 to your computer and use it in GitHub Desktop.
Save takahirom/f2dbcc3053adfd87ac7e321d95a23021 to your computer and use it in GitHub Desktop.
EventBus by Kotlin coroutine
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 }
}
}
@frel
Copy link

frel commented Jul 29, 2018

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.

@svenjacobs
Copy link

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.

@svenjacobs
Copy link

I found my own solution.

@gaerfield
Copy link

gaerfield commented Mar 13, 2020

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

@mbarkiMohamed
Copy link

mbarkiMohamed commented Jan 7, 2021

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 }

}

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment