Implement an Event Bus | androidinterview.com

An event bus is a map from event type to a list of handlers, and that part takes ten minutes. Everything worth marking is around it. How a subscription gets cancelled, which thread the handler runs on, and what happens when a handler publishes something while it is being called.

Say early that you know this pattern lost the argument on Android, then build a good one anyway. The interviewer is not asking you to ship an event bus. They are asking whether you can hold a shared mutable registry safely, deliver across threads, and notice a leak before it ships.

What this really tests

Whether you can keep a registry correct under concurrent reads and writes without putting a lock on the hot path, and whether you have ever actually caused a leak. The weak answer is a HashMap of type to ArrayList with a synchronized post and no way to unsubscribe. It demos fine and it leaks the first screen that rotates.

What to clarify first

Four of these change the design and the other two change how much you write.

Does a subscriber to a supertype hear about subclasses. Say yes and say why. A logger that subscribes to the parent type and receives everything is the main reason anyone wants a bus.
Which thread does a handler run on. The posting thread, always the main thread, or whichever the subscriber asked for. The third answer is the only one that scales, and it costs one field.
Are sticky events in scope. Replaying the last event to a late subscriber is easy to add and easy to misuse, so agree on it before you build it.
Does anything depend on delivery order. If yes, you owe a defined order. If no, say so out loud, because it frees you from a queue.
What happens when a handler throws. One bad subscriber must not stop the others, and swallowing the exception silently is how a bus becomes impossible to debug.
One bus or several. A single global instance is the version that rots. Prefer an instance that gets injected, so a test can make its own.

The subscription is an object, and that is the whole leak story

Start from the failure. The bus is long lived, usually as long as the process. A handler written inside a screen captures that screen. So the moment the bus holds the handler, the bus holds the screen, and it keeps holding it after the screen is finished, rotated away and replaced. Every rotation adds another one. That is the classic event bus leak and it is why the pattern got its reputation.

There are two bad fixes. Holding subscribers weakly sounds clever and is worse, because a handler that nothing else references gets collected at a moment nobody can predict, and you get events that arrive on Tuesday and not on Wednesday. Asking the caller to hand the handler back to an unsubscribe method is the other one, and it fails as soon as somebody writes the lambda inline, because there is then no reference to hand back.

The fix that works is to return a handle. subscribe gives back a Subscription with one useful method on it, and cancelling is the caller's problem again rather than the bus's. Then take the last step and make the common case automatic. Pass a LifecycleOwner to subscribe and the bus cancels the subscription itself when that owner is destroyed. The leak stops being something you remember not to write.

LifecycleOwner and Lifecycle here are ours, about forty lines, not the Jetpack ones. The bus needs exactly two things from a screen. Tell me when you are destroyed, and tell me honestly if you already are. That second one matters more than it looks. A subscription created on a background thread while the screen is finishing would otherwise never be cancelled, so registering a destroy callback on an already destroyed owner runs it immediately.

Delivery runs on an executor the subscriber chose

The bus should not decide which thread your code runs on. The subscriber knows, and it is the only one who does. So a subscription carries an Executor and the bus does nothing more than hand it a task.

That single field is the entire threading model. DirectExecutor runs the handler on whichever thread called post, which is right for a handler that only touches its own data. MainThreadExecutor is our stand in for the Android main looper, and modelling it is easy once you say what a looper is. A looper is one thread with a queue in front of it, which is a single thread executor. Publishing from a background thread and updating a view then needs no special case anywhere in the bus.

One detail that is easy to get wrong. The main executor queues work even when it is already on the main thread. Running it inline instead would let a handler that posts during delivery jump ahead of events that were already waiting, and a reordering like that is very hard to find later.

The classes

Six types carry this, and one of them is from the standard library.

EventBus owns the registry, the sticky map and the type cache, and it is the only class with any real logic in it. Its whole job is to turn one event into a list of subscribers and hand each of them the event.
Subscription is the handle. It is an interface with unsubscribe and isActive and nothing else, because that is all a caller ever needs.
Subscriber is one registration, and it is not public. It holds the type asked for, the handler and the executor, plus an alive flag. Callers only ever see it as a Subscription.
EventHandler is the single method interface a Java caller implements with a lambda. The Kotlin has no such file, because a function type says the same thing without a declaration.
Executor is java.util.concurrent.Executor, unchanged. Writing our own would be inventing an interface the platform already got right.
Lifecycle and LifecycleOwner are the smallest model of a screen that is useful. One hook that fires on destroy, one question about whether that already happened.

