Event System

A well-structured subscription-based event system with support for synchronous and asynchronous operations, event priorities, cancellation, and more.

Features

• Synchronous and Asynchronous Event Processing: Support for both blocking and non-blocking event publishing.
• Event Priorities: Define the order in which listeners receive events.
• Event Cancellation: Ability to cancel events to prevent further processing.
• Thread Safety: Designed for concurrent use with proper synchronization.
• Event Cascading: Events can trigger further events in a chain.
• Event Modification: Listeners can modify events before they are passed on.
• Multithreaded Processing: Parallel event processing for improved performance.
• Event Queues and Buffering: Batch processing of events for high-throughput scenarios.
• Logging and Debugging: Comprehensive logging throughout the system.
• Persistent Events: Optional storage of events to disk for later retrieval.
• Distributed Event Processing: Integration with Kafka and RabbitMQ for distributed systems.

Architecture

The event system is built around the following core components:

• Event: The base interface for all events in the system.
• EventListener: The interface for classes that want to listen for and handle events.
• EventBus: The central component that manages event publishing and listener registration.
• AsyncEventProcessor: Provides advanced asynchronous event processing capabilities.
• PersistentEventManager: Manages the persistence of events to disk.
• DistributedEventManager: Manages distributed event processing with message brokers.

Getting Started

Basic Usage

// Create an event busEventBuseventBus = newDefaultEventBus();
// Create and register a listenerEventListener<MyEvent> listener = newAbstractEventListener<MyEvent>() {
@OverridepublicvoidonEvent(MyEventevent) {
System.out.println("Received event: " + event);
}
};
eventBus.register(listener);
// Create and publish an eventMyEventevent = newMyEvent("Hello, world!");
eventBus.publish(event);
// Publish an event asynchronouslyeventBus.publishAsync(event).thenAccept(e -> {
System.out.println("Event processed asynchronously: " + e);
});
// Unregister the listener when doneeventBus.unregister(listener);
// Shutdown the event buseventBus.shutdown();

Using Event Priorities

// Create listeners with different prioritiesEventListener<MyEvent> highPriorityListener = newAbstractEventListener<MyEvent>(EventPriority.HIGH) {
@OverridepublicvoidonEvent(MyEventevent) {
System.out.println("High priority listener received event: " + event);
}
};
EventListener<MyEvent> lowPriorityListener = newAbstractEventListener<MyEvent>(EventPriority.LOW) {
@OverridepublicvoidonEvent(MyEventevent) {
System.out.println("Low priority listener received event: " + event);
}
};
// Register the listenerseventBus.register(highPriorityListener);
eventBus.register(lowPriorityListener);
// Publish an event - high priority listener will be called firsteventBus.publish(newMyEvent("Priority test"));

Event Cascading

// Create a cascading event listenerCascadingEventListener<MyEvent, MyOtherEvent> cascadingListener = newCascadingEventListener<MyEvent, MyOtherEvent>(eventBus) {
@OverrideprotectedMyOtherEventprocessEvent(MyEventevent) {
returnnewMyOtherEvent("Cascaded from: " + event.getMessage());
}
};
// Register the cascading listener and a listener for the cascaded eventeventBus.register(cascadingListener);
eventBus.register(newAbstractEventListener<MyOtherEvent>() {
@OverridepublicvoidonEvent(MyOtherEventevent) {
System.out.println("Received cascaded event: " + event);
}
});
// Publish an event - will trigger the cascaded eventeventBus.publish(newMyEvent("Cascade test"));

Event Filtering

// Create a filtering event listenerFilteringEventListener<MyEvent> filteringListener = newFilteringEventListener<MyEvent>() {
@OverrideprotectedbooleanfilterEvent(MyEventevent) {
// Only allow events with messages longer than 5 charactersbooleanallowed = event.getMessage().length() > 5;
System.out.println("Filtering event: " + event + ", allowed: " + allowed);
returnallowed;
}
};
// Register the filtering listenereventBus.register(filteringListener);
// These events will be filtered differentlyeventBus.publish(newMyEvent("Short")); // Will be cancelledeventBus.publish(newMyEvent("Long enough")); // Will be allowed

Asynchronous Event Processing

