diff --git a/README.md b/README.md index 9b4190b..cf9e760 100644 --- a/README.md +++ b/README.md @@ -202,7 +202,7 @@ This automatic cleanup ensures that modules can be safely loaded, activated, dea #### Purpose and Overview A lightweight, type-safe event bus that enables decoupled communication between game components. Components can register event handlers for specific event types and fire events that will be dispatched to all registered consumers. This pattern eliminates tight coupling between modules and simplifies complex interactions. -Events are processed asynchronously on a dedicated event processor thread using a blocking queue. When multiple handlers are registered for the same event type, they are executed concurrently using structured concurrency (virtual threads via `StructuredTaskScope`), providing excellent scalability for event-heavy workloads. Single handlers execute directly without concurrency overhead. Each consumer is identified by a unique string ID to enable selective unregistration. +Events are processed asynchronously on a dedicated event processor thread using a blocking queue. When multiple handlers are registered for the same event type, they are executed concurrently using structured concurrency (virtual threads), providing excellent scalability for event-heavy workloads. Single handlers execute directly without concurrency overhead. Each consumer is identified by a unique string ID to enable selective unregistration. #### Setup / Autoconfiguration **Configuration Class**: `EventsAutoConfiguration` diff --git a/pom.xml b/pom.xml index 22cd915..a12732c 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ 4.0.0 de.extio game-engine - 1.0.12 + 1.0.13 25 @@ -84,16 +84,12 @@ 3.15.0 ${java.version} - true org.apache.maven.plugins maven-surefire-plugin 3.5.5 - - --enable-preview - diff --git a/src/main/java/de/extio/game_engine/event/EventExecutor.java b/src/main/java/de/extio/game_engine/event/EventExecutor.java index ffcbfca..504bbfc 100644 --- a/src/main/java/de/extio/game_engine/event/EventExecutor.java +++ b/src/main/java/de/extio/game_engine/event/EventExecutor.java @@ -3,13 +3,11 @@ package de.extio.game_engine.event; import java.util.ArrayList; import java.util.List; import java.util.NoSuchElementException; +import java.util.concurrent.Callable; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.StructuredTaskScope; -import java.util.concurrent.StructuredTaskScope.FailedException; -import java.util.concurrent.StructuredTaskScope.Joiner; -import java.util.concurrent.StructuredTaskScope.Subtask; -import java.util.concurrent.StructuredTaskScope.Subtask.State; import java.util.function.Consumer; import org.slf4j.Logger; @@ -18,9 +16,11 @@ import org.slf4j.LoggerFactory; public class EventExecutor { private static final Logger LOGGER = LoggerFactory.getLogger(EventExecutor.class); + private static final ExecutorService EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); private final BlockingQueue eventQueue = new LinkedBlockingQueue<>(); + @SuppressWarnings("unused") private final Thread processorThread; private final EventHandlerRegistry registry; @@ -31,7 +31,7 @@ public class EventExecutor { .daemon(true) .name("Events-Processor") .start(() -> { - final var tasks = new ArrayList>(); + final var tasks = new ArrayList>(); while (!Thread.currentThread().isInterrupted()) { try { final QueuedEvent queuedEvent = this.eventQueue.take(); @@ -49,7 +49,8 @@ public class EventExecutor { this.eventQueue.offer(new QueuedEvent(event.getClass(), event)); } - private void processEvent(final List> tasks, final Class eventClass, final Event event) throws InterruptedException { + private void processEvent(final List> tasks, final Class eventClass, final Event event) throws InterruptedException { + tasks.clear(); final var consumers = this.registry.getHandlers(eventClass); if (consumers != null) { switch (consumers.size()) { @@ -58,7 +59,8 @@ public class EventExecutor { case 1: try { - ((Consumer) consumers.getFirst().consumer()).accept(event); + @SuppressWarnings("unchecked") final Consumer consumer = (Consumer) consumers.getFirst().consumer(); + consumer.accept(event); } catch (final NoSuchElementException e) { return; @@ -69,31 +71,26 @@ public class EventExecutor { return; default: - try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) { - tasks.clear(); - for (final var consumer : consumers) { - tasks.add(scope.fork(() -> { - ((Consumer) consumer.consumer()).accept(event); - })); - } - - scope.join(); - - for (final var task : tasks) { - if (task.state() == State.FAILED) { - LOGGER.error("Event consumer failed", task.exception()); + for (final var consumer : consumers) { + tasks.add(() -> { + @SuppressWarnings("unchecked") final Consumer eventConsumer = (Consumer) consumer.consumer(); + eventConsumer.accept(event); + return null; + }); + } + try { + for (final var future : EXECUTOR.invokeAll(tasks)) { + try { + future.get(); + } + catch (final Exception e) { + LOGGER.error("Event consumer failed", e.getCause() == null ? e : e.getCause()); } } } catch (final InterruptedException e) { throw e; } - catch (final FailedException e) { - LOGGER.error("One or more event consumers failed", e); - } - catch (final Exception e) { - LOGGER.error("Unexpected error while processing event", e); - } } } } diff --git a/src/main/java/de/extio/game_engine/module/ModuleExecutorImpl.java b/src/main/java/de/extio/game_engine/module/ModuleExecutorImpl.java index 66f074d..b961a7b 100644 --- a/src/main/java/de/extio/game_engine/module/ModuleExecutorImpl.java +++ b/src/main/java/de/extio/game_engine/module/ModuleExecutorImpl.java @@ -1,7 +1,10 @@ package de.extio.game_engine.module; -import java.util.concurrent.StructuredTaskScope; -import java.util.concurrent.StructuredTaskScope.Joiner; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.function.Consumer; import java.util.function.Function; @@ -11,6 +14,7 @@ import org.slf4j.LoggerFactory; public class ModuleExecutorImpl implements ModuleExecutor { private static final Logger LOGGER = LoggerFactory.getLogger(ModuleExecutorImpl.class); + private static final ExecutorService EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); private final ModuleService moduleManager; @@ -50,16 +54,25 @@ public class ModuleExecutorImpl implements ModuleExecutor { return; } - try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) { - synchronized (subscribers) { - for (final var module : subscribers) { - final var task = taskSupplier.apply(module); - if (task != null) { - scope.fork(task); - } + final List> tasks = new ArrayList<>(); + synchronized (subscribers) { + for (final var module : subscribers) { + final var task = taskSupplier.apply(module); + if (task != null) { + tasks.add(() -> { + task.run(); + return null; + }); } } - scope.join(); + } + + if (tasks.isEmpty()) { + return; + } + + try { + EXECUTOR.invokeAll(tasks); } catch (final InterruptedException e) { Thread.currentThread().interrupt();