Note what is not here. There is no Topic class, because the event type is the topic. There is no Message wrapper around the event, because a wrapper carrying nothing is a layer you have to explain forever.

A registration is an object that knows its own thread and can be cancelled, which is what turns the two classic event bus complaints, the leak and the surprise thread, into ordinary fields.

One publish, walked through

A background thread finishes a sign in and posts. A screen is subscribed to that exact event on the main executor, and an analytics logger is subscribed to the parent type on the direct executor.

The bus asks for the flattened types of the event, most specific first. That walk covers the concrete class, its superclasses and every interface, and it is cached per event class, so it costs one map lookup rather than a hierarchy walk on every post.
For each of those types in order, the bus reads the registry and copies the subscribers into one list. The screen comes first because it subscribed to the exact type, then the logger.
The copy is the important step. It is a snapshot, so a handler that subscribes or unsubscribes during delivery changes what the next post sees and never what this one is halfway through. The lists themselves are copy on write, so this read takes no lock.
The screen's subscriber checks its alive flag, casts the event to its own type, and hands a task to the main executor. The background thread does not run the handler, it queues it.
The logger's subscriber does the same thing with the direct executor, so its handler runs immediately, on the background thread, before post returns.
Whenever the main thread gets around to the queued task, it checks the alive flag a second time. This is not belt and braces. Between step four and here, the screen may have been destroyed and unsubscribed, and without the second check it would still be handed an event.

The order that came out of that is defined and worth stating. Within one type, registration order. Across types, most specific first. Across executors, nothing, because two executors are two threads and the bus makes no promise it cannot keep.

Sticky events, and the day they become a bug

A sticky post is stored as well as delivered, and a subscriber that arrives later gets the stored one replayed straight away. It exists for one honest reason. A screen that opens after the network went offline should not have to wait for the network to change again to find out.

That is also the shape of its misuse. Sticky is right for state, which is a value that is true until it is replaced. It is wrong for anything that reads as an instruction, because an instruction fires again. A sticky show this dialog event replays onto the next screen that subscribes, and the next, and nobody can work out where the dialog is coming from. The rule is to make the poster responsible for removing it, and to say out loud that a sticky event which is never removed is a memory leak with a schedule.

The interesting part is a race, and it is worth two sentences at the whiteboard. Registering a subscriber and reading the sticky map are two steps, and a sticky post landing between them either delivers twice or not at all. The fix is to make registration and the sticky read one operation under the same lock the sticky post uses, so a new subscriber is either already in the registry when the event is stored, or it registers afterwards and the replay hands it the new value. Delivery still happens outside that lock, because a handler is caller code and holding a lock across caller code is how deadlocks get written.

Patterns actually used

Observer, obviously, but decoupled at both ends. Plain observer has the subject holding its listeners, so the publisher knows who is listening. Here neither side knows the other and the event type is the only contract between them. That is the whole point, and it is also the whole problem, which the last section is about.
A handle instead of a registry lookup. Returning a Subscription rather than offering a matching unsubscribe is the same idea as a cancellable token, and it is what makes the lifecycle binding a three line method.
Strategy for delivery, expressed as an executor. The thread policy is a collaborator the caller supplies, not a flag inside the bus. Adding a background delivery mode is a new executor and no edit to the bus.

Three to leave out, and say why.

No singleton. A static instance is the most copied part of every event bus write up and it is the part that hurts. It makes two tests share a registry, and it means anything anywhere can post, which is the coupling problem in its purest form. Make the bus an ordinary object and let whatever wires the app decide there is one.
No annotations and no reflection. Scanning a subscriber for annotated methods buys a slightly prettier call site and costs you startup time, obscure failures when nothing matches, and a debugger that cannot find the caller. A lambda passed to a method has none of those problems.
No priorities. Every bus that adds subscriber priority ends up with code that only works because two handlers run in a particular order, and no compiler will ever tell you when that breaks.

The implementation

The design is the same in both trees. A ConcurrentHashMap of type to CopyOnWriteArrayList, one small class per registration, and an executor per subscription.

The Kotlin is where the call site gets better rather than just shorter. subscribe is inline with a reified type parameter, so a screen writes subscribe<SignedIn> { render(it) } with no class literal anywhere. The handler is a function type instead of an interface, which deletes a whole file. Executors take default arguments, DirectExecutor is an object, and the sample events are a sealed interface, which is exactly the case that makes supertype delivery worth having.

