Lab L15 · Java + Go

Implementing a bounded publish-subscribe component

Build an in-process broadcast broker. Every subscriber gets its own bounded mailbox, and publish(message) offers the message to each current subscriber's mailbox without ever waiting. If a subscriber's mailbox is full, that subscriber misses this message (it is dropped and counted) while everyone else still receives it. The broker also caps how many subscribers it accepts.

The task

Live driver-location updates are pushed to several listeners inside one service: a websocket pushing to the rider's map, an ETA recalculator and a metrics sampler. If the websocket client is on a bad connection, its listener falls behind. The publisher must not block, and the other listeners must not suffer. For location updates dropping a message is acceptable because a newer one follows shortly. This is not Kafka: no persistence, no acknowledgements, no replay.

One publish, three mailboxes. The slow subscriber's full mailbox rejects the new message; the publisher records the drop and moves on.

What you implement

public final class Broker<T> {
    public record Delivery(int delivered, int dropped) {}
    public Broker(int maxSubscribers, int mailboxCapacity);   // both > 0
    public Subscription subscribe();                          // IllegalStateException if closed or full
    public Delivery publish(T message);                       // never blocks
    public void close();

    public final class Subscription {
        public Optional<T> poll();          // next message or empty; never blocks
        public boolean isClosed();
        public void unsubscribe();
    }
}

Rules

publish visits each current subscriber once. Room in the mailbox → delivered. Full → this message is dropped for this subscriber; older messages stay. The return value counts both, so delivered + dropped equals the number of subscribers.
subscribe beyond maxSubscribers → IllegalStateException; unsubscribe frees a slot.
unsubscribe: the subscription receives no further messages and isClosed() becomes true. Calling it twice is harmless.
close(): idempotent; every subscription becomes closed; later publish/subscribe → IllegalStateException. Messages already in a mailbox can still be polled.
publish(null) → NullPointerException. Thread-safe: publishers and pollers may run concurrently.

Worked example

new Broker<>(2, 1); subscribe s1, s2. A third subscribe() → IllegalStateException.
publish(m1) → (2, 0). s1.poll() → m1.
publish(m2) → (1, 1): s1 had room, s2 still holds m1.
s2.poll() → m1; s2.poll() → empty — m2 was never delivered to s2.

Files and commands

Read the tests first, then replace each TODO in the starter until they pass. Java starter: Broker.java · tests: BrokerTest.java. Go starter: broker.go · tests: broker_test.go.

cd java
./gradlew test --tests 'labs.pubsub.*'

cd go
go test -race -count=1 ./pubsub

Reveal the reference solutions

Java: Broker.java
Go: broker.go

When to use

Use this lab for best-effort broadcast within one process. Each subscription owns a positive-capacity delivery queue, and the broker bounds the number of active subscriptions. A publish attempts each current subscriber once without waiting for consumers. A full queue drops the new message for that subscriber and reports the drop count. A subscriber receiving a message is not evidence that its application processed it successfully.

This is not a durable message broker. There are no acknowledgments, replay, delivery retries, persistence, topics, or exactly-once claims. Payload objects are shared, so callers must use immutable messages or provide safe ownership rules. Java rejects null messages; Go permits nil when its type permits it. Entry-count bounds do not limit payload bytes or the subscriptions retained by callers after unsubscribe.

Steps

Define admission. Reject subscriptions at the configured limit or after close. Return a subscription object rather than a writable delivery queue. In Go expose only a receive channel; in Java expose polling and a closed indicator, not queue mutation.
Serialize delivery decisions. Publish, unsubscribe, and close share one broker lock. Each accepted publish therefore has one subscriber set. Concurrent publishers have a lock-established order, preserved for the messages actually enqueued in each subscriber's FIFO.
Make slow-consumer policy visible. A subscriber with a full queue misses this newest message while others may receive it. Return delivered and dropped counts. A zero-subscriber publish succeeds with zero deliveries. Do not retry an entire partially delivered broadcast without a deduplication protocol; healthy subscribers may see duplicates.
Clarify unsubscribe. Once unsubscribe returns, future publishes cannot enqueue there. Already queued messages remain available to drain. Go closes the channel under the same lock used by sends, preventing send-on-closed races. Java marks the subscription closed but retains its queue for polling; an empty poll means no buffered message, not by itself terminal closure.
Close admission and drain. Broker close is idempotent, closes all active subscriptions, and clears the registry. It does not wait for consumers to process queued values. A caller can retain a closed subscription and its pending payloads, so owners still need a cleanup policy.
Test partial fan-out. Fill two capacity-one subscriptions, drain only one, then publish again. Exactly one receives the new value and one reports a drop. Unsubscribe the full subscription, drain its old message, verify terminal closure, close the broker, and drain the remaining queue. Add controlled publish/unsubscribe races before stronger concurrent-history claims.

Checklist

Recall and variation

Rebuild drop-newest delivery before reading Go or Java. Variation: unsubscribe a full subscriber while publishing. Acceptance: unsubscribe prevents future enqueue after returning, already queued messages still drain, and healthy subscribers retain FIFO delivery.

Subscriber and queue limits are enforced separately
A slow consumer cannot block a publisher waiting for that consumer
Delivery counts distinguish enqueued from dropped messages
Unsubscribe prevents new delivery without erasing pending messages
Closing a channel and sending to it use the same synchronization boundary

Signals

Working: a full queue affects only its subscriber and the drop outcome is observable.

Going wrong: queue fullness silently blocks all publishers, consumer processing is confused with enqueue success, or retries duplicate messages already delivered. The broker lock still serializes publishers; fan-out is O(number of active subscribers), not constant time.

Implementing a Bounded Blocking Queue, for bounded FIFO and drain semantics.