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