Build an In-Memory Event Bus in Java
Design a thread-safe event bus library with Java 21 virtual threads and Maven.
Introduction
30 Second Summary
Every modern application has components that need to talk to each other. When a user places an order, the inventory system needs to know. When a payment succeeds, the shipping module needs to react. The pattern behind this communication is called an event bus, and building one that handles thousands of concurrent events without crashing is a senior-level engineering challenge.
In this project, you will build a production-grade, thread-safe in-memory event bus library in Java 21. Your library will support synchronous and asynchronous event dispatching using virtual threads, bounded dead-letter queues for graceful failure routing, and Semaphore-based backpressure throttling to prevent resource starvation.
What You'll Build
A lightweight, type-safe event bus library packaged as a Maven JAR that implements the Observer Pattern with defensive concurrency design and clean API abstractions.
By the end of this project, you'll have:
- A thread-safe event bus that dispatches events synchronously and asynchronously using Java 21 virtual threads.
- A production-defensive architecture with bounded dead-letter queues, backpressure throttling, and graceful error routing.
- A comprehensive JUnit 5 test suite proving thread safety, filtering, and failure modes across 100 concurrent virtual threads.
- Secret Mission: Publish your library to GitHub Packages as a real Maven registry artifact that other projects can consume.
Are there any prerequisites?
This project assumes intermediate-to-advanced Java knowledge, including generics, interfaces, and concurrency basics. You should have JDK 21 and Maven already installed on your machine.
Set Up the Maven Library Project
Every library starts with a solid project skeleton. Before you write a single line of event bus logic, you need a directory layout that Maven recognizes and a build file that tells Maven how to compile your code against JDK 21.
In this step, you will create the project folder with Maven's conventional source tree, write a pom.xml that configures JAR packaging and JUnit 5 test dependencies, and verify the project compiles cleanly.
In this step, get ready to:
- Create the eventbus project directory with the full Maven source tree.
- Write a pom.xml with library-appropriate configuration and metadata.
- Verify the project compiles against Java 21 with no errors.
Create the project directory structure
Maven expects source code under src/main/java/ and tests under src/test/java/, with subdirectories matching your Java package name. Since this is a library with no main class, you will create the tree manually instead of using an archetype scaffold.
- Move to your Desktop by running this command:
cd ~/Desktop- Create the eventbus project folder and the full Maven source tree by running these commands:
macOS
mkdir eventbus
cd eventbus
mkdir -p src/main/java/io/github/dhruv0306/eventbus
mkdir -p src/test/java/io/github/dhruv0306/eventbusWindows
mkdir eventbus
cd eventbus
mkdir src/main/java/io/github/dhruv0306/eventbus
mkdir src/test/java/io/github/dhruv0306/eventbusWhy no -p flag on Windows?
PowerShell's mkdir automatically creates all parent directories in the path. On macOS and Linux, mkdir -p is required to achieve the same behavior.
- Confirm the directories exist by running:
ls src/main/java/io/github/dhruv0306/eventbusThe command should complete with no error and no output (the directory is empty but exists).
Seeing "No such file or directory"?
- Make sure you ran cd eventbus before creating the source tree. The mkdir commands use relative paths from inside the project root.
- Check for typos in the package path. Every segment (io, github, dhruv0306, eventbus) must match exactly.
- Help me troubleshoot my directory structure.
Write the Maven build file
The pom.xml is Maven's project descriptor. It defines your library's coordinates, compilation target, test framework, and plugin versions.
- In your IDE, open the eventbus folder you just created on your Desktop.
- Create a new file called pom.xml in the root of the eventbus folder.
- Define your library's build configuration by pasting this into pom.xml:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>io.github.dhruv0306</groupId>
<artifactId>eventbus</artifactId>
<version>1.0.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>EventBus</name>
<description>A lightweight, thread-safe in-memory event bus for Java 21 with virtual thread async dispatching, bounded dead-letter queue, and event filtering.</description>
<url>https://github.com/Dhruv0306/eventbus</url>
<developers>
<developer>
<id>dhruv0306</id>
<name>Dhruv Patel</name>
</developer>
</developers>
<scm>
<connection>scm:git:git://github.com/Dhruv0306/eventbus.git</connection>
<developerConnection>scm:git:ssh://github.com:Dhruv0306/eventbus.git</developerConnection>
<url>https://github.com/Dhruv0306/eventbus</url>
</scm>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.release>21</maven.compiler.release>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.junit</groupId>
<artifactId>junit-bom</artifactId>
<version>5.14.4</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.15.0</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.5.6</version>
</plugin>
</plugins>
</pluginManagement>
</build>
</project>What does this configuration do?
- groupId, artifactId, and version are Maven coordinates that uniquely identify your library in any repository.
- packaging set to jar tells Maven to produce a library JAR (no executable main class needed).
- maven.compiler.release set to 21 compiles your code against the Java 21 API surface and bytecode level.
- The junit-bom import manages JUnit 5 version 5.14.4 in one place, so individual JUnit dependencies omit the version tag.
- Plugin version pinning in pluginManagement ensures reproducible builds regardless of the Maven installation's bundled defaults.
- Save pom.xml.
- You will verify this file works in the next substep when you run Maven compilation.
IDE showing XML errors?
- Make sure the file is saved directly inside the eventbus root folder, not inside src/.
- If your IDE flags the JUnit BOM as unresolved, that is expected until Maven downloads it during the first compile.
- Help me fix my pom.xml errors.
✔️ Awesome, I've got everything!
Great. Double check you have saved pom.xml in the root of the eventbus folder.
ⓧ I'd like to double check the full code
Here is the complete pom.xml for comparison:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>io.github.dhruv0306</groupId>
<artifactId>eventbus</artifactId>
<version>1.0.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>EventBus</name>
<description>A lightweight, thread-safe in-memory event bus for Java 21 with virtual thread async dispatching, bounded dead-letter queue, and event filtering.</description>
<url>https://github.com/Dhruv0306/eventbus</url>
<developers>
<developer>
<id>dhruv0306</id>
<name>Dhruv Patel</name>
</developer>
</developers>
<scm>
<connection>scm:git:git://github.com/Dhruv0306/eventbus.git</connection>
<developerConnection>scm:git:ssh://github.com:Dhruv0306/eventbus.git</developerConnection>
<url>https://github.com/Dhruv0306/eventbus</url>
</scm>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.release>21</maven.compiler.release>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.junit</groupId>
<artifactId>junit-bom</artifactId>
<version>5.14.4</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.15.0</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.5.6</version>
</plugin>
</plugins>
</pluginManagement>
</build>
</project>Verify the project compiles
Running mvn compile triggers Maven to resolve dependencies, download plugins, and compile any source files. Even with no Java files yet, a successful compile proves your POM is valid and Java 21 is targeted correctly.
- Make sure your terminal is inside the eventbus folder, then run the Maven compile lifecycle by running this command:
mvn compile- You should see Maven download dependencies on the first run, then print BUILD SUCCESS at the bottom of the output.
- Confirm the output includes maven-compiler-plugin:3.15.0:compile to verify the correct plugin version is in use.
What should I see?
Maven prints a log of plugin executions and ends with a summary block. Look for BUILD SUCCESS in green (if your terminal supports color) and a total time line.
A target/ directory also appears in your project root after compilation.
Seeing BUILD FAILURE?
- Check that pom.xml is in the eventbus root, not inside src/. Maven looks for pom.xml in the current working directory.
- If the error mentions an unsupported release value, run java --version and confirm it shows 21. Maven uses whichever JDK your JAVA_HOME points to.
- If you see a network error during dependency download, check your internet connection and try running mvn compile again.
- Help me fix my Maven build failure.
Your project skeleton is ready and Maven can compile it cleanly. Next up, you will define the public API interfaces that consumers of your event bus will program against.
Define the Public API
Your Maven project compiles cleanly, but the src/main/java directory is empty. Before writing any implementation logic, you need to define the contracts that consumers of your library will program against.
This step establishes the complete public API surface. You will create interfaces for event dispatching, subscription management, and failure routing. You will also define a builder-style configuration object and a custom exception for backpressure overflow.
In this step, get ready to:
- Create the core interfaces: Event, EventBus, Subscription, and DeadLetterQueue.
- Create the configuration and exception types: EventBusConfig and EventDroppedException.
- Verify the entire API surface compiles with Maven.
Create the core interfaces and types
A well-designed library separates its public API (what consumers see) from its implementation (how it works internally). You will define four interfaces that form the contract of your event bus.
Why define interfaces before implementation?
Coding to interfaces lets consumers depend on the contract without coupling to a specific implementation. If you later swap the concurrency strategy or change the internal data structures, consumers do not need to recompile.
This is the Observer Pattern in action. Publishers produce events and subscribers react to them, all mediated through the bus interface.
- Create Event.java in src/main/java/io/github/dhruv0306/eventbus/ by pasting the following code:
package io.github.dhruv0306.eventbus;
/**
* Marker interface for all events dispatched through the EventBus.
* This interface is intentionally NOT sealed to allow consumers
* of the library to define their own custom event types.
*/
public interface Event {
}- In the same package directory, create Subscription.java by pasting this code:
package io.github.dhruv0306.eventbus;
/**
* Handle returned from EventBus.subscribe() that allows the caller
* to unsubscribe and check subscription status.
*/
public interface Subscription {
void unsubscribe();
boolean isActive();
}What do these two files do?
Event.java is a marker interface. Any class that implements Event becomes publishable through the bus. It is intentionally not sealed so that consumers of your library can define their own event types.
Subscription.java is the handle returned when a consumer subscribes. Calling unsubscribe() removes the handler from the bus. isActive() lets callers check whether the subscription is still live.
- Create EventBus.java in the same package directory by pasting this code:
package io.github.dhruv0306.eventbus;
import java.util.function.Consumer;
import java.util.function.Predicate;
/**
* Public API for the in-memory event bus.
* Supports synchronous and asynchronous dispatching with type-safe subscriptions.
*/
public interface EventBus extends AutoCloseable {
<T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler);
<T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler, Predicate<T> filter);
<T extends Event> void publish(T event);
<T extends Event> void publishAsync(T event);
@Override
void close();
}What does this interface define?
- subscribe() registers a handler for a specific event type. The overload with a Predicate<T> filter lets consumers selectively process only matching events.
- publish() dispatches synchronously on the caller's thread. publishAsync() dispatches on virtual threads for non-blocking delivery.
- The interface extends AutoCloseable so consumers can use try-with-resources to guarantee clean shutdown of the internal executor.
- The generic bound <T extends Event> enforces type safety at compile time. Only classes implementing your Event marker can be published.
- Create DeadLetterQueue.java in the same package directory by pasting this code:
package io.github.dhruv0306.eventbus;
import java.util.List;
/**
* Interface for a dead-letter queue that captures undeliverable events
* or events whose handlers threw exceptions.
*/
public interface DeadLetterQueue {
boolean offer(Event event, Throwable reason);
List<DeadLetter> drain();
int size();
boolean isFull();
record DeadLetter(Event event, Throwable reason, long timestamp) {}
}What is a dead-letter queue?
A dead-letter queue (DLQ) captures events that could not be delivered. This happens when no subscribers are registered for an event type, or when a subscriber's handler throws an exception.
offer() returns a boolean instead of throwing. This lets the caller decide what to do when the queue is full. The nested DeadLetter record bundles the failed event, the reason for failure, and a timestamp for debugging.
- Save all four files: Event.java, Subscription.java, EventBus.java, and DeadLetterQueue.java.
You will not compile yet. The next substep adds two more files that complete the API surface, then you will verify everything together.
Create the configuration and exception types
Production libraries need configurable resource limits. You will create a builder-style config object that controls backpressure and DLQ capacity, plus a custom exception for overflow situations.
- Create EventBusConfig.java in the same package directory by pasting this code:
package io.github.dhruv0306.eventbus;
/**
* Configuration for the ConcurrentEventBus, built via the Builder pattern.
* Controls backpressure limits and dead-letter queue capacity.
*/
public final class EventBusConfig {
private final int maxConcurrentDispatches;
private final int deadLetterQueueCapacity;
private EventBusConfig(Builder builder) {
this.maxConcurrentDispatches = builder.maxConcurrentDispatches;
this.deadLetterQueueCapacity = builder.deadLetterQueueCapacity;
}
public int maxConcurrentDispatches() {
return maxConcurrentDispatches;
}
public int deadLetterQueueCapacity() {
return deadLetterQueueCapacity;
}
public static Builder builder() {
return new Builder();
}
public static EventBusConfig defaults() {
return new Builder().build();
}
public static final class Builder {
private int maxConcurrentDispatches = 256;
private int deadLetterQueueCapacity = 1024;
private Builder() {}
public Builder maxConcurrentDispatches(int maxConcurrentDispatches) {
if (maxConcurrentDispatches <= 0) {
throw new IllegalArgumentException("maxConcurrentDispatches must be positive");
}
this.maxConcurrentDispatches = maxConcurrentDispatches;
return this;
}
public Builder deadLetterQueueCapacity(int deadLetterQueueCapacity) {
if (deadLetterQueueCapacity <= 0) {
throw new IllegalArgumentException("deadLetterQueueCapacity must be positive");
}
this.deadLetterQueueCapacity = deadLetterQueueCapacity;
return this;
}
public EventBusConfig build() {
return new EventBusConfig(this);
}
}
}What does this configuration control?
- maxConcurrentDispatches (default 256) limits how many async handlers can run concurrently. This is the backpressure knob that prevents resource starvation.
- deadLetterQueueCapacity (default 1024) bounds the DLQ size. A bounded queue prevents unbounded memory growth under sustained failure.
- The Builder pattern gives consumers a fluent API while keeping EventBusConfig immutable after construction. Input validation in each setter rejects invalid values early.
- Create EventDroppedException.java in the same package directory by pasting this code:
package io.github.dhruv0306.eventbus;
/**
* Thrown when an event cannot be routed (no subscribers) and the dead-letter queue
* is at full capacity. Indicates the system is under sustained pressure.
*/
public class EventDroppedException extends RuntimeException {
private final Event droppedEvent;
public EventDroppedException(Event droppedEvent, String message) {
super(message);
this.droppedEvent = droppedEvent;
}
public Event getDroppedEvent() {
return droppedEvent;
}
}Why a custom exception?
EventDroppedException extends RuntimeException (unchecked). It fires when a publish fails AND the DLQ is full. This signals sustained system pressure to the caller. The droppedEvent field lets callers inspect which event was lost.
- Save both EventBusConfig.java and EventDroppedException.java.
Verify all interfaces compile
All six source files are in place. Running the compiler now validates that your API surface is syntactically correct and that all cross-references between the types resolve.
- Compile the project by running this command from the eventbus directory:
mvn compile- Confirm the output ends with BUILD SUCCESS and shows no errors or warnings related to generics or missing types.
Seeing compilation errors?
- Check that every file starts with package io.github.dhruv0306.eventbus; and that the directory path matches exactly: src/main/java/io/github/dhruv0306/eventbus/.
- If you see "cannot find symbol" for Event or Subscription, confirm that Event.java and Subscription.java exist in the same directory as EventBus.java.
- Verify your JDK version is 21 by running java --version. The record keyword inside DeadLetterQueue.java requires Java 16+.
Help me debug my Maven compilation errors for the eventbus project
✔️ Awesome, I've got everything!
Great. Double check you have saved all six files before moving on.
ⓧ I'd like to double check the full code
Here are all six source files for comparison. Make sure each file matches exactly.
package io.github.dhruv0306.eventbus;
/**
* Marker interface for all events dispatched through the EventBus.
* This interface is intentionally NOT sealed to allow consumers
* of the library to define their own custom event types.
*/
public interface Event {
}package io.github.dhruv0306.eventbus;
/**
* Handle returned from EventBus.subscribe() that allows the caller
* to unsubscribe and check subscription status.
*/
public interface Subscription {
void unsubscribe();
boolean isActive();
}package io.github.dhruv0306.eventbus;
import java.util.function.Consumer;
import java.util.function.Predicate;
/**
* Public API for the in-memory event bus.
* Supports synchronous and asynchronous dispatching with type-safe subscriptions.
*/
public interface EventBus extends AutoCloseable {
<T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler);
<T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler, Predicate<T> filter);
<T extends Event> void publish(T event);
<T extends Event> void publishAsync(T event);
@Override
void close();
}package io.github.dhruv0306.eventbus;
import java.util.List;
/**
* Interface for a dead-letter queue that captures undeliverable events
* or events whose handlers threw exceptions.
*/
public interface DeadLetterQueue {
boolean offer(Event event, Throwable reason);
List<DeadLetter> drain();
int size();
boolean isFull();
record DeadLetter(Event event, Throwable reason, long timestamp) {}
}package io.github.dhruv0306.eventbus;
/**
* Configuration for the ConcurrentEventBus, built via the Builder pattern.
* Controls backpressure limits and dead-letter queue capacity.
*/
public final class EventBusConfig {
private final int maxConcurrentDispatches;
private final int deadLetterQueueCapacity;
private EventBusConfig(Builder builder) {
this.maxConcurrentDispatches = builder.maxConcurrentDispatches;
this.deadLetterQueueCapacity = builder.deadLetterQueueCapacity;
}
public int maxConcurrentDispatches() {
return maxConcurrentDispatches;
}
public int deadLetterQueueCapacity() {
return deadLetterQueueCapacity;
}
public static Builder builder() {
return new Builder();
}
public static EventBusConfig defaults() {
return new Builder().build();
}
public static final class Builder {
private int maxConcurrentDispatches = 256;
private int deadLetterQueueCapacity = 1024;
private Builder() {}
public Builder maxConcurrentDispatches(int maxConcurrentDispatches) {
if (maxConcurrentDispatches <= 0) {
throw new IllegalArgumentException("maxConcurrentDispatches must be positive");
}
this.maxConcurrentDispatches = maxConcurrentDispatches;
return this;
}
public Builder deadLetterQueueCapacity(int deadLetterQueueCapacity) {
if (deadLetterQueueCapacity <= 0) {
throw new IllegalArgumentException("deadLetterQueueCapacity must be positive");
}
this.deadLetterQueueCapacity = deadLetterQueueCapacity;
return this;
}
public EventBusConfig build() {
return new EventBusConfig(this);
}
}
}package io.github.dhruv0306.eventbus;
/**
* Thrown when an event cannot be routed (no subscribers) and the dead-letter queue
* is at full capacity. Indicates the system is under sustained pressure.
*/
public class EventDroppedException extends RuntimeException {
private final Event droppedEvent;
public EventDroppedException(Event droppedEvent, String message) {
super(message);
this.droppedEvent = droppedEvent;
}
public Event getDroppedEvent() {
return droppedEvent;
}
}Your API surface is locked in and compiling cleanly. Next up, you will write the concurrent implementation that brings these interfaces to life with virtual threads, semaphore-based backpressure, and bounded failure routing.
Implement the Concurrent Event Bus
Your API surface compiles cleanly. The interfaces define what the event bus can do, but there is no engine behind the contracts yet.
In this step, you'll implement two classes that bring the event bus to life. The first is a bounded dead-letter queue that catches undeliverable events without growing unbounded in memory. The second is the full concurrent event bus using Java 21 virtual threads and Semaphore-based backpressure.
In this step, get ready to:
- Implement the bounded dead-letter queue with ArrayBlockingQueue.
- Implement the concurrent event bus with virtual threads and backpressure.
- Verify the full implementation compiles cleanly.
Implement the bounded dead-letter queue
Events that have no subscribers or whose handlers throw exceptions need somewhere to go. An unbounded list would eventually consume all available memory under sustained failure. A bounded queue gives the library a fixed ceiling.
- Create a new file called BoundedDeadLetterQueue.java in src/main/java/io/github/dhruv0306/eventbus/.
- Paste the following implementation into the file:
package io.github.dhruv0306.eventbus;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
// Bounded DLQ backed by ArrayBlockingQueue - returns false on overflow instead of blocking
public final class BoundedDeadLetterQueue implements DeadLetterQueue {
private final ArrayBlockingQueue<DeadLetter> queue;
public BoundedDeadLetterQueue(int capacity) {
if (capacity <= 0) {
throw new IllegalArgumentException("capacity must be positive");
}
// Fixed-size queue enforces the memory ceiling
this.queue = new ArrayBlockingQueue<>(capacity);
}
@Override
public boolean offer(Event event, Throwable reason) {
// Wrap the event with metadata (reason + timestamp) for diagnostics
DeadLetter deadLetter = new DeadLetter(event, reason, System.currentTimeMillis());
// Non-blocking offer: returns false if full, letting the caller decide policy
return queue.offer(deadLetter);
}
@Override
public List<DeadLetter> drain() {
// drainTo is atomic - moves all elements in one operation
List<DeadLetter> drained = new ArrayList<>();
queue.drainTo(drained);
return drained;
}
@Override
public int size() {
return queue.size();
}
@Override
public boolean isFull() {
return queue.remainingCapacity() == 0;
}
}What does this code do?
This class wraps an ArrayBlockingQueue with a fixed capacity you configure at construction time. The capacity is your memory safety ceiling.
- offer() is non-blocking. When the queue is full, it returns false instead of throwing or waiting. This lets the caller (the event bus) decide what to do next.
- drain() uses drainTo() for atomic bulk retrieval. All dead letters are moved out in one operation, so no consumer sees a partial batch.
- isFull() checks remainingCapacity() == 0 because ArrayBlockingQueue tracks capacity internally.
- Save BoundedDeadLetterQueue.java.
Why ArrayBlockingQueue instead of an unbounded list?
In production, a sustained failure (e.g. a downstream service is down) can generate thousands of dead letters per second. An unbounded collection would silently consume all heap until the JVM runs out of memory. A bounded queue makes the failure visible through the offer() return value, letting the system fail fast instead of fail catastrophically.
Implement the concurrent event bus
This is the core of the library. The ConcurrentEventBus ties together virtual threads for async dispatch, a ConcurrentHashMap for lock-free subscriber registration, and a Semaphore to prevent resource starvation.
- Create a new file called ConcurrentEventBus.java in src/main/java/io/github/dhruv0306/eventbus/.
- Paste the following implementation into the file:
package io.github.dhruv0306.eventbus;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.logging.Level;
import java.util.logging.Logger;
public final class ConcurrentEventBus implements EventBus {
private static final Logger LOGGER = Logger.getLogger(ConcurrentEventBus.class.getName());
// Subscriber registry: event type -> set of filtered subscribers
private final ConcurrentHashMap<Class<? extends Event>, Set<FilteredSubscriber<?>>> subscribers;
// Virtual thread executor - one new VT per async task, unbounded creation
private final ExecutorService asyncExecutor;
// Global backpressure throttle - limits concurrent handler executions
private final Semaphore asyncThrottle;
private final DeadLetterQueue deadLetterQueue;
// Ensures close() is idempotent and publish() rejects after shutdown
private final AtomicBoolean closed;
public ConcurrentEventBus(EventBusConfig config) {
this.subscribers = new ConcurrentHashMap<>();
this.asyncExecutor = Executors.newVirtualThreadPerTaskExecutor();
this.asyncThrottle = new Semaphore(config.maxConcurrentDispatches());
this.deadLetterQueue = new BoundedDeadLetterQueue(config.deadLetterQueueCapacity());
this.closed = new AtomicBoolean(false);
}
public ConcurrentEventBus() {
this(EventBusConfig.defaults());
}
@Override
public <T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler) {
return subscribe(eventType, handler, event -> true);
}
@Override
public <T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler, Predicate<T> filter) {
if (closed.get()) {
throw new IllegalStateException("EventBus is closed");
}
FilteredSubscriber<T> subscriber = new FilteredSubscriber<>(eventType, handler, filter);
// computeIfAbsent is atomic - safe for concurrent registration
Set<FilteredSubscriber<?>> eventSubscribers = subscribers.computeIfAbsent(
eventType, k -> ConcurrentHashMap.newKeySet()
);
eventSubscribers.add(subscriber);
// Return an anonymous Subscription that removes this subscriber on unsubscribe
return new Subscription() {
private final AtomicBoolean active = new AtomicBoolean(true);
@Override
public void unsubscribe() {
if (active.compareAndSet(true, false)) {
eventSubscribers.remove(subscriber);
}
}
@Override
public boolean isActive() {
return active.get();
}
};
}
@Override
@SuppressWarnings("unchecked")
public <T extends Event> void publish(T event) {
if (closed.get()) {
throw new IllegalStateException("EventBus is closed");
}
Set<FilteredSubscriber<?>> eventSubscribers = subscribers.get(event.getClass());
// No subscribers - route to DLQ
if (eventSubscribers == null || eventSubscribers.isEmpty()) {
boolean accepted = deadLetterQueue.offer(event, new NoSubscriberException(event));
if (!accepted) {
throw new EventDroppedException(event,
"Dead letter queue is full, event dropped: " + event.getClass().getName());
}
return;
}
// Synchronous dispatch: deliver to each matching subscriber on the caller's thread
for (FilteredSubscriber<?> subscriber : eventSubscribers) {
FilteredSubscriber<T> typed = (FilteredSubscriber<T>) subscriber;
if (typed.filter().test(event)) {
try {
typed.handler().accept(event);
} catch (Exception e) {
deadLetterQueue.offer(event, e);
}
}
}
}
@Override
@SuppressWarnings("unchecked")
public <T extends Event> void publishAsync(T event) {
if (closed.get()) {
throw new IllegalStateException("EventBus is closed");
}
Set<FilteredSubscriber<?>> eventSubscribers = subscribers.get(event.getClass());
if (eventSubscribers == null || eventSubscribers.isEmpty()) {
boolean accepted = deadLetterQueue.offer(event, new NoSubscriberException(event));
if (!accepted) {
throw new EventDroppedException(event,
"Dead letter queue is full, event dropped: " + event.getClass().getName());
}
return;
}
for (FilteredSubscriber<?> subscriber : eventSubscribers) {
FilteredSubscriber<T> typed = (FilteredSubscriber<T>) subscriber;
if (typed.filter().test(event)) {
// Each matching subscriber gets its own virtual thread
asyncExecutor.submit(() -> {
try {
// Acquire permit INSIDE the VT - parks without pinning the carrier
asyncThrottle.acquire();
try {
typed.handler().accept(event);
} finally {
// Release in finally prevents permit leaks on handler exceptions
asyncThrottle.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOGGER.log(Level.WARNING, "Async dispatch interrupted for event: " + event, e);
} catch (Exception e) {
deadLetterQueue.offer(event, e);
LOGGER.log(Level.WARNING, "Subscriber threw exception for event: " + event, e);
}
});
}
}
}
public DeadLetterQueue deadLetterQueue() {
return deadLetterQueue;
}
@Override
public void close() {
if (closed.compareAndSet(false, true)) {
// Ordered shutdown: stop accepting -> drain in-flight -> force-stop -> clear
asyncExecutor.shutdown();
try {
if (!asyncExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
asyncExecutor.shutdownNow();
}
} catch (InterruptedException e) {
asyncExecutor.shutdownNow();
Thread.currentThread().interrupt();
}
subscribers.clear();
}
}
// Record gives us component-wise equals - each lambda registration is unique
private record FilteredSubscriber<T extends Event>(
Class<T> eventType,
Consumer<T> handler,
Predicate<T> filter
) {}
private static class NoSubscriberException extends RuntimeException {
NoSubscriberException(Event event) {
super("No subscribers registered for event type: " + event.getClass().getName());
}
}
}What does this code do?
This is the production core of your library. Here's how the key pieces fit together:
- ConcurrentHashMap with ConcurrentHashMap.newKeySet() gives O(1) subscriber registration and removal with lock-striping. No full-array copies on every mutation.
- Executors.newVirtualThreadPerTaskExecutor() creates a new virtual thread for each async task. Virtual threads are lightweight (millions possible), so no pooling is needed.
- The Semaphore is acquired inside the virtual thread. When all permits are taken, additional VTs park without pinning their carrier OS threads. This is zero-overhead backpressure.
- The close() method follows ordered shutdown: stop accepting new tasks, wait up to 10 seconds for in-flight tasks, force-stop stragglers, then clear subscribers.
- @SuppressWarnings("unchecked") covers the type-safe cast from the heterogeneous subscriber map. The cast is safe because subscribe() guarantees the type match at registration time.
- Save ConcurrentEventBus.java.
Why virtual threads instead of a fixed thread pool?
Virtual threads are the idiomatic Java 21 approach to concurrent I/O and async work. They are managed by the JDK scheduler, not the OS. Creating millions of them is cheap because they share a small pool of carrier OS threads underneath.
A fixed thread pool requires you to guess how many threads are "enough." Too few starves throughput. Too many wastes memory. Virtual threads remove that tradeoff entirely. The Semaphore handles the throughput ceiling instead.
Seeing a compile error about unchecked casts?
- Make sure both @SuppressWarnings("unchecked") annotations are placed directly above the publish() and publishAsync() methods, not on the class.
- Verify the FilteredSubscriber record is declared inside the class body (it's a private inner record, not a separate file).
- Check that your import list matches exactly. Missing java.util.concurrent.Semaphore or java.util.concurrent.atomic.AtomicBoolean will cause unresolved symbol errors.
Why won't my ConcurrentEventBus compile?
Verify the full implementation compiles
Both implementation files are in place. Before moving to tests, confirm the entire project compiles with no errors or unchecked warnings.
- Run the compilation check by running this command:
mvn compile- Confirm the output shows BUILD SUCCESS with no warnings related to concurrency or generics.
- Verify you now have 8 source files by checking the src/main/java/io/github/dhruv0306/eventbus/ directory.
What should I see?
The Maven output should show the compilation target as Java 21. The final line reads BUILD SUCCESS. If you see any unchecked warnings, the @SuppressWarnings annotations are misplaced.
Seeing BUILD FAILURE?
- Check that BoundedDeadLetterQueue.java is in the same package as your interfaces. The file must be at src/main/java/io/github/dhruv0306/eventbus/BoundedDeadLetterQueue.java.
- Verify the package io.github.dhruv0306.eventbus; declaration is the first line in both new files.
- Make sure you're running mvn compile from the eventbus root directory (where pom.xml lives).
Help me fix my Maven compile error for the event bus project.
✔️ Awesome, I've got everything!
Great. Make sure both files are saved and the compile passed.
ⓧ I'd like to double check the full code
Here's BoundedDeadLetterQueue.java:
package io.github.dhruv0306.eventbus;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
public final class BoundedDeadLetterQueue implements DeadLetterQueue {
private final ArrayBlockingQueue<DeadLetter> queue;
public BoundedDeadLetterQueue(int capacity) {
if (capacity <= 0) {
throw new IllegalArgumentException("capacity must be positive");
}
this.queue = new ArrayBlockingQueue<>(capacity);
}
@Override
public boolean offer(Event event, Throwable reason) {
DeadLetter deadLetter = new DeadLetter(event, reason, System.currentTimeMillis());
return queue.offer(deadLetter);
}
@Override
public List<DeadLetter> drain() {
List<DeadLetter> drained = new ArrayList<>();
queue.drainTo(drained);
return drained;
}
@Override
public int size() {
return queue.size();
}
@Override
public boolean isFull() {
return queue.remainingCapacity() == 0;
}
}Here's ConcurrentEventBus.java:
package io.github.dhruv0306.eventbus;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.logging.Level;
import java.util.logging.Logger;
public final class ConcurrentEventBus implements EventBus {
private static final Logger LOGGER = Logger.getLogger(ConcurrentEventBus.class.getName());
private final ConcurrentHashMap<Class<? extends Event>, Set<FilteredSubscriber<?>>> subscribers;
private final ExecutorService asyncExecutor;
private final Semaphore asyncThrottle;
private final DeadLetterQueue deadLetterQueue;
private final AtomicBoolean closed;
public ConcurrentEventBus(EventBusConfig config) {
this.subscribers = new ConcurrentHashMap<>();
this.asyncExecutor = Executors.newVirtualThreadPerTaskExecutor();
this.asyncThrottle = new Semaphore(config.maxConcurrentDispatches());
this.deadLetterQueue = new BoundedDeadLetterQueue(config.deadLetterQueueCapacity());
this.closed = new AtomicBoolean(false);
}
public ConcurrentEventBus() {
this(EventBusConfig.defaults());
}
@Override
public <T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler) {
return subscribe(eventType, handler, event -> true);
}
@Override
public <T extends Event> Subscription subscribe(Class<T> eventType, Consumer<T> handler, Predicate<T> filter) {
if (closed.get()) {
throw new IllegalStateException("EventBus is closed");
}
FilteredSubscriber<T> subscriber = new FilteredSubscriber<>(eventType, handler, filter);
Set<FilteredSubscriber<?>> eventSubscribers = subscribers.computeIfAbsent(
eventType, k -> ConcurrentHashMap.newKeySet()
);
eventSubscribers.add(subscriber);
return new Subscription() {
private final AtomicBoolean active = new AtomicBoolean(true);
@Override
public void unsubscribe() {
if (active.compareAndSet(true, false)) {
eventSubscribers.remove(subscriber);
}
}
@Override
public boolean isActive() {
return active.get();
}
};
}
@Override
@SuppressWarnings("unchecked")
public <T extends Event> void publish(T event) {
if (closed.get()) {
throw new IllegalStateException("EventBus is closed");
}
Set<FilteredSubscriber<?>> eventSubscribers = subscribers.get(event.getClass());
if (eventSubscribers == null || eventSubscribers.isEmpty()) {
boolean accepted = deadLetterQueue.offer(event, new NoSubscriberException(event));
if (!accepted) {
throw new EventDroppedException(event,
"Dead letter queue is full, event dropped: " + event.getClass().getName());
}
return;
}
for (FilteredSubscriber<?> subscriber : eventSubscribers) {
FilteredSubscriber<T> typed = (FilteredSubscriber<T>) subscriber;
if (typed.filter().test(event)) {
try {
typed.handler().accept(event);
} catch (Exception e) {
deadLetterQueue.offer(event, e);
}
}
}
}
@Override
@SuppressWarnings("unchecked")
public <T extends Event> void publishAsync(T event) {
if (closed.get()) {
throw new IllegalStateException("EventBus is closed");
}
Set<FilteredSubscriber<?>> eventSubscribers = subscribers.get(event.getClass());
if (eventSubscribers == null || eventSubscribers.isEmpty()) {
boolean accepted = deadLetterQueue.offer(event, new NoSubscriberException(event));
if (!accepted) {
throw new EventDroppedException(event,
"Dead letter queue is full, event dropped: " + event.getClass().getName());
}
return;
}
for (FilteredSubscriber<?> subscriber : eventSubscribers) {
FilteredSubscriber<T> typed = (FilteredSubscriber<T>) subscriber;
if (typed.filter().test(event)) {
asyncExecutor.submit(() -> {
try {
asyncThrottle.acquire();
try {
typed.handler().accept(event);
} finally {
asyncThrottle.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOGGER.log(Level.WARNING, "Async dispatch interrupted for event: " + event, e);
} catch (Exception e) {
deadLetterQueue.offer(event, e);
LOGGER.log(Level.WARNING, "Subscriber threw exception for event: " + event, e);
}
});
}
}
}
public DeadLetterQueue deadLetterQueue() {
return deadLetterQueue;
}
@Override
public void close() {
if (closed.compareAndSet(false, true)) {
asyncExecutor.shutdown();
try {
if (!asyncExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
asyncExecutor.shutdownNow();
}
} catch (InterruptedException e) {
asyncExecutor.shutdownNow();
Thread.currentThread().interrupt();
}
subscribers.clear();
}
}
private record FilteredSubscriber<T extends Event>(
Class<T> eventType,
Consumer<T> handler,
Predicate<T> filter
) {}
private static class NoSubscriberException extends RuntimeException {
NoSubscriberException(Event event) {
super("No subscribers registered for event type: " + event.getClass().getName());
}
}
}Your event bus implementation compiles and is ready for action. Next up, you'll write a comprehensive test suite to prove the thread safety, filtering, DLQ behavior, and backpressure all work under pressure.
Write Tests and Publish to Maven Local
Your event bus compiles cleanly, but compilation alone does not prove correctness. Thread-safe code is notoriously tricky to validate by eye.
In this step, you will write a comprehensive JUnit 5 test suite that exercises every contract your library exposes. You will then install the packaged JAR to your local Maven repository so other projects on your machine can depend on it.
In this step, get ready to:
- Write 10 JUnit 5 tests covering dispatch, filtering, DLQ routing, and concurrency.
- Run the test suite and inspect the Surefire reports.
- Install the library JAR to Maven Local for downstream consumption.
Write the JUnit 5 test class
Your test class exercises the event bus from a consumer's perspective. Each test targets one behavioral contract so failures pinpoint exactly which guarantee broke.
- Create the file src/test/java/io/github/dhruv0306/eventbus/ConcurrentEventBusTest.java and paste the following test class:
package io.github.dhruv0306.eventbus;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.*;
class ConcurrentEventBusTest {
private ConcurrentEventBus bus;
// Domain event records used across all tests
record UserCreatedEvent(String username) implements Event {}
record OrderPlacedEvent(String orderId, double amount) implements Event {}
@BeforeEach
void setUp() {
// Small config values make failures surface faster in tests
bus = new ConcurrentEventBus(EventBusConfig.builder()
.maxConcurrentDispatches(16)
.deadLetterQueueCapacity(64)
.build());
}
@AfterEach
void tearDown() {
bus.close();
}
@Test
void shouldDeliverEventToSubscriber() {
List<String> received = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> received.add(event.username()));
bus.publish(new UserCreatedEvent("dhruv"));
assertEquals(1, received.size());
assertEquals("dhruv", received.getFirst());
}
@Test
void shouldDeliverEventToMultipleSubscribers() {
List<String> first = new CopyOnWriteArrayList<>();
List<String> second = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> first.add(event.username()));
bus.subscribe(UserCreatedEvent.class, event -> second.add(event.username()));
bus.publish(new UserCreatedEvent("dhruv"));
assertEquals(1, first.size());
assertEquals(1, second.size());
}
@Test
void shouldOnlyDeliverToMatchingEventType() {
List<String> users = new CopyOnWriteArrayList<>();
List<String> orders = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> users.add(event.username()));
bus.subscribe(OrderPlacedEvent.class, event -> orders.add(event.orderId()));
bus.publish(new UserCreatedEvent("dhruv"));
assertEquals(1, users.size());
assertEquals(0, orders.size());
}
@Test
void shouldFilterEventsWithPredicate() {
List<OrderPlacedEvent> highValue = new CopyOnWriteArrayList<>();
bus.subscribe(OrderPlacedEvent.class, highValue::add, order -> order.amount() > 100.0);
bus.publish(new OrderPlacedEvent("order-1", 50.0));
bus.publish(new OrderPlacedEvent("order-2", 200.0));
assertEquals(1, highValue.size());
assertEquals("order-2", highValue.getFirst().orderId());
}
@Test
void shouldUnsubscribeSuccessfully() {
List<String> received = new CopyOnWriteArrayList<>();
Subscription sub = bus.subscribe(UserCreatedEvent.class, event -> received.add(event.username()));
bus.publish(new UserCreatedEvent("first"));
sub.unsubscribe();
bus.publish(new UserCreatedEvent("second"));
assertEquals(1, received.size());
assertFalse(sub.isActive());
}
@Test
void shouldDeliverAsyncEvents() throws InterruptedException {
// CountDownLatch blocks the test thread until the async handler fires
CountDownLatch latch = new CountDownLatch(1);
List<String> received = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> {
received.add(event.username());
latch.countDown();
});
bus.publishAsync(new UserCreatedEvent("async-dhruv"));
assertTrue(latch.await(5, TimeUnit.SECONDS));
assertEquals("async-dhruv", received.getFirst());
}
@Test
void shouldRouteUnhandledEventsToDeadLetterQueue() {
// No subscriber registered for OrderPlacedEvent
bus.publish(new OrderPlacedEvent("orphan-order", 99.0));
assertEquals(1, bus.deadLetterQueue().size());
}
@Test
void shouldRouteSubscriberExceptionsToDeadLetterQueue() {
bus.subscribe(UserCreatedEvent.class, event -> {
throw new RuntimeException("Handler failure");
});
bus.publish(new UserCreatedEvent("failing"));
assertEquals(1, bus.deadLetterQueue().size());
}
@Test
void shouldThrowWhenDLQIsFull() {
// Create a bus with a tiny DLQ to force overflow
ConcurrentEventBus smallBus = new ConcurrentEventBus(EventBusConfig.builder()
.deadLetterQueueCapacity(2)
.build());
smallBus.publish(new UserCreatedEvent("orphan-1"));
smallBus.publish(new UserCreatedEvent("orphan-2"));
assertThrows(EventDroppedException.class, () ->
smallBus.publish(new UserCreatedEvent("orphan-3")));
smallBus.close();
}
@Test
void shouldHandleConcurrentPublishing() throws InterruptedException {
List<String> received = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> received.add(event.username()));
// Launch 100 virtual threads publishing simultaneously
int threadCount = 100;
CountDownLatch latch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
int index = i;
Thread.ofVirtual().start(() -> {
bus.publish(new UserCreatedEvent("user-" + index));
latch.countDown();
});
}
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertEquals(threadCount, received.size());
}
}What do these tests prove?
- Basic dispatch tests (first 3) confirm events reach correct subscribers and ignore unrelated types.
- Predicate filtering proves the third-argument Predicate<T> gates delivery. Only orders above 100.0 pass through.
- Unsubscription verifies the Subscription handle removes the handler from the concurrent set.
- Async delivery uses a CountDownLatch to block the test thread until the virtual thread completes dispatch.
- DLQ routing (tests 7-8) confirms unhandled events and handler exceptions both land in the dead-letter queue.
- DLQ overflow forces EventDroppedException by filling a capacity-2 queue, then publishing a third orphan event.
- Concurrent stress launches 100 virtual threads publishing simultaneously to prove thread safety under contention.
- Save ConcurrentEventBusTest.java.
Why CopyOnWriteArrayList?
Tests that assert on a shared collection from multiple threads need a thread-safe list. CopyOnWriteArrayList guarantees safe reads and writes without external synchronization, making assertions reliable even under concurrent publishing.
Compilation error on getFirst()?
The List.getFirst() method was added in Java 21 via SequencedCollection. If your IDE shows a red underline, confirm your project SDK is set to JDK 21. In IntelliJ, check File > Project Structure > Project SDK.
help me fix a compilation error with List.getFirst() in my JUnit test
Run the test suite
With your test class saved, Maven's Surefire plugin will discover and execute all @Test methods automatically.
- Run the full test lifecycle by executing this command from your eventbus directory:
mvn test- You should see output ending with Tests run: 10, Failures: 0, Errors: 0, Skipped: 0 followed by BUILD SUCCESS.
Where are the Surefire reports?
Maven Surefire writes XML and plain-text reports to target/surefire-reports/. The XML files (TEST-*.xml) are machine-readable and used by CI tools to display test results in dashboards.
Tests failing or not discovered?
If you see 0 tests run, confirm your test file is in src/test/java/ (not src/main/java/). Surefire only scans the test source tree.
If a specific test fails, read the assertion message carefully. A mismatch in assertEquals means either the event bus implementation has a bug or the test data does not match what you pasted.
help me debug failing JUnit tests in my event bus project
Install the library to Maven Local
Publishing to Maven Local makes your JAR available to any project on your machine. Other projects can declare a dependency on io.github.dhruv0306:eventbus:1.0.0-SNAPSHOT and Maven will resolve it from the local repository.
- Install the library by running:
mvn install- You should see BUILD SUCCESS in the output. The install phase runs compile, test, and package before copying the JAR.
- Verify the JAR exists by running:
ls ~/.m2/repository/io/github/dhruv0306/eventbus/1.0.0-SNAPSHOT/- You should see eventbus-1.0.0-SNAPSHOT.jar along with the POM and metadata files in the listing.
What is Maven Local?
Maven Local is a repository on your filesystem at ~/.m2/repository/. When Maven resolves dependencies, it checks this directory first before reaching out to remote repositories like Maven Central.
Installing here means you can immediately use your event bus library in a separate project without publishing it to the internet.
Install failed with test errors?
The install lifecycle includes the test phase. If any test fails, the build aborts before packaging. Fix the failing test first, then re-run mvn install.
help me fix a failing mvn install caused by test errors
✔️ Awesome, I've got everything!
Great. Double check you have saved ConcurrentEventBusTest.java and confirmed the JAR in your local repository.
ⓧ I'd like to double check the full code
Here is the complete test file for comparison:
package io.github.dhruv0306.eventbus;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.*;
class ConcurrentEventBusTest {
private ConcurrentEventBus bus;
record UserCreatedEvent(String username) implements Event {}
record OrderPlacedEvent(String orderId, double amount) implements Event {}
@BeforeEach
void setUp() {
bus = new ConcurrentEventBus(EventBusConfig.builder()
.maxConcurrentDispatches(16)
.deadLetterQueueCapacity(64)
.build());
}
@AfterEach
void tearDown() {
bus.close();
}
@Test
void shouldDeliverEventToSubscriber() {
List<String> received = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> received.add(event.username()));
bus.publish(new UserCreatedEvent("dhruv"));
assertEquals(1, received.size());
assertEquals("dhruv", received.getFirst());
}
@Test
void shouldDeliverEventToMultipleSubscribers() {
List<String> first = new CopyOnWriteArrayList<>();
List<String> second = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> first.add(event.username()));
bus.subscribe(UserCreatedEvent.class, event -> second.add(event.username()));
bus.publish(new UserCreatedEvent("dhruv"));
assertEquals(1, first.size());
assertEquals(1, second.size());
}
@Test
void shouldOnlyDeliverToMatchingEventType() {
List<String> users = new CopyOnWriteArrayList<>();
List<String> orders = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> users.add(event.username()));
bus.subscribe(OrderPlacedEvent.class, event -> orders.add(event.orderId()));
bus.publish(new UserCreatedEvent("dhruv"));
assertEquals(1, users.size());
assertEquals(0, orders.size());
}
@Test
void shouldFilterEventsWithPredicate() {
List<OrderPlacedEvent> highValue = new CopyOnWriteArrayList<>();
bus.subscribe(OrderPlacedEvent.class, highValue::add, order -> order.amount() > 100.0);
bus.publish(new OrderPlacedEvent("order-1", 50.0));
bus.publish(new OrderPlacedEvent("order-2", 200.0));
assertEquals(1, highValue.size());
assertEquals("order-2", highValue.getFirst().orderId());
}
@Test
void shouldUnsubscribeSuccessfully() {
List<String> received = new CopyOnWriteArrayList<>();
Subscription sub = bus.subscribe(UserCreatedEvent.class, event -> received.add(event.username()));
bus.publish(new UserCreatedEvent("first"));
sub.unsubscribe();
bus.publish(new UserCreatedEvent("second"));
assertEquals(1, received.size());
assertFalse(sub.isActive());
}
@Test
void shouldDeliverAsyncEvents() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
List<String> received = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> {
received.add(event.username());
latch.countDown();
});
bus.publishAsync(new UserCreatedEvent("async-dhruv"));
assertTrue(latch.await(5, TimeUnit.SECONDS));
assertEquals("async-dhruv", received.getFirst());
}
@Test
void shouldRouteUnhandledEventsToDeadLetterQueue() {
bus.publish(new OrderPlacedEvent("orphan-order", 99.0));
assertEquals(1, bus.deadLetterQueue().size());
}
@Test
void shouldRouteSubscriberExceptionsToDeadLetterQueue() {
bus.subscribe(UserCreatedEvent.class, event -> {
throw new RuntimeException("Handler failure");
});
bus.publish(new UserCreatedEvent("failing"));
assertEquals(1, bus.deadLetterQueue().size());
}
@Test
void shouldThrowWhenDLQIsFull() {
ConcurrentEventBus smallBus = new ConcurrentEventBus(EventBusConfig.builder()
.deadLetterQueueCapacity(2)
.build());
smallBus.publish(new UserCreatedEvent("orphan-1"));
smallBus.publish(new UserCreatedEvent("orphan-2"));
assertThrows(EventDroppedException.class, () ->
smallBus.publish(new UserCreatedEvent("orphan-3")));
smallBus.close();
}
@Test
void shouldHandleConcurrentPublishing() throws InterruptedException {
List<String> received = new CopyOnWriteArrayList<>();
bus.subscribe(UserCreatedEvent.class, event -> received.add(event.username()));
int threadCount = 100;
CountDownLatch latch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
int index = i;
Thread.ofVirtual().start(() -> {
bus.publish(new UserCreatedEvent("user-" + index));
latch.countDown();
});
}
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertEquals(threadCount, received.size());
}
}Your library is tested, packaged, and installed. Any Maven project on your machine can now pull it in as a dependency. Next up, the Secret Mission challenges you to publish it to a real remote registry.
Secret mission
Publish to GitHub Packages
Your event bus library is installed to Maven Local. Can you take it a step further and publish it to a real Maven registry? In this secret mission, you'll configure GitHub Packages as a remote repository and deploy your library so any Maven project can consume it over the network.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Local eventbus project folder on your Desktop.
- Maven Local artifact at ~/.m2/repository/io/github/dhruv0306/eventbus/.
- GitHub repository eventbus (Secret Mission).
- GitHub Package io.github.dhruv0306:eventbus:1.0.0-SNAPSHOT (Secret Mission).
- GitHub Personal Access Token with write:packages and read:packages scopes (Secret Mission).
- Maven settings.xml with stored credentials (Secret Mission).
Keep everything running
No action needed. Choose this if you plan to continue developing the event bus library or use it as a dependency in other projects.
- Your eventbus folder stays on your Desktop, ready for further development.
- The Maven Local artifact remains available for any project that declares the dependency.
- If you completed the Secret Mission, your GitHub repository and package remain published and consumable by others.
- Your Personal Access Token and settings.xml credentials remain active for future deployments.
Pause - I'll come back to this later
Since this is a local project with no running processes, there is nothing to pause. Your files and artifacts sit dormant until you need them.
- If you completed the Secret Mission, consider revoking your Personal Access Token to reduce your security surface while you are away. You can create a new one when you return.
- Navigate to your GitHub account Settings, then Developer settings, then Personal access tokens, then Tokens (classic).
- Find the token you created for this project and click Delete.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow each step below to fully clean up.
Delete the local project folder:
- Open a terminal and run this command to delete the project folder:
rmdir /s /q "C:\Users\Dhruv Patel\Desktop\eventbus"Remove the Maven Local artifact:
- Delete the cached artifact from your local Maven repository by running:
rmdir /s /q "C:\Users\Dhruv Patel\.m2\repository\io\github\dhruv0306\eventbus"Remove the stored credentials from settings.xml (Secret Mission):
- Open C:\Users\Dhruv Patel\.m2\settings.xml in your editor.
- Delete the entire <server> block with id github. If this was the only server entry, you can delete the entire file.
Revoke your Personal Access Token (Secret Mission):
- In GitHub, click your profile avatar in the top right corner.
- Select Settings.
- In the left sidebar, scroll down and click Developer settings.
- Click Personal access tokens, then Tokens (classic).
- Find the token you created for this project and click Delete.
Delete the GitHub Package (Secret Mission):
- Navigate to your eventbus repository on GitHub.
- Click the package name under the Packages section in the right sidebar.
- Click Package settings.
- Scroll to the bottom and click Delete this package.
Delete the GitHub repository (Secret Mission):
- Navigate to your eventbus repository on GitHub.
- Click Settings in the repository navigation bar.
- Scroll to the bottom of the General page to find the Danger Zone section.
- Click Delete this repository and follow the confirmation prompts.
Nice Work!
Nice Work!
You just built a production-grade, thread-safe in-memory event bus library from scratch using Java 21. This library is packaged, tested, and installed in your local Maven repository, ready for any downstream project to consume.
You've learned how to:
- Design a type-safe event bus implementing the Observer Pattern with Java 21 virtual threads for zero-overhead async dispatching.
- Implement production-defensive concurrency features including Semaphore-based backpressure throttling and a bounded dead-letter queue backed by ArrayBlockingQueue that fails gracefully instead of causing out-of-memory errors.
- Write a comprehensive JUnit 5 test suite proving thread safety across 100 concurrent virtual threads, then package and install the library as a consumable Maven JAR artifact.
- Secret Mission: Published the library to GitHub Packages as a real Maven registry artifact consumable by other projects.
Ready to quiz yourself?