5 of 5

Advanced

Thread safety internals, interface segregation, and once-listener guarantees.

Thread safety

Both EventBus and SuspendingEventBus are fully thread-safe. You can subscribe, unsubscribe, emit, and clear concurrently from different threads or coroutines without external synchronization.

The implementation uses synchronized blocks around all mutable state. Dispatch uses snapshot-based iteration — the listener list is copied under a lock before dispatch begins, so mutations during emit never cause ConcurrentModificationException:

// Safe: subscribe from one thread while emitting from another
thread { bus.subscribe<UserCreated, SendWelcomeEmail>() }
thread { bus.emit(UserCreated("Alice")) }

// Safe: cancel a subscription during emit
bus.on<UserCreated> { event ->
    if (event.name == "stop") subscription.cancel()
}

The middleware chain is cached and invalidated only when use() or clear() is called, avoiding per-emit allocation.

Interface segregation

The event bus is split into focused interfaces so each component receives only the capabilities it needs:

interface Emitter     // emit()
interface Subscriber  // subscribe(), unsubscribe(), on(), once(), use(), register(), clear()
interface Inspector   // hasListeners(), listenerCount()
interface EventBus : Emitter, Subscriber, Inspector

Inject only what each part of your code requires:

class OrderService(private val events: Emitter) {
    fun placeOrder(order: Order) {
        // can only emit — cannot subscribe or inspect
        events.emit(OrderPlaced(order.id))
    }
}

class EventConfigurer(private val subscriber: Subscriber) {
    fun configure() {
        // can only subscribe — cannot emit
        subscriber.subscribe<OrderPlaced, NotifyWarehouse>()
    }
}

The coroutines module follows the same pattern with SuspendingEmitter, SuspendingSubscriber, and Inspector.

Once guarantees

One-shot listeners (once()) are guaranteed to fire at most once, even under concurrent or reentrant emit. The implementation uses AtomicBoolean to ensure exactly-once semantics:

// Guaranteed to print exactly once, even if emitted concurrently
bus.once<UserCreated> { event ->
    println("First: ${event.name}")
}

// Reentrant: emitting inside a handler doesn't double-fire once listeners
bus.once<UserCreated> { event ->
    bus.emit(UserCreated("reentrant"))
}

If you cancel a once-subscription before it fires, it will never fire. If you cancel after it has already fired, the cancel is a no-op.