LLD-016 · DESIGN PRACTICE

Pub/Sub Event Bus

Define subscriber snapshots, callback isolation and per-consumer backpressure before claiming delivery guarantees.

Local Java 21 reference verified: 17 checks. Learner drafts are not executed. Author self-review; independent review pending.

Existing introductory lesson · Version 1

Understand the problem

Design a typed in-process event bus with synchronous and buffered subscribers. Support subscribe/unsubscribe, isolated failures, bounded queues, explicit drain and close. Explain what publication acknowledges and how subscription changes during a callback affect the current event.

Before you start

• Observer and narrow callback contracts
• Type-safe topic identity
• Snapshot iteration and reentrancy
• Bounded producer/consumer queues

Scope and explicit decisions

• One in-process bus. No persistence, broker, replay log, automatic retry, distributed delivery, wildcard topics or exactly-once side-effect guarantee. Payloads must be immutable or caller-owned snapshots; generic T does not deep-copy mutable objects.
• A topic name is registered with one runtime Class. Reusing its name with another class rejects. Topic<T> gives compile-time help, while runtime registration/type checks protect the bus boundary.
• Synchronous recipients are snapshotted at publish. Subscription changes made inside a callback affect later publications. A synchronous subscriber removed after that snapshot still receives this event; a newly added one waits until a future snapshot.
• Buffered admission happens under the same registry monitor at publication. Each subscription owns a fixed-capacity FIFO queue; full queue drops that subscriber’s newest event and increments the per-publication dropped count. Other subscriptions admit independently.
• unsubscribe disables future selection and discards any still-buffered events. A callback already claimed by a drainer or included in a synchronous snapshot may finish; unsubscribe does not retract in-flight work. Subscription records are retained, so tokens remain inspectable and can be drained to empty.
• Callbacks run outside the registry monitor. Ordinary handler exceptions are counted and siblings continue. Interruptions restore the thread flag; buffered drain stops on interruption, leaving remaining work. JVM Errors are not caught as ordinary failures.
• Synchronous recursive publish is allowed up to 32 nested publish calls on one thread. The next recursive call rejects before admitting an event. This prevents unbounded direct recursion, not every possible cross-thread handler dependency deadlock.
• Buffered drain is explicitly invoked with an event-count budget and one drainer per subscription. No background thread or deadline exists. close stops new topics/subscriptions/publications, but already accepted buffered work can still drain.
• Publication sequence is allocated under the monitor. Buffered FIFO follows admission order; concurrent synchronous callback completion can differ from sequence order. Sequence is not a durable offset or acknowledgement of external business effects.

Worked examples

Try before revealing

Requirements and invariants

R1 — One topic name has one registered payload class. topic protects I1.

R2 — Current synchronous callbacks use a stable recipient snapshot outside the registry lock. publish protects I2.

R3 — Each buffered queue stays within its own capacity and has one drainer. drain protects I3.

R4 — A handler failure cannot silently skip later synchronous subscribers. publish protects I2.

I1: Registered topic name→Class identity cannot change while the bus is open.

I2: Publish selects synchronous recipients once, and invokes each outside registry ownership with isolated ordinary failures.

I3: Each buffered queue depth<=capacity, with at most one claimed callback in flight under one drainer.

Derive the design

A mutable list of callbacks needs a snapshot contract

A basic topic→list of handlers can deliver synchronously, but iterating a live mutable list breaks when a handler subscribes or unsubscribes during delivery. Copying a selected recipient set under the registry owner establishes one clear publication boundary. The reference then adds independent bounded queues rather than letting a slow handler require an unbounded global buffer.

Alternative: A mature broker adds durable offsets, acknowledgements and redelivery, but those are different contracts. Replacing this bus with a broker requires changing event identity, consumer acknowledgement and idempotent side-effect expectations; it is not merely swapping a transport class.

Responsibilities and ownership

Method contracts

What, when, why and how

Each concept is attached to the requirement it protects, its implementation, an alternative and a test.

Read and run the Java reference