Java

com.androidinterview.eventbus.EventBus.java

package com.androidinterview.eventbus;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Deque;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;

import com.androidinterview.eventbus.lifecycle.LifecycleOwner;
import com.androidinterview.eventbus.thread.DirectExecutor;

// The bus. Three pieces of state and nothing else.
//
// subscribers is the registry, one list per event type. It is a
// ConcurrentHashMap of copy on write lists because the read to write ratio is
// extreme. Posts happen constantly, subscriptions happen when a screen opens.
// A copy on write list gives every post a stable snapshot to walk with no
// lock at all, which is also what makes a handler safe to subscribe or
// unsubscribe from inside a delivery.
//
// typeCache holds the flattened supertypes of each concrete event class, so
// the class hierarchy is walked once per event class rather than once per
// post.
//
// sticky holds the last event posted for each concrete type. It is a plain
// LinkedHashMap guarded by its own monitor, because the store and the replay
// have to be one decision and iteration order has to be stable.
public final class EventBus {

    private final Map<Class<?>, CopyOnWriteArrayList<Subscriber<?>>> subscribers = new ConcurrentHashMap<>();
    private final Map<Class<?>, List<Class<?>>> typeCache = new ConcurrentHashMap<>();
    private final Map<Class<?>, Object> sticky = new LinkedHashMap<>();

    private final Executor defaultExecutor;

    public EventBus() {
        this(DirectExecutor.INSTANCE);
    }

    public EventBus(Executor defaultExecutor) {
        this.defaultExecutor = defaultExecutor;
    }

    public <T> Subscription subscribe(Class<T> type, EventHandler<? super T> handler) {
        return subscribe(type, defaultExecutor, handler);
    }

    // The registration and the sticky replay happen under one lock, and that
    // pairing is the whole trick. Either this subscriber is already in the
    // registry when a sticky event is stored, in which case the post finds it
    // and the replay below sees the older value, or it registers afterwards
    // and the replay hands it the new one. It cannot see both and it cannot
    // miss both.
    //
    // Delivery itself happens outside the lock, because a handler is caller
    // code and holding a lock across it invites a deadlock.
    public <T> Subscription subscribe(Class<T> type, Executor on, EventHandler<? super T> handler) {
        Subscriber<T> subscriber = new Subscriber<>(this, type, on, handler);
        List<Object> replay = new ArrayList<>();
        synchronized (sticky) {
            subscribers.computeIfAbsent(type, key -> new CopyOnWriteArrayList<>()).add(subscriber);
            for (Object event : sticky.values()) {
                if (type.isInstance(event)) {
                    replay.add(event);
                }
            }
        }
        for (Object event : replay) {
            subscriber.deliver(event);
        }
        return subscriber;
    }

    public <T> Subscription subscribe(LifecycleOwner owner, Class<T> type, EventHandler<? super T> handler) {
        return subscribe(owner, type, defaultExecutor, handler);
    }

    // The overload every screen should be using. Forgetting to unsubscribe is
    // the classic event bus leak, because the bus is a long lived object and
    // a handler written inside an Activity holds that Activity. Tying the
    // subscription to the owner makes the leak impossible to write rather
    // than merely documented.
    public <T> Subscription subscribe(
            LifecycleOwner owner, Class<T> type, Executor on, EventHandler<? super T> handler) {
        Subscription subscription = subscribe(type, on, handler);
        owner.lifecycle().addOnDestroy(subscription::unsubscribe);
        return subscription;
    }

    // Collect the targets first, then deliver. Collecting produces a snapshot,
    // so a handler that subscribes or unsubscribes during delivery changes
    // what the next post sees and never what this one is halfway through.
    public void post(Object event) {
        for (Subscriber<?> subscriber : targetsOf(event)) {
            subscriber.deliver(event);
        }
    }

    // Same as post, and it also remembers the event so a subscriber that
    // arrives later is told the current answer instead of waiting for the
    // next change. Useful for state, wrong for anything that reads as a
    // one off instruction, because that instruction will fire again.
    public void postSticky(Object event) {
        List<Subscriber<?>> targets;
        synchronized (sticky) {
            sticky.put(event.getClass(), event);
            targets = targetsOf(event);
        }
        for (Subscriber<?> subscriber : targets) {
            subscriber.deliver(event);
        }
    }

