From 3373576c750400a3db7bef8e0a9c9e61f4a3fbd0 Mon Sep 17 00:00:00 2001 From: extio-de Date: Sun, 21 Dec 2025 23:34:19 +0100 Subject: [PATCH] Implement event handling system with EventExecutor, EventService, and EventHandlerRegistry; add tests for event processing --- pom.xml | 24 +- .../de/extio/game_engine/event/Event.java | 4 + .../game_engine/event/EventExecutor.java | 102 ++++++ .../event/EventHandlerRegistry.java | 32 ++ .../extio/game_engine/event/EventService.java | 26 ++ .../event/EventsAutoConfiguration.java | 26 ++ .../game_engine/event/EventServiceTest.java | 294 ++++++++++++++++++ 7 files changed, 507 insertions(+), 1 deletion(-) create mode 100644 src/main/java/de/extio/game_engine/event/Event.java create mode 100644 src/main/java/de/extio/game_engine/event/EventExecutor.java create mode 100644 src/main/java/de/extio/game_engine/event/EventHandlerRegistry.java create mode 100644 src/main/java/de/extio/game_engine/event/EventService.java create mode 100644 src/main/java/de/extio/game_engine/event/EventsAutoConfiguration.java create mode 100644 src/test/java/de/extio/game_engine/event/EventServiceTest.java diff --git a/pom.xml b/pom.xml index bda60e8..daa52ee 100644 --- a/pom.xml +++ b/pom.xml @@ -36,11 +36,33 @@ spring-boot-starter-test test + + + org.awaitility + awaitility + test + - + + org.apache.maven.plugins + maven-compiler-plugin + 3.14.1 + + ${java.version} + true + + + + org.apache.maven.plugins + maven-surefire-plugin + 3.5.4 + + --enable-preview + + \ No newline at end of file diff --git a/src/main/java/de/extio/game_engine/event/Event.java b/src/main/java/de/extio/game_engine/event/Event.java new file mode 100644 index 0000000..72b25e4 --- /dev/null +++ b/src/main/java/de/extio/game_engine/event/Event.java @@ -0,0 +1,4 @@ +package de.extio.game_engine.event; + +public interface Event { +} diff --git a/src/main/java/de/extio/game_engine/event/EventExecutor.java b/src/main/java/de/extio/game_engine/event/EventExecutor.java new file mode 100644 index 0000000..b863911 --- /dev/null +++ b/src/main/java/de/extio/game_engine/event/EventExecutor.java @@ -0,0 +1,102 @@ +package de.extio.game_engine.event; + +import java.util.ArrayList; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.concurrent.BlockingQueue; +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 org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class EventExecutor { + + private static final Logger LOGGER = LoggerFactory.getLogger(EventExecutor.class); + + private final BlockingQueue eventQueue = new LinkedBlockingQueue<>(); + + private final Thread processorThread; + + private final EventHandlerRegistry registry; + + public EventExecutor(final EventHandlerRegistry registry) { + this.registry = registry; + this.processorThread = Thread.ofPlatform() + .daemon(true) + .name("Events-Processor") + .start(() -> { + final var tasks = new ArrayList>(); + while (!Thread.currentThread().isInterrupted()) { + try { + final QueuedEvent queuedEvent = this.eventQueue.take(); + processEvent(tasks, queuedEvent.eventClass(), queuedEvent.event()); + } + catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + }); + } + + public void submit(final Event event) { + this.eventQueue.offer(new QueuedEvent(event.getClass(), event)); + } + + private void processEvent(final List> tasks, final Class eventClass, final Event event) throws InterruptedException { + final var consumers = this.registry.getHandlers(eventClass); + if (consumers != null) { + switch (consumers.size()) { + case 0: + return; + + case 1: + try { + consumers.getFirst().consumer().accept(event); + } + catch (final NoSuchElementException e) { + return; + } + catch (final Exception e) { + LOGGER.error("Event consumer failed", e); + } + return; + + default: + try (var scope = StructuredTaskScope.open(Joiner.awaitAll())) { + tasks.clear(); + for (final var consumer : consumers) { + tasks.add(scope.fork(() -> { + consumer.consumer().accept(event); + })); + } + + scope.join(); + + for (final var task : tasks) { + if (task.state() == State.FAILED) { + LOGGER.error("Event consumer failed", task.exception()); + } + } + } + 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); + } + } + } + } + + private record QueuedEvent(Class eventClass, Event event) { + } +} diff --git a/src/main/java/de/extio/game_engine/event/EventHandlerRegistry.java b/src/main/java/de/extio/game_engine/event/EventHandlerRegistry.java new file mode 100644 index 0000000..a75690d --- /dev/null +++ b/src/main/java/de/extio/game_engine/event/EventHandlerRegistry.java @@ -0,0 +1,32 @@ +package de.extio.game_engine.event; + +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; + +public class EventHandlerRegistry { + + private final ConcurrentMap, List> handlers = new ConcurrentHashMap<>(); + + public void register(final Class eventClass, final String consumerId, final Consumer consumer) { + final var eventConsumers = this.handlers.computeIfAbsent(eventClass, k -> new CopyOnWriteArrayList<>()); + eventConsumers.removeIf(ec -> ec.id().equals(consumerId)); + eventConsumers.add(new EventConsumer(consumerId, consumer)); + } + + public void unregister(final Class eventClass, final String consumerId) { + final var consumers = this.handlers.get(eventClass); + if (consumers != null) { + consumers.removeIf(ec -> ec.id().equals(consumerId)); + } + } + + List getHandlers(final Class eventClass) { + return this.handlers.get(eventClass); + } + + record EventConsumer(String id, Consumer consumer) { + } +} diff --git a/src/main/java/de/extio/game_engine/event/EventService.java b/src/main/java/de/extio/game_engine/event/EventService.java new file mode 100644 index 0000000..b70e503 --- /dev/null +++ b/src/main/java/de/extio/game_engine/event/EventService.java @@ -0,0 +1,26 @@ +package de.extio.game_engine.event; + +import java.util.function.Consumer; + +public class EventService { + + private final EventHandlerRegistry registry; + private final EventExecutor executor; + + public EventService(final EventHandlerRegistry registry, final EventExecutor executor) { + this.registry = registry; + this.executor = executor; + } + + public void register(final Class eventClass, final String consumerId, final Consumer consumer) { + this.registry.register(eventClass, consumerId, consumer); + } + + public void unregister(final Class eventClass, final String consumerId) { + this.registry.unregister(eventClass, consumerId); + } + + public void fire(final Event event) { + this.executor.submit(event); + } +} diff --git a/src/main/java/de/extio/game_engine/event/EventsAutoConfiguration.java b/src/main/java/de/extio/game_engine/event/EventsAutoConfiguration.java new file mode 100644 index 0000000..7a0c143 --- /dev/null +++ b/src/main/java/de/extio/game_engine/event/EventsAutoConfiguration.java @@ -0,0 +1,26 @@ +package de.extio.game_engine.event; + +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; + +@AutoConfiguration +@ConditionalOnProperty(name = "game-engine.events.enabled", havingValue = "true", matchIfMissing = true) +public class EventsAutoConfiguration { + + @Bean + EventHandlerRegistry eventHandlerRegistry() { + return new EventHandlerRegistry(); + } + + @Bean + EventExecutor eventExecutor(final EventHandlerRegistry registry) { + return new EventExecutor(registry); + } + + @Bean + EventService eventService(final EventHandlerRegistry registry, final EventExecutor executor) { + final EventService eventService = new EventService(registry, executor); + return eventService; + } +} diff --git a/src/test/java/de/extio/game_engine/event/EventServiceTest.java b/src/test/java/de/extio/game_engine/event/EventServiceTest.java new file mode 100644 index 0000000..da1d045 --- /dev/null +++ b/src/test/java/de/extio/game_engine/event/EventServiceTest.java @@ -0,0 +1,294 @@ +package de.extio.game_engine.event; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import java.time.Duration; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest(classes = EventsAutoConfiguration.class) +class EventServiceTest { + + @Autowired + private EventService eventService; + + private ConcurrentLinkedQueue receivedEvents; + private AtomicInteger invocationCount; + + @BeforeEach + void setUp() { + this.receivedEvents = new ConcurrentLinkedQueue<>(); + this.invocationCount = new AtomicInteger(0); + } + + @Test + void testZeroConsumers() { + final var event = new TestEvent("test"); + this.eventService.fire(event); + + await().pollDelay(Duration.ofMillis(50)) + .atMost(Duration.ofMillis(200)) + .untilAsserted(() -> assertThat(this.receivedEvents).isEmpty()); + } + + @Test + void testSingleConsumer() throws InterruptedException { + final var latch = new CountDownLatch(1); + this.eventService.register(TestEvent.class, "consumer1", event -> { + this.receivedEvents.add(event); + this.invocationCount.incrementAndGet(); + latch.countDown(); + }); + + final var event = new TestEvent("single"); + this.eventService.fire(event); + + assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(this.receivedEvents).hasSize(1); + assertThat(this.receivedEvents.peek()).isEqualTo(event); + assertThat(this.invocationCount.get()).isEqualTo(1); + } + + @Test + void testMultipleConsumers() throws InterruptedException { + final var latch = new CountDownLatch(3); + + this.eventService.register(TestEvent.class, "consumer1", event -> { + this.receivedEvents.add(event); + this.invocationCount.incrementAndGet(); + latch.countDown(); + }); + + this.eventService.register(TestEvent.class, "consumer2", event -> { + this.receivedEvents.add(event); + this.invocationCount.incrementAndGet(); + latch.countDown(); + }); + + this.eventService.register(TestEvent.class, "consumer3", event -> { + this.receivedEvents.add(event); + this.invocationCount.incrementAndGet(); + latch.countDown(); + }); + + final var event = new TestEvent("multiple"); + this.eventService.fire(event); + + assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(this.receivedEvents).hasSize(3); + assertThat(this.invocationCount.get()).isEqualTo(3); + assertThat(this.receivedEvents).allMatch(e -> e.equals(event)); + } + + @Test + void testConsumerUnregistration() throws InterruptedException { + final var latch = new CountDownLatch(1); + + this.eventService.register(TestEvent.class, "consumer1", event -> { + this.receivedEvents.add(event); + latch.countDown(); + }); + + this.eventService.unregister(TestEvent.class, "consumer1"); + + final var event = new TestEvent("unregistered"); + this.eventService.fire(event); + + await().pollDelay(Duration.ofMillis(50)) + .atMost(Duration.ofMillis(200)) + .untilAsserted(() -> assertThat(this.receivedEvents).isEmpty()); + } + + @Test + void testConsumerReregistration() throws InterruptedException { + final var firstLatch = new CountDownLatch(1); + final var secondLatch = new CountDownLatch(1); + + this.eventService.register(TestEvent.class, "consumer1", event -> { + this.receivedEvents.add(event); + firstLatch.countDown(); + }); + + this.eventService.register(TestEvent.class, "consumer1", event -> { + this.receivedEvents.add(event); + this.invocationCount.incrementAndGet(); + secondLatch.countDown(); + }); + + final var event = new TestEvent("reregistered"); + this.eventService.fire(event); + + assertThat(secondLatch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(this.receivedEvents).hasSize(1); + assertThat(this.invocationCount.get()).isEqualTo(1); + assertThat(firstLatch.getCount()).isEqualTo(1); + } + + @Test + void testSingleConsumerWithException() throws InterruptedException { + final var latch = new CountDownLatch(1); + + this.eventService.register(TestEvent.class, "failingConsumer", event -> { + latch.countDown(); + throw new RuntimeException("Consumer failed"); + }); + + final var event = new TestEvent("exception"); + this.eventService.fire(event); + + assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void testMultipleConsumersWithPartialFailure() throws InterruptedException { + final var latch = new CountDownLatch(2); + + this.eventService.register(TestEvent.class, "successConsumer1", event -> { + this.receivedEvents.add(event); + latch.countDown(); + }); + + this.eventService.register(TestEvent.class, "failingConsumer", event -> { + throw new RuntimeException("Consumer failed"); + }); + + this.eventService.register(TestEvent.class, "successConsumer2", event -> { + this.receivedEvents.add(event); + latch.countDown(); + }); + + final var event = new TestEvent("partial-failure"); + this.eventService.fire(event); + + assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(this.receivedEvents).hasSize(2); + } + + @Test + void testMultipleEvents() throws InterruptedException { + final var latch = new CountDownLatch(3); + + this.eventService.register(TestEvent.class, "consumer1", event -> { + this.receivedEvents.add(event); + latch.countDown(); + }); + + final var event1 = new TestEvent("event1"); + final var event2 = new TestEvent("event2"); + final var event3 = new TestEvent("event3"); + + this.eventService.fire(event1); + this.eventService.fire(event2); + this.eventService.fire(event3); + + assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(this.receivedEvents).containsExactly(event1, event2, event3); + } + + @Test + void testDifferentEventTypes() throws InterruptedException { + final var testEventLatch = new CountDownLatch(1); + final var otherEventLatch = new CountDownLatch(1); + + this.eventService.register(TestEvent.class, "testConsumer", event -> { + this.receivedEvents.add(event); + testEventLatch.countDown(); + }); + + this.eventService.register(OtherTestEvent.class, "otherConsumer", event -> { + this.receivedEvents.add(event); + otherEventLatch.countDown(); + }); + + final var testEvent = new TestEvent("test"); + final var otherEvent = new OtherTestEvent("other"); + + this.eventService.fire(testEvent); + this.eventService.fire(otherEvent); + + assertThat(testEventLatch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(otherEventLatch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(this.receivedEvents).containsExactlyInAnyOrder(testEvent, otherEvent); + } + + @Test + void testConcurrentEventSubmission() throws InterruptedException { + final var latch = new CountDownLatch(100); + + this.eventService.register(TestEvent.class, "consumer1", event -> { + this.invocationCount.incrementAndGet(); + latch.countDown(); + }); + + for (int i = 0; i < 100; i++) { + final var event = new TestEvent("event-" + i); + this.eventService.fire(event); + } + + assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); + assertThat(this.invocationCount.get()).isEqualTo(100); + } + + @Test + void testSlowConsumer() throws InterruptedException { + final var latch = new CountDownLatch(1); + + this.eventService.register(TestEvent.class, "slowConsumer", event -> { + try { + Thread.sleep(100); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + this.receivedEvents.add(event); + latch.countDown(); + }); + + final var event = new TestEvent("slow"); + this.eventService.fire(event); + + assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(this.receivedEvents).hasSize(1); + } + + @Test + void testMultipleSlowConsumersExecuteInParallel() throws InterruptedException { + final var startLatch = new CountDownLatch(3); + final var finishLatch = new CountDownLatch(3); + final long sleepDuration = 200; + + for (int i = 1; i <= 3; i++) { + this.eventService.register(TestEvent.class, "slowConsumer" + i, event -> { + startLatch.countDown(); + try { + Thread.sleep(sleepDuration); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + finishLatch.countDown(); + }); + } + + final long startTime = System.currentTimeMillis(); + this.eventService.fire(new TestEvent("parallel-test")); + + assertThat(startLatch.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(finishLatch.await(1, TimeUnit.SECONDS)).isTrue(); + + final long duration = System.currentTimeMillis() - startTime; + assertThat(duration).isLessThan(sleepDuration * 2); + } + + private record TestEvent(String data) implements Event { + } + + private record OtherTestEvent(String data) implements Event { + } +}