1import java.util.*;
2/** In-process typed bus. Payloads must be immutable or externally owned snapshots. */
3public final class Solution {
4    private Solution() {}
5    public enum Mode { SYNC, BUFFERED }
6    public record Topic<T>(String name,Class<T> type){public Topic{if(name==null||name.isBlank())throw new IllegalArgumentException("topic");Objects.requireNonNull(type);}}
7    @FunctionalInterface public interface Handler<T>{void accept(T value)throws Exception;}
8    private record Envelope(long sequence,Object value){}
9    public record Report(long sequence,int succeeded,int failed,int queued,int dropped){}
10    private static final class Subscription {
11        final long id;final String topic;final Mode mode;final int capacity;final Handler<Object> handler;
12        final ArrayDeque<Envelope> queue=new ArrayDeque<>();boolean active=true,draining;
13        Subscription(long id,String topic,Mode mode,int capacity,Handler<Object> handler){this.id=id;this.topic=topic;this.mode=mode;this.capacity=capacity;this.handler=handler;}
14    }
15    public static final class Bus {
16        private final Map<String,Class<?>> topics=new HashMap<>();
17        private final Map<Long,Subscription> subscriptions=new LinkedHashMap<>();
18        private final ThreadLocal<Integer> depth=ThreadLocal.withInitial(()->0);
19        private long sequence,ids;private boolean closed;
20        public synchronized <T> Topic<T> topic(String name,Class<T> type){if(closed)throw new IllegalStateException("closed");var result=new Topic<>(name,type);var prior=topics.get(name);if(prior!=null&&prior!=type)throw new IllegalArgumentException("topic type mismatch");topics.put(name,type);return result;}
21        private void registered(Topic<?> topic){if(topics.get(topic.name())!=topic.type())throw new IllegalArgumentException("unregistered or mismatched topic");}
22        public synchronized <T> long subscribe(Topic<T> topic,Mode mode,int capacity,Handler<T> handler){
23            registered(topic);Objects.requireNonNull(mode);Objects.requireNonNull(handler);if(closed)throw new IllegalStateException("closed");if(mode==Mode.BUFFERED&&capacity<1)throw new IllegalArgumentException("capacity");
24            long id=++ids;subscriptions.put(id,new Subscription(id,topic.name(),mode,capacity,value->handler.accept(topic.type().cast(value))));return id;
25        }
26        public synchronized void unsubscribe(long id){var s=subscriptions.get(id);if(s==null)throw new IllegalArgumentException("subscription");s.active=false;s.queue.clear();}
27        public <T> Report publish(Topic<T> topic,T value){
28            if(depth.get()>=32)throw new IllegalStateException("reentrant publish limit");depth.set(depth.get()+1);
29            try{
30                final List<Subscription> synchronous=new ArrayList<>();final long eventSequence;int queued=0,dropped=0;
31                synchronized(this){registered(topic);Objects.requireNonNull(value);if(!topic.type().isInstance(value))throw new IllegalArgumentException("payload type");if(closed)throw new IllegalStateException("closed");eventSequence=++sequence;
32                    for(var s:subscriptions.values())if(s.active&&s.topic.equals(topic.name())){
33                        if(s.mode==Mode.SYNC)synchronous.add(s);else if(s.queue.size()==s.capacity)dropped++;else{s.queue.addLast(new Envelope(eventSequence,value));queued++;}
34                    }
35                }
36                int ok=0,failed=0;for(var s:synchronous)try{s.handler.accept(value);ok++;}catch(Exception ex){if(ex instanceof InterruptedException)Thread.currentThread().interrupt();failed++;}
37                return new Report(eventSequence,ok,failed,queued,dropped);
38            }finally{int remaining=depth.get()-1;if(remaining==0)depth.remove();else depth.set(remaining);}
39        }
40        public int drain(long id,int maxEvents){
41            if(maxEvents<1)throw new IllegalArgumentException("budget");final Subscription s;
42            synchronized(this){s=subscriptions.get(id);if(s==null||s.mode!=Mode.BUFFERED)throw new IllegalArgumentException("buffered subscription");if(s.draining)throw new IllegalStateException("already draining");s.draining=true;}
43            int failed=0;
44            try{for(int i=0;i<maxEvents;i++){final Envelope e;synchronized(this){e=s.queue.pollFirst();}if(e==null)break;try{s.handler.accept(e.value());}catch(Exception ex){failed++;if(ex instanceof InterruptedException){Thread.currentThread().interrupt();break;}}}return failed;}
45            finally{synchronized(this){s.draining=false;}}
46        }
47        public synchronized int queued(long id){var s=subscriptions.get(id);if(s==null)throw new IllegalArgumentException("subscription");return s.queue.size();}
48        public synchronized void close(){closed=true;}
49    }
50}
51

Complexity and limits

• publish scans S retained subscription records: O(S) routing plus selected synchronous handler costs. Buffered enqueue/drop is O(1) per selected subscriber.
• drain processes at most N events with O(N) queue work plus arbitrary handler latency; maxEvents bounds count, not wall-clock time.
• Space is retained registry plus per-buffer capacity×payload size, with one possible in-flight callback per drainer. Payloads are not copied or byte-bounded by the bus.
• Topic/subscription histories and tokens are not reclaimed automatically. A production lifecycle needs safe registry compaction and durable broker semantics if required.

Walk through the actual reference

These walkthrough fixtures are emitted only after the Java harness verifies their operations and expected state. Concurrency tests additionally exercise competing callers.

Mistakes and counterexamples

Engineer → Senior → Staff

Return to the lesson →

Provenance and scope of evidence

Original teaching material. The local contract is implemented by Solution.java and tested by Test.java on Java 21. Distributed persistence, real provider/device adapters and deployment are not claimed.

Reference verification is not independent editorial review, production deployment verification or a claim that distributed adapters are implemented.