    // Whoever posts a sticky event owns clearing it. A dialog request that is
    // never removed is replayed to the next screen that subscribes, and that
    // is the bug the sticky section of the answer is about.
    public void removeSticky(Class<?> type) {
        synchronized (sticky) {
            sticky.keySet().removeIf(type::isAssignableFrom);
        }
    }

    void remove(Subscriber<?> subscriber) {
        CopyOnWriteArrayList<Subscriber<?>> list = subscribers.get(subscriber.type());
        if (list != null) {
            list.remove(subscriber);
        }
    }

    // Most specific first. Subscribers to the concrete class hear about the
    // event before subscribers to its supertypes, which is the order a reader
    // expects when a specific handler and a generic logger both exist.
    private List<Subscriber<?>> targetsOf(Object event) {
        List<Subscriber<?>> targets = new ArrayList<>();
        for (Class<?> type : typeCache.computeIfAbsent(event.getClass(), EventBus::flatten)) {
            CopyOnWriteArrayList<Subscriber<?>> list = subscribers.get(type);
            if (list != null) {
                targets.addAll(list);
            }
        }
        return targets;
    }

    // A breadth first walk of the class and its interfaces, so a subscription
    // to a sealed interface receives every implementation. Object is skipped,
    // because a subscription to Object is a subscription to everything and it
    // is never what anybody meant.
    private static List<Class<?>> flatten(Class<?> eventType) {
        List<Class<?>> flat = new ArrayList<>();
        Deque<Class<?>> queue = new ArrayDeque<>();
        queue.add(eventType);
        while (!queue.isEmpty()) {
            Class<?> next = queue.poll();
            if (next == Object.class || flat.contains(next)) {
                continue;
            }
            flat.add(next);
            if (next.getSuperclass() != null) {
                queue.add(next.getSuperclass());
            }
            Collections.addAll(queue, next.getInterfaces());
        }
        return List.copyOf(flat);
    }
}

com.androidinterview.eventbus.EventHandler.java

package com.androidinterview.eventbus;

// What a subscriber actually is, one method taking one event. Keeping it a
// single method interface means a lambda works everywhere a handler is asked
// for, and it keeps the bus from ever needing to know about annotations or
// reflection.
//
// The Kotlin side has no equivalent file, because a function type says the
// same thing without a declaration.
@FunctionalInterface
public interface EventHandler<T> {
    void onEvent(T event);
}

com.androidinterview.eventbus.Subscriber.java

package com.androidinterview.eventbus;

import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;

// One registration. It knows the type it asked for, the handler to call and
// the executor to call it on. It is package private on purpose, because
// callers only ever hold it as a Subscription.
final class Subscriber<T> implements Subscription {

    private final EventBus bus;
    private final Class<T> type;
    private final Executor executor;
    private final EventHandler<? super T> handler;
    private final AtomicBoolean alive = new AtomicBoolean(true);

    Subscriber(EventBus bus, Class<T> type, Executor executor, EventHandler<? super T> handler) {
        this.bus = bus;
        this.type = type;
        this.executor = executor;
        this.handler = handler;
    }

    Class<T> type() {
        return type;
    }

    // The alive flag is read twice, and that is deliberate. Once here, so a
    // cancelled subscription is skipped at post time, and once inside the
    // task, because an executor may run the task long after unsubscribe
    // returned. Without the second check, a screen that unsubscribed in
    // onDestroy can still be handed an event on the main thread afterwards.
    void deliver(Object event) {
        if (!alive.get()) {
            return;
        }
        T typed = type.cast(event);
        executor.execute(() -> {
            if (alive.get()) {
                handler.onEvent(typed);
            }
        });
    }

    // Compare and set, so a double unsubscribe is harmless and the removal
    // from the registry happens exactly once.
    @Override
    public void unsubscribe() {
        if (alive.compareAndSet(true, false)) {
            bus.remove(this);
        }
    }

    @Override
    public boolean isActive() {
        return alive.get();
    }
}

com.androidinterview.eventbus.Subscription.java

package com.androidinterview.eventbus;

// The handle that subscribe hands back, and the reason the bus has no
// unsubscribe method that takes a type and a handler.
//
// A bus that asks you to hand back what you registered forces every caller to
// keep the type and the lambda in fields just so it can find its own
// registration again, and the moment a lambda is written inline the caller
// cannot cancel at all. One object you can cancel removes both problems.
public interface Subscription {