// Create an async event processorAsyncEventProcessorasyncProcessor = newAsyncEventProcessor(eventBus);
asyncProcessor.start();
// Queue events for asynchronous processingasyncProcessor.queueEvent(newMyEvent("Queued event 1"));
asyncProcessor.queueEvent(newMyEvent("Queued event 2"));
// Schedule an event to be published after a delayasyncProcessor.scheduleEvent(newMyEvent("Delayed event"), 5, TimeUnit.SECONDS);
// Schedule a repeating eventasyncProcessor.scheduleRepeatingEvent(
() -> newMyEvent("Repeating event at " + Instant.now()),
0, 10, TimeUnit.SECONDS
);
// Stop the processor when doneasyncProcessor.stop();

Persistent Events

// Create a persistent event managerPersistentEventManagerpersistentManager = newPersistentEventManager();
persistentManager.start();
// Store events for persistencepersistentManager.storeEvent(newMyEvent("Persistent event 1"));
persistentManager.storeEvent(newMyEvent("Persistent event 2"));
// Save events to diskpersistentManager.saveAllEvents();
// Load events from diskList<MyEvent> loadedEvents = persistentManager.loadEvents(MyEvent.class);
for (MyEventevent : loadedEvents) {
System.out.println("Loaded event: " + event);
}
// Stop the manager when donepersistentManager.stop();

Distributed Event Processing

// Create a distributed event managerDistributedEventManagerdistributedManager = newDistributedEventManager(eventBus);
// Register a Kafka adapterdistributedManager.registerBrokerAdapter("kafka", newKafkaMessageBrokerAdapter("localhost:9092", eventBus));
// Register a RabbitMQ adapterdistributedManager.registerBrokerAdapter("rabbitmq", newRabbitMQMessageBrokerAdapter("localhost", "guest", "guest", eventBus));
// Start the distributed event managerdistributedManager.start();
// Publish an event to a specific brokerdistributedManager.publishToRemote("kafka", newMyEvent("Kafka event"));
// Publish an event to all brokersdistributedManager.publishToAllRemotes(newMyEvent("Broadcast event"));
// Stop the manager when donedistributedManager.stop();

Creating Custom Events

To create a custom event, extend the AbstractEvent class:

publicclassMyEventextendsAbstractEvent {
privatefinalStringmessage;
publicMyEvent(Stringmessage) {
super("MyEvent", null, true); // Name, source, cancellablethis.message = message;
}
publicStringgetMessage() {
returnmessage;
}
}

Logging

The Event System uses a flexible logging approach that works with or without SLF4J being available on the classpath:

• If SLF4J is available, it will use SLF4J for logging
• If SLF4J is not available, it will use a no-op logger implementation that does nothing

Configuration

To enable logging, add SLF4J and a compatible logging implementation (like Logback) to your classpath:

// In your build.gradle or build.gradle.kts
implementation("org.slf4j:slf4j-api:2.0.9")
implementation("ch.qos.logback:logback-classic:1.4.11")

Configure Logback by adding a logback.xml file to your classpath (typically in src/main/resources):

<configuration>
<appendername="CONSOLE"class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<!-- Set the default log level -->
<rootlevel="INFO">
<appender-refref="CONSOLE" />
</root>
<!-- Configure specific package log levels -->
<loggername="org.example.eventsystem"level="DEBUG" />
</configuration>

Using Logging in Custom Components

To use logging in your custom components:

importorg.example.eventsystem.util.LoggerFactory;
importorg.example.eventsystem.util.LoggerFactory.Logger;
publicclassMyCustomComponent {
// Create a logger instance for your classprivatestaticfinalLoggerlogger = LoggerFactory.getLogger(MyCustomComponent.class);
publicvoiddoSomething() {
// Log at different levelslogger.trace("Detailed trace information");
logger.debug("Debugging information");
logger.info("Informational message");
logger.warn("Warning message");
logger.error("Error message");
// Log with parameterslogger.info("Processing event: {}", eventName);
// Check if a log level is enabled before expensive operationsif (logger.isDebugEnabled()) {
StringexpensiveToGenerate = generateDetailedDebugInfo();
logger.debug("Detailed debug info: {}", expensiveToGenerate);
}
}
}

License

This project is licensed under the MIT License - see the LICENSE file for details.