Implement event handling system with EventExecutor, EventService, and EventHandlerRegistry; add tests for event processing

pom.xml

@@ -36,11 +36,33 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
         <plugins>
-
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.14.1</version>
+                <configuration>
+                    <release>${java.version}</release>
+                    <enablePreview>true</enablePreview>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>3.5.4</version>
+                <configuration>
+                    <argLine>--enable-preview</argLine>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>

src/main/java/de/extio/game_engine/event/Event.java

@@ -0,0 +1,4 @@
+package de.extio.game_engine.event;
+
+public interface Event {
+}

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<QueuedEvent> 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<Subtask<Object>>();
+					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<Subtask<Object>> tasks, final Class<? extends Event> 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<? extends Event> eventClass, Event event) {
+	}
+}

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<Class<? extends Event>, List<EventConsumer>> handlers = new ConcurrentHashMap<>();
+	
+	public void register(final Class<? extends Event> eventClass, final String consumerId, final Consumer<Event> 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<? extends Event> eventClass, final String consumerId) {
+		final var consumers = this.handlers.get(eventClass);
+		if (consumers != null) {
+			consumers.removeIf(ec -> ec.id().equals(consumerId));
+		}
+	}
+	
+	List<EventConsumer> getHandlers(final Class<? extends Event> eventClass) {
+		return this.handlers.get(eventClass);
+	}
+	
+	record EventConsumer(String id, Consumer<Event> consumer) {
+	}
+}

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<? extends Event> eventClass, final String consumerId, final Consumer<Event> consumer) {
+		this.registry.register(eventClass, consumerId, consumer);
+	}
+
+	public void unregister(final Class<? extends Event> eventClass, final String consumerId) {
+		this.registry.unregister(eventClass, consumerId);
+	}
+
+	public void fire(final Event event) {
+		this.executor.submit(event);
+	}
+}

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;
+	}
+}

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<Event> 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 {
+	}
+}