    void unsubscribe();

    boolean isActive();
}

com.androidinterview.eventbus.events.AppEvent.java

package com.androidinterview.eventbus.events;

// A small event hierarchy, here only to show what supertype delivery buys.
// Subscribe to SignedIn and you get one kind of event. Subscribe to AppEvent
// and you get all three, which is how an analytics logger or a crash
// breadcrumb trail is written without touching any publisher.
//
// Events are values. They carry no callbacks and no references to a screen,
// because a bus holds the last sticky one for as long as the process lives.
public sealed interface AppEvent {

    record SignedIn(String userId) implements AppEvent {
    }

    record SignedOut(String reason) implements AppEvent {
    }

    record NetworkChanged(boolean online) implements AppEvent {
    }
}

com.androidinterview.eventbus.lifecycle.Lifecycle.java

package com.androidinterview.eventbus.lifecycle;

import java.util.ArrayList;
import java.util.List;

// The smallest useful model of the Android lifecycle. The real one has a
// state machine and seven events. The bus needs two things from it, a way to
// be told when the owner is destroyed, and an honest answer to whether that
// has already happened.
public final class Lifecycle {

    private final List<Runnable> onDestroy = new ArrayList<>();
    private boolean destroyed;

    // Registering after destruction runs the action straight away. That case
    // looks unlikely and is not. A subscription made on a background thread
    // while the screen is finishing would otherwise never be cancelled, which
    // is the exact leak this class exists to stop.
    public void addOnDestroy(Runnable action) {
        synchronized (this) {
            if (!destroyed) {
                onDestroy.add(action);
                return;
            }
        }
        action.run();
    }

    // The actions run outside the lock. Unsubscribing takes the bus registry
    // locks, and holding two locks in two different orders is how a deadlock
    // gets written by accident.
    public void destroy() {
        List<Runnable> actions;
        synchronized (this) {
            if (destroyed) {
                return;
            }
            destroyed = true;
            actions = new ArrayList<>(onDestroy);
            onDestroy.clear();
        }
        for (Runnable action : actions) {
            action.run();
        }
    }

    public synchronized boolean isDestroyed() {
        return destroyed;
    }
}

com.androidinterview.eventbus.lifecycle.LifecycleOwner.java

package com.androidinterview.eventbus.lifecycle;

// Anything that can be destroyed, which on Android is an Activity, a Fragment
// or a view. The bus only ever sees this interface, so nothing in the design
// depends on the Android SDK and every test can destroy an owner on demand.
public interface LifecycleOwner {

    Lifecycle lifecycle();
}

com.androidinterview.eventbus.thread.DirectExecutor.java

package com.androidinterview.eventbus.thread;

import java.util.concurrent.Executor;

// Delivers on whichever thread called post. It is the cheapest option and the
// right one for a handler that only touches its own data.
//
// It is also where re-entrancy shows up. A handler that posts during delivery
// runs the second event all the way to completion inside the first, so the
// call stack grows and the observed order is depth first rather than the
// order the two events were posted in.
public final class DirectExecutor implements Executor {

    public static final DirectExecutor INSTANCE = new DirectExecutor();

    private DirectExecutor() {
    }

    @Override
    public void execute(Runnable task) {
        task.run();
    }
}

com.androidinterview.eventbus.thread.MainThreadExecutor.java

package com.androidinterview.eventbus.thread;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

// Stands in for the Android main looper. A looper is one thread with a queue
// in front of it, which is exactly a single thread executor, so modelling it
// this way costs one class and keeps the bus free of the SDK.
public final class MainThreadExecutor implements Executor {

    private volatile Thread thread;

    private final ExecutorService delegate = Executors.newSingleThreadExecutor(runnable -> {
        Thread created = new Thread(runnable, "main");
        created.setDaemon(true);
        thread = created;
        return created;
    });

    // Everything goes through the queue, including work posted from the main
    // thread itself. Running those inline would let a handler that posts
    // during delivery jump ahead of events that were already waiting, and a
    // reordering like that is very hard to find later.
    @Override
    public void execute(Runnable task) {
        delegate.execute(task);
    }

    public boolean isMainThread() {
        return Thread.currentThread() == thread;
    }

    public void shutdown() {
        delegate.shutdown();
    }
}

Kotlin

com.androidinterview.eventbus.EventBus.kt

package com.androidinterview.eventbus

import com.androidinterview.eventbus.lifecycle.LifecycleOwner
import com.androidinterview.eventbus.thread.DirectExecutor
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.Executor
import java.util.concurrent.atomic.AtomicBoolean

// The handle subscribe hands back, and the reason there is no unsubscribe
// that takes a type and a handler. A bus that asks you to hand back what you
// registered forces callers to keep the lambda in a field, and a lambda
// written inline could never be cancelled at all.
interface Subscription {
    val isActive: Boolean
    fun unsubscribe()
}

// One registration. Internal because callers only ever hold it as a
// Subscription. There is no EventHandler interface anywhere in the Kotlin,
// because a function type says the same thing without a declaration.
internal class Subscriber<T : Any>(
    private val bus: EventBus,
    val type: Class<T>,
    private val executor: Executor,
    private val handler: (T) -> Unit,
) : Subscription {

    private val alive = AtomicBoolean(true)

    override val isActive: Boolean get() = alive.get()

    // The alive flag is read twice on purpose. Once here, so a cancelled
    // subscription is skipped at post time, and once inside the task, because
    // an executor may run it long after unsubscribe returned. Without the
    // second check a screen that unsubscribed in onDestroy can still be
    // handed an event on the main thread afterwards.
    fun deliver(event: Any) {
        if (!alive.get()) return
        val typed = type.cast(event)
        executor.execute { if (alive.get()) handler(typed) }
    }

    // Compare and set, so a double unsubscribe is harmless and the registry
    // is touched exactly once.
    override fun unsubscribe() {
        if (alive.compareAndSet(true, false)) bus.remove(this)
    }
}

// The bus. Three pieces of state and nothing else.
//
// subscribers is the registry, one list per event type. A ConcurrentHashMap
// of copy on write lists, because posts happen constantly and subscriptions
// happen when a screen opens. Copy on write gives every post a stable
// snapshot with no lock, which is also what makes a handler safe to subscribe
// or unsubscribe from inside a delivery.
//
// typeCache holds the flattened supertypes of each concrete event class, so
// the hierarchy is walked once per class rather than once per post.
//
// sticky holds the last event of each concrete type, in a plain LinkedHashMap
// guarded by its own monitor, because the store and the replay have to be one
// decision and the iteration order has to be stable.
class EventBus(val defaultExecutor: Executor = DirectExecutor) {

    private val subscribers = ConcurrentHashMap<Class<*>, CopyOnWriteArrayList<Subscriber<*>>>()
    private val typeCache = ConcurrentHashMap<Class<*>, List<Class<*>>>()
    private val sticky = LinkedHashMap<Class<*>, Any>()

    // Reified, so the call site reads subscribe<SignedIn> { render(it) } with
    // no class literal in it. This is the one place the Kotlin is genuinely
    // nicer than the Java rather than just shorter.
    inline fun <reified T : Any> subscribe(
        on: Executor = defaultExecutor,
        noinline handler: (T) -> Unit,
    ): Subscription = subscribe(T::class.java, on, handler)

    // The overload every screen should use. Forgetting to unsubscribe is the
    // classic event bus leak, because the bus outlives every screen and a
    // lambda written inside an Activity holds that Activity. Binding the
    // subscription to the owner makes the leak impossible to write rather
    // than merely documented.
    inline fun <reified T : Any> subscribe(
        owner: LifecycleOwner,
        on: Executor = defaultExecutor,
        noinline handler: (T) -> Unit,
    ): Subscription = subscribe(owner, T::class.java, on, handler)

    // Registration and sticky replay under one lock, and that pairing is the
    // whole trick. Either this subscriber is already in the registry when a
    // sticky event is stored, so the post finds it and the replay below sees
    // the older value, or it registers afterwards and the replay hands it the
    // new one. It cannot see both and it cannot miss both.
    //
    // Delivery happens outside the lock, because a handler is caller code and
    // holding a lock across caller code invites a deadlock.
    fun <T : Any> subscribe(type: Class<T>, on: Executor, handler: (T) -> Unit): Subscription {
        val subscriber = Subscriber(this, type, on, handler)
        val replay = synchronized(sticky) {
            subscribers.getOrPut(type) { CopyOnWriteArrayList() }.add(subscriber)
            sticky.values.filter(type::isInstance)
        }
        replay.forEach(subscriber::deliver)
        return subscriber
    }

    fun <T : Any> subscribe(
        owner: LifecycleOwner,
        type: Class<T>,
        on: Executor,
        handler: (T) -> Unit,
    ): Subscription = subscribe(type, on, handler).also { owner.lifecycle.addOnDestroy(it::unsubscribe) }

    // Collect the targets, then deliver. Collecting produces a snapshot, so a
    // handler that subscribes or unsubscribes during delivery changes what the
    // next post sees and never what this one is halfway through.
    fun post(event: Any) = targetsOf(event).forEach { it.deliver(event) }

    // Post, and also remember the event so a subscriber arriving later is told
    // the current answer instead of waiting for the next change. Right for
    // state, wrong for anything that reads as a one off instruction, because
    // that instruction will fire again.
    fun postSticky(event: Any) {
        val targets = synchronized(sticky) {
            sticky[event.javaClass] = event
            targetsOf(event)
        }
        targets.forEach { it.deliver(event) }
    }

    // Whoever posts a sticky event owns clearing it. A dialog request left in
    // the map is replayed to the next screen that subscribes, which is the
    // sticky bug worth naming out loud.
    fun removeSticky(type: Class<*>) {
        synchronized(sticky) { sticky.keys.removeIf(type::isAssignableFrom) }
    }

    internal fun remove(subscriber: Subscriber<*>) {
        subscribers[subscriber.type]?.remove(subscriber)
    }

    // Most specific first, so a concrete handler hears about the event before
    // a generic logger subscribed to the sealed parent.
    private fun targetsOf(event: Any): List<Subscriber<*>> =
        typesOf(event.javaClass).flatMap { subscribers[it].orEmpty() }

    // A breadth first walk of the class and its interfaces, so a subscription
    // to a sealed interface receives every implementation. Any is skipped,
    // because subscribing to it is subscribing to everything and nobody means
    // that.
    private fun typesOf(eventType: Class<*>): List<Class<*>> = typeCache.getOrPut(eventType) {
        val flat = mutableListOf<Class<*>>()
        val queue = ArrayDeque<Class<*>>().apply { add(eventType) }
        while (queue.isNotEmpty()) {
            val next = queue.removeFirst()
            if (next == Any::class.java || next in flat) continue
            flat += next
            next.superclass?.let(queue::add)
            queue.addAll(next.interfaces)
        }
        flat
    }
}

com.androidinterview.eventbus.events.AppEvent.kt

package com.androidinterview.eventbus.events

// A small event hierarchy, here only to show what supertype delivery buys.
// Subscribe to SignedIn and you get one kind of event. Subscribe to AppEvent
// and you get all three, which is how an analytics logger or a crash
// breadcrumb trail is written without touching any publisher.
//
// Events are values. They carry no callbacks and no reference to a screen,
// because the bus holds the last sticky one for as long as the process lives.
sealed interface AppEvent {
    data class SignedIn(val userId: String) : AppEvent
    data class SignedOut(val reason: String) : AppEvent
    data class NetworkChanged(val online: Boolean) : AppEvent
}

com.androidinterview.eventbus.lifecycle.Lifecycle.kt

package com.androidinterview.eventbus.lifecycle

// The smallest useful model of the Android lifecycle. The real one has a
// state machine and seven events. The bus needs two things, a way to be told
// when the owner is destroyed and an honest answer to whether that already
// happened.
class Lifecycle {

    private val lock = Any()
    private val onDestroy = mutableListOf<() -> Unit>()
    private var destroyed = false

    val isDestroyed: Boolean get() = synchronized(lock) { destroyed }

    // Registering after destruction runs the action straight away. That case
    // looks unlikely and is not. A subscription made on a background thread
    // while the screen is finishing would otherwise never be cancelled, which
    // is the exact leak this class exists to stop.
    fun addOnDestroy(action: () -> Unit) {
        synchronized(lock) {
            if (!destroyed) {
                onDestroy += action
                return
            }
        }
        action()
    }

    // The actions run outside the lock. Unsubscribing takes the bus registry
    // locks, and holding two locks in two different orders is how a deadlock
    // gets written by accident.
    fun destroy() {
        val actions = synchronized(lock) {
            if (destroyed) return
            destroyed = true
            onDestroy.toList().also { onDestroy.clear() }
        }
        actions.forEach { it() }
    }
}

// Anything that can be destroyed, which on Android is an Activity, a Fragment
// or a view. The bus sees only this, so nothing in the design depends on the
// SDK and a test can destroy an owner whenever it likes.
interface LifecycleOwner {
    val lifecycle: Lifecycle
}

com.androidinterview.eventbus.thread.Executors.kt

package com.androidinterview.eventbus.thread

import java.util.concurrent.Executor
import java.util.concurrent.Executors

// Stands in for the Android main looper. A looper is one thread with a queue
// in front of it, which is exactly a single thread executor, so modelling it
// this way costs a few lines and keeps the bus free of the SDK.
class MainThreadExecutor : Executor {

    @Volatile
    private var thread: Thread? = null

    private val delegate = Executors.newSingleThreadExecutor { runnable ->
        Thread(runnable, "main").also {
            it.isDaemon = true
            thread = it
        }
    }

    val isMainThread: Boolean get() = Thread.currentThread() === thread

    // Everything goes through the queue, including work posted from the main
    // thread itself. Running those inline would let a handler that posts
    // during delivery jump ahead of events already waiting, and a reordering
    // like that is very hard to find later.
    override fun execute(task: Runnable) = delegate.execute(task)

    fun shutdown() = delegate.shutdown()
}

// Delivers on whichever thread called post. Cheapest option and the right one
// for a handler that only touches its own data.
//
// It is also where re-entrancy shows up. A handler that posts during delivery
// runs the second event to completion inside the first, so the stack grows
// and the observed order is depth first rather than post order.
object DirectExecutor : Executor {
    override fun execute(task: Runnable) = task.run()
}

Concurrency and edge cases

Why copy on write and not a lock. The read to write ratio here is extreme. Posts happen constantly, subscriptions happen when a screen opens. A copy on write list makes every post a lock free read of a stable array, and pays for it on subscribe, which is the rare operation. It also makes re-entrancy safe for free, because a handler that subscribes during delivery cannot invalidate the array the post is already walking.

A handler that publishes during delivery. This is the question behind the snapshot. On the direct executor the second event runs to completion inside the first, so the stack grows and the observed order is depth first rather than post order. On the main executor the second event goes to the back of the queue, so the first finishes and then the second runs. Neither is wrong, but they are different, and a bus that cannot tell you which one it does is a bus nobody can debug. If a handler can trigger itself, the recursion is unbounded and the direct executor will overflow the stack, so name that and say a queue is what stops it.

Unsubscribing during delivery. Removal from the list happens on the next post, and the alive flag handles this one. It is checked when the task is created and again when it runs, which is what makes unsubscribe mean something even for a delivery already sitting in a queue.

A handler that throws. One bad subscriber must not stop the rest, so catch it per delivery and report it somewhere that is not a swallowed exception. On the main executor this is already the case, because each delivery is a separate task, and that difference between the two executors is worth noticing rather than relying on.

The rest, a sentence each.

Subscribing twice with the same handler. Two registrations, two deliveries. That is honest, since the bus cannot compare two lambdas usefully.
Posting an event with no subscribers. Nothing happens and nothing is logged. That silence is the real cost of the pattern, because a publisher can never tell whether anyone is listening.
Unsubscribing twice. A compare and set on the alive flag, so the second call does nothing.
A subscriber that outlives the bus. It cannot, since the bus holds the subscriber and not the other way round.
An event that holds a view. Never do it, especially not a sticky one, because the bus keeps the last sticky event for the life of the process. Events are values.

Why nobody reaches for this any more

Be direct about it, because the interviewer is waiting for it. A bus makes every publisher and every subscriber reachable from everywhere, so tracing a flow means searching the whole codebase for a type rather than following a call. Nothing tells you who listens, deleting a subscriber is silent, and adding one is invisible to the person whose feature just changed behaviour. It scales beautifully until about the fourth developer.

What replaced it on Android is narrower on purpose. A repository exposes a StateFlow and the screens that care collect it, which keeps the same decoupling but names the producer, so you can navigate to it. For one off events a SharedFlow on a ViewModel does the same job for a known consumer. Both are lifecycle aware by construction, which removes the leak the bus spent a decade being blamed for.

A bus is still the right call in two places. When modules genuinely must not know each other, a feature module and the app shell that hosts it, an event type in a shared module is a cleaner contract than a direct dependency. And when the set of listeners is open, a plugin system or an in process analytics fan out, where the publisher cannot name its consumers because it is not allowed to.