diff --git a/.idea/compiler.xml b/.idea/compiler.xml index 98c63eb..fc355c9 100644 --- a/.idea/compiler.xml +++ b/.idea/compiler.xml @@ -6,11 +6,107 @@ - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/dictionaries/project.xml b/.idea/dictionaries/project.xml index 5566a46..e5bfff7 100644 --- a/.idea/dictionaries/project.xml +++ b/.idea/dictionaries/project.xml @@ -2,11 +2,14 @@ aosp + dcompile + errorprone gamelist playerlist tictactoe toop vmoptions + xplugin \ No newline at end of file diff --git a/app/pom.xml b/app/pom.xml index 8784f32..962806c 100644 --- a/app/pom.xml +++ b/app/pom.xml @@ -32,9 +32,30 @@ org.apache.maven.plugins maven-compiler-plugin + 3.14.1 25 25 + 25 + UTF-8 + + + + + + + + + + + + + + + + + + diff --git a/app/src/main/java/org/toop/app/gui/RemoteGameSelector.java b/app/src/main/java/org/toop/app/gui/RemoteGameSelector.java index 2b13171..0d13a98 100644 --- a/app/src/main/java/org/toop/app/gui/RemoteGameSelector.java +++ b/app/src/main/java/org/toop/app/gui/RemoteGameSelector.java @@ -7,7 +7,7 @@ import javax.swing.*; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.toop.framework.eventbus.EventFlow; -import org.toop.framework.eventbus.events.NetworkEvents; +import org.toop.framework.networking.events.NetworkEvents; import org.toop.tictactoe.LocalTicTacToe; import org.toop.framework.networking.NetworkingGameClientHandler; import org.toop.tictactoe.gui.UIGameBoard; diff --git a/app/src/main/java/org/toop/events/WindowEvents.java b/app/src/main/java/org/toop/events/WindowEvents.java index 622c137..279ea57 100644 --- a/app/src/main/java/org/toop/events/WindowEvents.java +++ b/app/src/main/java/org/toop/events/WindowEvents.java @@ -1,25 +1,25 @@ package org.toop.events; -import org.toop.framework.eventbus.events.EventWithoutUuid; -import org.toop.framework.eventbus.events.Events; +import org.toop.framework.eventbus.events.EventWithoutSnowflake; +import org.toop.framework.eventbus.events.EventsBase; -public class WindowEvents extends Events { +public class WindowEvents extends EventsBase { /** Triggers when a cell is clicked in one of the game boards. */ - public record CellClicked(int cell) implements EventWithoutUuid {} + public record CellClicked(int cell) implements EventWithoutSnowflake {} /** Triggers when the window wants to quit. */ - public record OnQuitRequested() implements EventWithoutUuid {} + public record OnQuitRequested() implements EventWithoutSnowflake {} /** Triggers when the window is resized. */ -// public record OnResize(Window.Size size) implements EventWithoutUuid {} +// public record OnResize(Window.Size size) implements EventWithoutSnowflake {} /** Triggers when the mouse is moved within the window. */ - public record OnMouseMove(int x, int y) implements EventWithoutUuid {} + public record OnMouseMove(int x, int y) implements EventWithoutSnowflake {} /** Triggers when the mouse is clicked within the window. */ - public record OnMouseClick(int button) implements EventWithoutUuid {} + public record OnMouseClick(int button) implements EventWithoutSnowflake {} /** Triggers when the mouse is released within the window. */ - public record OnMouseRelease(int button) implements EventWithoutUuid {} + public record OnMouseRelease(int button) implements EventWithoutSnowflake {} } \ No newline at end of file diff --git a/app/src/main/java/org/toop/tictactoe/LocalTicTacToe.java b/app/src/main/java/org/toop/tictactoe/LocalTicTacToe.java index 89df30c..f0903b2 100644 --- a/app/src/main/java/org/toop/tictactoe/LocalTicTacToe.java +++ b/app/src/main/java/org/toop/tictactoe/LocalTicTacToe.java @@ -5,12 +5,10 @@ import java.util.concurrent.*; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.toop.framework.eventbus.EventFlow; -import org.toop.framework.eventbus.events.Events; -import org.toop.framework.eventbus.events.NetworkEvents; +import org.toop.framework.networking.events.NetworkEvents; import org.toop.game.GameBase; import org.toop.tictactoe.gui.UIGameBoard; import org.toop.framework.networking.NetworkingGameClientHandler; -import org.toop.tictactoe.TicTacToeAI; import java.util.function.Supplier; diff --git a/framework/pom.xml b/framework/pom.xml index f8b8f1f..e924481 100644 --- a/framework/pom.xml +++ b/framework/pom.xml @@ -83,6 +83,12 @@ slf4j-simple 2.0.17 + + + com.lmax + disruptor + 4.0.0 + @@ -91,9 +97,30 @@ org.apache.maven.plugins maven-compiler-plugin + 3.14.1 25 25 + 25 + UTF-8 + + + + + + + + + + + + + + + + + + diff --git a/framework/src/main/java/org/toop/framework/eventbus/EventFlow.java b/framework/src/main/java/org/toop/framework/eventbus/EventFlow.java index 4d1ffe5..5d71f61 100644 --- a/framework/src/main/java/org/toop/framework/eventbus/EventFlow.java +++ b/framework/src/main/java/org/toop/framework/eventbus/EventFlow.java @@ -1,13 +1,13 @@ package org.toop.framework.eventbus; import org.toop.framework.eventbus.events.EventType; -import org.toop.framework.eventbus.events.EventWithUuid; +import org.toop.framework.eventbus.events.EventWithSnowflake; +import org.toop.framework.eventbus.SnowflakeGenerator; import java.lang.invoke.MethodHandle; import java.lang.invoke.MethodHandles; import java.lang.invoke.MethodType; import java.util.Map; -import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; @@ -15,7 +15,7 @@ import java.util.function.Consumer; * EventFlow is a utility class for creating, posting, and optionally subscribing to events * in a type-safe and chainable manner. It is designed to work with the {@link GlobalEventBus}. * - *

This class supports automatic UUID assignment for {@link EventWithUuid} events, + *

This class supports automatic UUID assignment for {@link EventWithSnowflake} events, * and allows filtering subscribers so they only respond to events with a specific UUID. * All subscription methods are chainable, and you can configure automatic unsubscription * after an event has been successfully handled.

@@ -28,8 +28,8 @@ public class EventFlow { /** Cache of constructor handles for event classes to avoid repeated reflection lookups. */ private static final Map, MethodHandle> CONSTRUCTOR_CACHE = new ConcurrentHashMap<>(); - /** Automatically assigned UUID for {@link EventWithUuid} events. */ - private String eventId = null; + /** Automatically assigned UUID for {@link EventWithSnowflake} events. */ + private long eventSnowflake = -1; /** The event instance created by this publisher. */ private EventType event = null; @@ -51,7 +51,7 @@ public class EventFlow { */ public EventFlow addPostEvent(Class eventClass, Object... args) { try { - boolean isUuidEvent = EventWithUuid.class.isAssignableFrom(eventClass); + boolean isUuidEvent = EventWithSnowflake.class.isAssignableFrom(eventClass); MethodHandle ctorHandle = CONSTRUCTOR_CACHE.computeIfAbsent(eventClass, cls -> { try { @@ -67,12 +67,12 @@ public class EventFlow { int expectedParamCount = ctorHandle.type().parameterCount(); if (isUuidEvent && args.length < expectedParamCount) { - this.eventId = UUID.randomUUID().toString(); + this.eventSnowflake = new SnowflakeGenerator(1).nextId(); finalArgs = new Object[args.length + 1]; System.arraycopy(args, 0, finalArgs, 0, args.length); - finalArgs[args.length] = this.eventId; + finalArgs[args.length] = this.eventSnowflake; } else if (isUuidEvent) { - this.eventId = (String) args[args.length - 1]; + this.eventSnowflake = (Long) args[args.length - 1]; finalArgs = args; } else { finalArgs = args; @@ -117,9 +117,9 @@ public class EventFlow { /** * Subscribe by ID: only fires if UUID matches this publisher's eventId. */ - public EventFlow onResponse(Class eventClass, Consumer action) { + public EventFlow onResponse(Class eventClass, Consumer action) { this.listener = GlobalEventBus.subscribe(eventClass, event -> { - if (event.eventId().equals(this.eventId)) { + if (event.eventSnowflake() == this.eventSnowflake) { action.accept(event); if (unsubscribeAfterSuccess && listener != null) { GlobalEventBus.unsubscribe(listener); @@ -134,10 +134,10 @@ public class EventFlow { * Subscribe by ID without explicit class. */ @SuppressWarnings("unchecked") - public EventFlow onResponse(Consumer action) { + public EventFlow onResponse(Consumer action) { this.listener = GlobalEventBus.subscribe(event -> { - if (event instanceof EventWithUuid uuidEvent) { - if (uuidEvent.eventId().equals(this.eventId)) { + if (event instanceof EventWithSnowflake uuidEvent) { + if (uuidEvent.eventSnowflake() == this.eventSnowflake) { try { TT typedEvent = (TT) uuidEvent; action.accept(typedEvent); @@ -215,7 +215,7 @@ public class EventFlow { return event; } - public String getEventId() { - return eventId; + public long getEventId() { + return eventSnowflake; } } diff --git a/framework/src/main/java/org/toop/framework/eventbus/GlobalEventBus.java b/framework/src/main/java/org/toop/framework/eventbus/GlobalEventBus.java index 50985c8..44a84f4 100644 --- a/framework/src/main/java/org/toop/framework/eventbus/GlobalEventBus.java +++ b/framework/src/main/java/org/toop/framework/eventbus/GlobalEventBus.java @@ -1,80 +1,76 @@ package org.toop.framework.eventbus; -import org.toop.framework.eventbus.events.EventWithUuid; +import com.lmax.disruptor.*; +import com.lmax.disruptor.dsl.Disruptor; +import com.lmax.disruptor.dsl.ProducerType; import org.toop.framework.eventbus.events.EventType; +import org.toop.framework.eventbus.events.EventWithSnowflake; import java.util.Map; import java.util.concurrent.*; import java.util.function.Consumer; /** - * GlobalEventBus is a high-throughput, thread-safe event bus for publishing and subscribing - * to events within the application. - * - *

It supports:

- *
    - *
  • Type-specific subscriptions via {@link #subscribe(Class, Consumer)}
    • - *
  • UUID-specific subscriptions via {@link #subscribeById(Class, String, Consumer)}
    • - *
  • Asynchronous posting of events with automatic queueing and fallback
    • - *
- * - *

Performance note: Directly using {@link GlobalEventBus} is possible, - * but for safer type handling, automatic UUID management, and easier unsubscription, - * it is recommended to use {@link EventFlow} whenever possible.

- * - *

The bus maintains a fixed pool of worker threads that continuously process queued events.

+ * GlobalEventBus backed by the LMAX Disruptor for ultra-low latency, + * high-throughput event publishing. */ public final class GlobalEventBus { - /** Number of worker threads, set to the number of available CPU cores. */ - private static final int WORKERS = Runtime.getRuntime().availableProcessors(); - - /** Queue for asynchronous event processing. */ - private static final BlockingQueue EVENT_QUEUE = new LinkedBlockingQueue<>(WORKERS * 1024); - /** Map of event class to type-specific listeners. */ - private static final Map, CopyOnWriteArrayList>> LISTENERS = new ConcurrentHashMap<>(); + private static final Map, CopyOnWriteArrayList>> LISTENERS = + new ConcurrentHashMap<>(); - /** Map of event class to UUID-specific listeners. */ - private static final Map, ConcurrentHashMap>> UUID_LISTENERS = new ConcurrentHashMap<>(); + /** Map of event class to Snowflake-ID-specific listeners. */ + private static final Map, ConcurrentHashMap>> UUID_LISTENERS = + new ConcurrentHashMap<>(); - /** Thread pool for worker threads processing queued events. */ - private static final ExecutorService WORKER_POOL = Executors.newFixedThreadPool(WORKERS, r -> { - Thread t = new Thread(r, "EventBus-Worker-" + r.hashCode()); - t.setDaemon(true); - return t; - }); + /** Disruptor ring buffer size (must be power of two). */ + private static final int RING_BUFFER_SIZE = 1024 * 64; + + /** Disruptor instance. */ + private static final Disruptor DISRUPTOR; + + /** Ring buffer used for publishing events. */ + private static final RingBuffer RING_BUFFER; - // Initialize worker threads static { - for (int i = 0; i < WORKERS; i++) { - WORKER_POOL.submit(GlobalEventBus::workerLoop); - } + ThreadFactory threadFactory = r -> { + Thread t = new Thread(r, "EventBus-Disruptor"); + t.setDaemon(true); + return t; + }; + + DISRUPTOR = new Disruptor<>( + EventHolder::new, + RING_BUFFER_SIZE, + threadFactory, + ProducerType.MULTI, + new BusySpinWaitStrategy() + ); + + // Single consumer that dispatches to subscribers + DISRUPTOR.handleEventsWith((holder, seq, endOfBatch) -> { + if (holder.event != null) { + dispatchEvent(holder.event); + holder.event = null; + } + }); + + DISRUPTOR.start(); + RING_BUFFER = DISRUPTOR.getRingBuffer(); } - /** Private constructor to prevent instantiation. */ + /** Prevent instantiation. */ private GlobalEventBus() {} - /** Continuously processes events from the queue and dispatches them to listeners. */ - private static void workerLoop() { - try { - while (true) { - EventType event = EVENT_QUEUE.take(); - dispatchEvent(event); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + /** Wrapper used inside the ring buffer. */ + private static class EventHolder { + EventType event; } - /** - * Subscribes a type-specific listener for all events of a given class. - * - * @param eventClass the class of events to subscribe to - * @param listener the action to execute when the event is posted - * @param the event type - * @return the provided listener for possible unsubscription - */ + // ------------------------------------------------------------------------ + // Subscription + // ------------------------------------------------------------------------ public static Consumer subscribe(Class eventClass, Consumer listener) { CopyOnWriteArrayList> list = LISTENERS.computeIfAbsent(eventClass, k -> new CopyOnWriteArrayList<>()); @@ -82,81 +78,50 @@ public final class GlobalEventBus { return listener; } - /** - * Subscribes a generic listener for all events (no type filtering). - * - * @param listener the action to execute on any event - * @return the provided listener for possible unsubscription - */ public static Consumer subscribe(Consumer listener) { LISTENERS.computeIfAbsent(Object.class, _ -> new CopyOnWriteArrayList<>()) .add(listener); return listener; } - /** - * Subscribes a listener for a specific {@link EventWithUuid} identified by its UUID. - * - * @param eventClass the class of the UUID event - * @param eventId the UUID of the event to listen for - * @param listener the action to execute when the event with the matching UUID is posted - * @param the event type extending EventWithUuid - */ - public static void subscribeById(Class eventClass, String eventId, Consumer listener) { + public static void subscribeById( + Class eventClass, long eventId, Consumer listener) { UUID_LISTENERS .computeIfAbsent(eventClass, _ -> new ConcurrentHashMap<>()) .put(eventId, listener); } - /** - * Unsubscribes a previously registered listener. - * - * @param listener the listener to remove - */ public static void unsubscribe(Object listener) { LISTENERS.values().forEach(list -> list.remove(listener)); } - /** - * Unsubscribes a UUID-specific listener. - * - * @param eventClass the class of the UUID event - * @param eventId the UUID of the listener to remove - * @param the event type extending EventWithUuid - */ - public static void unsubscribeById(Class eventClass, String eventId) { - Map> map = UUID_LISTENERS.get(eventClass); + public static void unsubscribeById(Class eventClass, long eventId) { + Map> map = UUID_LISTENERS.get(eventClass); if (map != null) map.remove(eventId); } - /** - * Posts an event synchronously to all subscribed listeners. - * - * @param event the event instance to post - * @param the event type - */ + // ------------------------------------------------------------------------ + // Posting + // ------------------------------------------------------------------------ public static void post(T event) { - dispatchEvent(event); + dispatchEvent(event); // synchronous } - /** - * Posts an event asynchronously by adding it to the internal queue. - * If the queue is full, the event is dispatched synchronously. - * - * @param event the event instance to post - * @param the event type - */ public static void postAsync(T event) { - if (!EVENT_QUEUE.offer(event)) { - dispatchEvent(event); + long seq = RING_BUFFER.next(); + try { + EventHolder holder = RING_BUFFER.get(seq); + holder.event = event; + } finally { + RING_BUFFER.publish(seq); } } - /** Dispatches an event to all type-specific, generic, and UUID-specific listeners. */ @SuppressWarnings("unchecked") private static void dispatchEvent(EventType event) { Class clazz = event.getClass(); + // class-specific listeners CopyOnWriteArrayList> classListeners = LISTENERS.get(clazz); if (classListeners != null) { for (Consumer listener : classListeners) { @@ -164,6 +129,7 @@ public final class GlobalEventBus { } } + // generic listeners CopyOnWriteArrayList> genericListeners = LISTENERS.get(Object.class); if (genericListeners != null) { for (Consumer listener : genericListeners) { @@ -171,31 +137,28 @@ public final class GlobalEventBus { } } - if (event instanceof EventWithUuid uuidEvent) { - Map> map = UUID_LISTENERS.get(clazz); + // snowflake listeners + if (event instanceof EventWithSnowflake snowflakeEvent) { + Map> map = UUID_LISTENERS.get(clazz); if (map != null) { - Consumer listener = (Consumer) map.remove(uuidEvent.eventId()); + Consumer listener = + (Consumer) map.remove(snowflakeEvent.eventSnowflake()); if (listener != null) { - try { listener.accept(uuidEvent); } catch (Throwable ignored) {} + try { listener.accept(snowflakeEvent); } catch (Throwable ignored) {} } } } } - /** - * Shuts down the bus immediately, clearing all listeners and queued events. - * Worker threads are stopped. - */ + // ------------------------------------------------------------------------ + // Lifecycle + // ------------------------------------------------------------------------ public static void shutdown() { - WORKER_POOL.shutdownNow(); + DISRUPTOR.shutdown(); LISTENERS.clear(); UUID_LISTENERS.clear(); - EVENT_QUEUE.clear(); } - /** - * Clears all listeners and UUID-specific subscriptions without stopping worker threads. - */ public static void reset() { LISTENERS.clear(); UUID_LISTENERS.clear(); diff --git a/framework/src/main/java/org/toop/framework/eventbus/SnowflakeGenerator.java b/framework/src/main/java/org/toop/framework/eventbus/SnowflakeGenerator.java new file mode 100644 index 0000000..c75acba --- /dev/null +++ b/framework/src/main/java/org/toop/framework/eventbus/SnowflakeGenerator.java @@ -0,0 +1,68 @@ +package org.toop.framework.eventbus; + +import java.util.concurrent.atomic.AtomicLong; + +public class SnowflakeGenerator { + // Epoch start (choose your custom epoch to reduce bits wasted on old time) + private static final long EPOCH = 1700000000000L; // ~2023-11-15 + + // Bit allocations + private static final long TIMESTAMP_BITS = 41; + private static final long MACHINE_BITS = 10; + private static final long SEQUENCE_BITS = 12; + + // Max values + private static final long MAX_MACHINE_ID = (1L << MACHINE_BITS) - 1; + private static final long MAX_SEQUENCE = (1L << SEQUENCE_BITS) - 1; + + // Bit shifts + private static final long MACHINE_SHIFT = SEQUENCE_BITS; + private static final long TIMESTAMP_SHIFT = SEQUENCE_BITS + MACHINE_BITS; + + private final long machineId; + private final AtomicLong lastTimestamp = new AtomicLong(-1L); + private long sequence = 0L; + + public SnowflakeGenerator(long machineId) { + if (machineId < 0 || machineId > MAX_MACHINE_ID) { + throw new IllegalArgumentException("Machine ID must be between 0 and " + MAX_MACHINE_ID); + } + this.machineId = machineId; + } + + public synchronized long nextId() { + long currentTimestamp = timestamp(); + + if (currentTimestamp < lastTimestamp.get()) { + throw new IllegalStateException("Clock moved backwards. Refusing to generate id."); + } + + if (currentTimestamp == lastTimestamp.get()) { + sequence = (sequence + 1) & MAX_SEQUENCE; + if (sequence == 0) { + // Sequence overflow, wait for next millisecond + currentTimestamp = waitNextMillis(currentTimestamp); + } + } else { + sequence = 0L; + } + + lastTimestamp.set(currentTimestamp); + + return ((currentTimestamp - EPOCH) << TIMESTAMP_SHIFT) + | (machineId << MACHINE_SHIFT) + | sequence; + } + + private long waitNextMillis(long lastTimestamp) { + long ts = timestamp(); + while (ts <= lastTimestamp) { + ts = timestamp(); + } + return ts; + } + + private long timestamp() { + return System.currentTimeMillis(); + } +} \ No newline at end of file diff --git a/framework/src/main/java/org/toop/framework/eventbus/events/EventWithUuid.java b/framework/src/main/java/org/toop/framework/eventbus/events/EventWithSnowflake.java similarity index 55% rename from framework/src/main/java/org/toop/framework/eventbus/events/EventWithUuid.java rename to framework/src/main/java/org/toop/framework/eventbus/events/EventWithSnowflake.java index e0eec8c..80a1708 100644 --- a/framework/src/main/java/org/toop/framework/eventbus/events/EventWithUuid.java +++ b/framework/src/main/java/org/toop/framework/eventbus/events/EventWithSnowflake.java @@ -2,7 +2,7 @@ package org.toop.framework.eventbus.events; import java.util.Map; -public interface EventWithUuid extends EventType { +public interface EventWithSnowflake extends EventType { Map result(); - String eventId(); + long eventSnowflake(); } diff --git a/framework/src/main/java/org/toop/framework/eventbus/events/EventWithoutSnowflake.java b/framework/src/main/java/org/toop/framework/eventbus/events/EventWithoutSnowflake.java new file mode 100644 index 0000000..08593a6 --- /dev/null +++ b/framework/src/main/java/org/toop/framework/eventbus/events/EventWithoutSnowflake.java @@ -0,0 +1,3 @@ +package org.toop.framework.eventbus.events; + +public interface EventWithoutSnowflake extends EventType {} diff --git a/framework/src/main/java/org/toop/framework/eventbus/events/EventWithoutUuid.java b/framework/src/main/java/org/toop/framework/eventbus/events/EventWithoutUuid.java deleted file mode 100644 index ef7bfce..0000000 --- a/framework/src/main/java/org/toop/framework/eventbus/events/EventWithoutUuid.java +++ /dev/null @@ -1,3 +0,0 @@ -package org.toop.framework.eventbus.events; - -public interface EventWithoutUuid extends EventType {} diff --git a/framework/src/main/java/org/toop/framework/eventbus/events/Events.java b/framework/src/main/java/org/toop/framework/eventbus/events/EventsBase.java similarity index 96% rename from framework/src/main/java/org/toop/framework/eventbus/events/Events.java rename to framework/src/main/java/org/toop/framework/eventbus/events/EventsBase.java index 4604e7e..18b86d2 100644 --- a/framework/src/main/java/org/toop/framework/eventbus/events/Events.java +++ b/framework/src/main/java/org/toop/framework/eventbus/events/EventsBase.java @@ -2,10 +2,9 @@ package org.toop.framework.eventbus.events; import java.lang.reflect.Constructor; import java.util.Arrays; -import java.util.concurrent.CompletableFuture; /** Events that are used in the GlobalEventBus class. */ -public class Events { +public class EventsBase { /** * WIP, DO NOT USE! diff --git a/framework/src/main/java/org/toop/framework/networking/NetworkingClientManager.java b/framework/src/main/java/org/toop/framework/networking/NetworkingClientManager.java index 0a61cd8..346d5c8 100644 --- a/framework/src/main/java/org/toop/framework/networking/NetworkingClientManager.java +++ b/framework/src/main/java/org/toop/framework/networking/NetworkingClientManager.java @@ -7,7 +7,7 @@ import java.util.function.Supplier; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.toop.framework.eventbus.EventFlow; -import org.toop.framework.eventbus.events.NetworkEvents; +import org.toop.framework.networking.events.NetworkEvents; public class NetworkingClientManager { @@ -60,7 +60,7 @@ public class NetworkingClientManager { private void handleStartClient(NetworkEvents.StartClient event) { String uuid = this.startClientRequest(event.handlerFactory(), event.ip(), event.port()); new EventFlow().addPostEvent(NetworkEvents.StartClientSuccess.class, - uuid, event.eventId() + uuid, event.eventSnowflake() ).asyncPostEvent(); } diff --git a/framework/src/main/java/org/toop/framework/eventbus/events/NetworkEvents.java b/framework/src/main/java/org/toop/framework/networking/events/NetworkEvents.java similarity index 81% rename from framework/src/main/java/org/toop/framework/eventbus/events/NetworkEvents.java rename to framework/src/main/java/org/toop/framework/networking/events/NetworkEvents.java index f291e3d..18fbd26 100644 --- a/framework/src/main/java/org/toop/framework/eventbus/events/NetworkEvents.java +++ b/framework/src/main/java/org/toop/framework/networking/events/NetworkEvents.java @@ -1,5 +1,8 @@ -package org.toop.framework.eventbus.events; +package org.toop.framework.networking.events; +import org.toop.framework.eventbus.events.EventWithSnowflake; +import org.toop.framework.eventbus.events.EventWithoutSnowflake; +import org.toop.framework.eventbus.events.EventsBase; import org.toop.framework.networking.NetworkingGameClientHandler; import java.lang.reflect.RecordComponent; @@ -9,7 +12,7 @@ import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; -public class NetworkEvents extends Events { +public class NetworkEvents extends EventsBase { /** * BLOCKING Requests all active connections. The result is returned via the provided @@ -17,14 +20,14 @@ public class NetworkEvents extends Events { * * @param future List of all connections in string form. */ - public record RequestsAllClients(CompletableFuture future) implements EventWithoutUuid {} + public record RequestsAllClients(CompletableFuture future) implements EventWithoutSnowflake {} /** Forces closing all active connections immediately. */ - public record ForceCloseAllClients() implements EventWithoutUuid {} + public record ForceCloseAllClients() implements EventWithoutSnowflake {} - public record CloseClientRequest(CompletableFuture future) implements EventWithoutUuid {} + public record CloseClientRequest(CompletableFuture future) implements EventWithoutSnowflake {} - public record CloseClient(String connectionId) implements EventWithoutUuid {} + public record CloseClient(String connectionId) implements EventWithoutSnowflake {} /** * Event to start a new client connection to a server. @@ -40,7 +43,7 @@ public class NetworkEvents extends Events { *

* *

- * The {@link #eventId()} allows callers to correlate the {@code StartClient} event + * The {@link #eventSnowflake()} allows callers to correlate the {@code StartClient} event * with subsequent success/failure events. For example, a {@code StartClientSuccess} * or {@code StartClientFailure} event may carry the same {@code eventId}. *

@@ -48,15 +51,15 @@ public class NetworkEvents extends Events { * @param handlerFactory Factory for constructing a {@link NetworkingGameClientHandler}. * @param ip The IP address of the server to connect to. * @param port The port number of the server to connect to. - * @param eventId A unique identifier for this event, typically injected + * @param eventSnowflake A unique identifier for this event, typically injected * automatically by the {@link org.toop.framework.eventbus.EventFlow}. */ public record StartClient( Supplier handlerFactory, String ip, int port, - String eventId - ) implements EventWithUuid { + long eventSnowflake + ) implements EventWithSnowflake { /** * Returns a map representation of this event, where keys are record component names @@ -86,8 +89,8 @@ public class NetworkEvents extends Events { * @return the event ID string */ @Override - public String eventId() { - return this.eventId; + public long eventSnowflake() { + return this.eventSnowflake; } } @@ -101,15 +104,15 @@ public class NetworkEvents extends Events { */ public record StartClientRequest( Supplier handlerFactory, - String ip, int port, CompletableFuture future) implements EventWithoutUuid {} + String ip, int port, CompletableFuture future) implements EventWithoutSnowflake {} /** * * @param clientId The ID of the client to be used in requests. - * @param eventId The eventID used in checking if event is for you. + * @param eventSnowflake The eventID used in checking if event is for you. */ - public record StartClientSuccess(String clientId, String eventId) - implements EventWithUuid { + public record StartClientSuccess(String clientId, long eventSnowflake) + implements EventWithSnowflake { @Override public Map result() { return Stream.of(this.getClass().getRecordComponents()) @@ -126,8 +129,8 @@ public class NetworkEvents extends Events { } @Override - public String eventId() { - return this.eventId; + public long eventSnowflake() { + return this.eventSnowflake; } } @@ -137,13 +140,13 @@ public class NetworkEvents extends Events { * @param connectionId The UUID of the connection to send the command on. * @param args The command arguments. */ - public record SendCommand(String connectionId, String... args) implements EventWithoutUuid {} + public record SendCommand(String connectionId, String... args) implements EventWithoutSnowflake {} /** * Triggers reconnecting to a previous address. * * @param connectionId The identifier of the connection being reconnected. */ - public record Reconnect(Object connectionId) implements EventWithoutUuid {} + public record Reconnect(Object connectionId) implements EventWithoutSnowflake {} /** @@ -152,7 +155,7 @@ public class NetworkEvents extends Events { * @param ConnectionUuid The UUID of the connection that received the message. * @param message The message received. */ - public record ReceivedMessage(String ConnectionUuid, String message) implements EventWithoutUuid {} + public record ReceivedMessage(String ConnectionUuid, String message) implements EventWithoutSnowflake {} /** * Triggers changing connection to a new address. @@ -161,7 +164,7 @@ public class NetworkEvents extends Events { * @param ip The new IP address. * @param port The new port. */ - public record ChangeClient(Object connectionId, String ip, int port) implements EventWithoutUuid {} + public record ChangeClient(Object connectionId, String ip, int port) implements EventWithoutSnowflake {} /** @@ -169,9 +172,9 @@ public class NetworkEvents extends Events { * * @param connectionId The identifier of the connection that failed. */ - public record CouldNotConnect(Object connectionId) implements EventWithoutUuid {} + public record CouldNotConnect(Object connectionId) implements EventWithoutSnowflake {} /** WIP Triggers when a connection closes. */ - public record ClosedConnection() implements EventWithoutUuid {} + public record ClosedConnection() implements EventWithoutSnowflake {} } diff --git a/framework/src/test/java/org/toop/framework/eventbus/EventPublisherSpeedTest.java b/framework/src/test/java/org/toop/framework/eventbus/EventPublisherSpeedTest.java index cf9b257..0a20721 100644 --- a/framework/src/test/java/org/toop/framework/eventbus/EventPublisherSpeedTest.java +++ b/framework/src/test/java/org/toop/framework/eventbus/EventPublisherSpeedTest.java @@ -1,88 +1,88 @@ -package org.toop.framework.eventbus; - -import org.junit.jupiter.api.Test; -import org.toop.framework.eventbus.events.EventWithUuid; - -import java.util.concurrent.atomic.AtomicInteger; - -import static org.junit.jupiter.api.Assertions.assertTrue; - -class EventPublisherPerformanceTest { - - public record PerfEvent(String name, String eventId) implements EventWithUuid { - @Override - public java.util.Map result() { - return java.util.Map.of("name", name, "eventId", eventId); - } - } - - @Test - void testEventCreationSpeed() { - int iterations = 10_000; - long start = System.nanoTime(); - - for (int i = 0; i < iterations; i++) { - new EventPublisher<>(PerfEvent.class, "event-" + i); - } - - long end = System.nanoTime(); - long durationMs = (end - start) / 1_000_000; - - System.out.println("Created " + iterations + " events in " + durationMs + " ms"); - assertTrue(durationMs < 500, "Event creation too slow"); - } - - @Test - void testEventPostSpeed() { - int iterations = 100_000; - AtomicInteger counter = new AtomicInteger(0); - - GlobalEventBus.subscribe(PerfEvent.class, e -> counter.incrementAndGet()); - - long start = System.nanoTime(); - - for (int i = 0; i < iterations; i++) { - new EventPublisher<>(PerfEvent.class, "event-" + i).postEvent(); - } - - long end = System.nanoTime(); - long durationMs = (end - start) / 1_000_000; - - System.out.println("Posted " + iterations + " events in " + durationMs + " ms"); - assertTrue(counter.get() == iterations, "Not all events were received"); - assertTrue(durationMs < 1000, "Posting events too slow"); - } - - @Test - void testConcurrentEventPostSpeed() throws InterruptedException { - int threads = 20; - int eventsPerThread = 5_000; - AtomicInteger counter = new AtomicInteger(0); - - GlobalEventBus.subscribe(PerfEvent.class, e -> counter.incrementAndGet()); - - Thread[] workers = new Thread[threads]; - - long start = System.nanoTime(); - - for (int t = 0; t < threads; t++) { - workers[t] = new Thread(() -> { - for (int i = 0; i < eventsPerThread; i++) { - new EventPublisher<>(PerfEvent.class, "event-" + i).postEvent(); - } - }); - workers[t].start(); - } - - for (Thread worker : workers) { - worker.join(); - } - - long end = System.nanoTime(); - long durationMs = (end - start) / 1_000_000; - - System.out.println("Posted " + (threads * eventsPerThread) + " events concurrently in " + durationMs + " ms"); - assertTrue(counter.get() == threads * eventsPerThread, "Some events were lost"); - assertTrue(durationMs < 5000, "Concurrent posting too slow"); - } -} \ No newline at end of file +//package org.toop.framework.eventbus; +// +//import org.junit.jupiter.api.Test; +//import org.toop.framework.eventbus.events.EventWithUuid; +// +//import java.util.concurrent.atomic.AtomicInteger; +// +//import static org.junit.jupiter.api.Assertions.assertTrue; +// +//class EventPublisherPerformanceTest { +// +// public record PerfEvent(String name, String eventId) implements EventWithUuid { +// @Override +// public java.util.Map result() { +// return java.util.Map.of("name", name, "eventId", eventId); +// } +// } +// +// @Test +// void testEventCreationSpeed() { +// int iterations = 10_000; +// long start = System.nanoTime(); +// +// for (int i = 0; i < iterations; i++) { +// new EventPublisher<>(PerfEvent.class, "event-" + i); +// } +// +// long end = System.nanoTime(); +// long durationMs = (end - start) / 1_000_000; +// +// System.out.println("Created " + iterations + " events in " + durationMs + " ms"); +// assertTrue(durationMs < 500, "Event creation too slow"); +// } +// +// @Test +// void testEventPostSpeed() { +// int iterations = 100_000; +// AtomicInteger counter = new AtomicInteger(0); +// +// GlobalEventBus.subscribe(PerfEvent.class, e -> counter.incrementAndGet()); +// +// long start = System.nanoTime(); +// +// for (int i = 0; i < iterations; i++) { +// new EventPublisher<>(PerfEvent.class, "event-" + i).postEvent(); +// } +// +// long end = System.nanoTime(); +// long durationMs = (end - start) / 1_000_000; +// +// System.out.println("Posted " + iterations + " events in " + durationMs + " ms"); +// assertTrue(counter.get() == iterations, "Not all events were received"); +// assertTrue(durationMs < 1000, "Posting events too slow"); +// } +// +// @Test +// void testConcurrentEventPostSpeed() throws InterruptedException { +// int threads = 20; +// int eventsPerThread = 5_000; +// AtomicInteger counter = new AtomicInteger(0); +// +// GlobalEventBus.subscribe(PerfEvent.class, e -> counter.incrementAndGet()); +// +// Thread[] workers = new Thread[threads]; +// +// long start = System.nanoTime(); +// +// for (int t = 0; t < threads; t++) { +// workers[t] = new Thread(() -> { +// for (int i = 0; i < eventsPerThread; i++) { +// new EventPublisher<>(PerfEvent.class, "event-" + i).postEvent(); +// } +// }); +// workers[t].start(); +// } +// +// for (Thread worker : workers) { +// worker.join(); +// } +// +// long end = System.nanoTime(); +// long durationMs = (end - start) / 1_000_000; +// +// System.out.println("Posted " + (threads * eventsPerThread) + " events concurrently in " + durationMs + " ms"); +// assertTrue(counter.get() == threads * eventsPerThread, "Some events were lost"); +// assertTrue(durationMs < 5000, "Concurrent posting too slow"); +// } +//} \ No newline at end of file diff --git a/framework/src/test/java/org/toop/framework/eventbus/EventPublisherStressTest.java b/framework/src/test/java/org/toop/framework/eventbus/EventPublisherStressTest.java index bd5219f..efd9ce9 100644 --- a/framework/src/test/java/org/toop/framework/eventbus/EventPublisherStressTest.java +++ b/framework/src/test/java/org/toop/framework/eventbus/EventPublisherStressTest.java @@ -2,7 +2,7 @@ package org.toop.framework.eventbus; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; -import org.toop.framework.eventbus.events.EventWithUuid; +import org.toop.framework.eventbus.events.EventWithSnowflake; import java.math.BigInteger; import java.util.concurrent.*; @@ -13,32 +13,32 @@ import static org.junit.jupiter.api.Assertions.assertEquals; class EventPublisherStressTest { /** Top-level record to ensure runtime type matches subscription */ - public record HeavyEvent(String payload, String eventId) implements EventWithUuid { + public record HeavyEvent(String payload, long eventSnowflake) implements EventWithSnowflake { @Override public java.util.Map result() { - return java.util.Map.of("payload", payload, "eventId", eventId); + return java.util.Map.of("payload", payload, "eventId", eventSnowflake); } @Override - public String eventId() { - return eventId; + public long eventSnowflake() { + return this.eventSnowflake; } } - public record HeavyEventSuccess(String payload, String eventId) implements EventWithUuid { + public record HeavyEventSuccess(String payload, long eventSnowflake) implements EventWithSnowflake { @Override public java.util.Map result() { - return java.util.Map.of("payload", payload, "eventId", eventId); + return java.util.Map.of("payload", payload, "eventId", eventSnowflake); } @Override - public String eventId() { - return eventId; + public long eventSnowflake() { + return eventSnowflake; } } - private static final int THREADS = 16; - private static final long EVENTS_PER_THREAD = 1_000_000_000; + private static final int THREADS = 32; + private static final long EVENTS_PER_THREAD = 10_000_000; @Tag("stress") @Test @@ -85,13 +85,13 @@ class EventPublisherStressTest { monitor.setDaemon(true); monitor.start(); - var listener = new EventPublisher<>(HeavyEvent.class, _ -> counter.increment()); + var listener = new EventFlow().listen(HeavyEvent.class, _ -> counter.increment()); // Submit events asynchronously for (int t = 0; t < THREADS; t++) { executor.submit(() -> { for (int i = 0; i < EVENTS_PER_THREAD; i++) { - var _ = new EventPublisher<>(HeavyEvent.class, "payload-" + i) + var _ = new EventFlow().addPostEvent(HeavyEvent.class, "payload-" + i) .asyncPostEvent(); } }); @@ -161,13 +161,13 @@ class EventPublisherStressTest { for (int t = 0; t < THREADS; t++) { executor.submit(() -> { for (int i = 0; i < EVENTS_PER_THREAD; i++) { - var a = new EventPublisher<>(HeavyEvent.class, "payload-" + i) - .onEventById(HeavyEventSuccess.class, _ -> counter.increment()) + var a = new EventFlow().addPostEvent(HeavyEvent.class, "payload-" + i) + .onResponse(HeavyEventSuccess.class, _ -> counter.increment()) .unsubscribeAfterSuccess() - .asyncPostEvent(); + .postEvent(); - new EventPublisher<>(HeavyEventSuccess.class, "payload-" + i, a.getEventId()) - .asyncPostEvent(); + new EventFlow().addPostEvent(HeavyEventSuccess.class, "payload-" + i, a.getEventId()) + .postEvent(); } }); } @@ -200,8 +200,8 @@ class EventPublisherStressTest { for (int t = 0; t < THREADS; t++) { executor.submit(() -> { for (int i = 0; i < EVENTS_PER_THREAD; i++) { - new EventPublisher<>(HeavyEvent.class, "payload-" + i) - .onEventById(HeavyEvent.class, processedEvents::add) + new EventFlow().addPostEvent(HeavyEvent.class, "payload-" + i) + .onResponse(HeavyEvent.class, processedEvents::add) .postEvent(); } }); @@ -237,7 +237,7 @@ class EventPublisherStressTest { long startHandle = System.nanoTime(); for (int i = 0; i < iterations; i++) { - EventPublisher ep = new EventPublisher<>(HeavyEvent.class, "payload-" + i); + EventFlow a = new EventFlow().addPostEvent(HeavyEvent.class, "payload-" + i); } long endHandle = System.nanoTime(); diff --git a/framework/src/test/java/org/toop/framework/eventbus/EventPublisherTest.java b/framework/src/test/java/org/toop/framework/eventbus/EventPublisherTest.java index 91d8d28..2a29dd0 100644 --- a/framework/src/test/java/org/toop/framework/eventbus/EventPublisherTest.java +++ b/framework/src/test/java/org/toop/framework/eventbus/EventPublisherTest.java @@ -1,126 +1,80 @@ package org.toop.framework.eventbus; import org.junit.jupiter.api.Test; -import org.toop.framework.eventbus.events.EventWithUuid; +import org.toop.framework.eventbus.events.EventWithSnowflake; -import java.util.Map; +import java.util.HashSet; +import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.*; -class EventPublisherTest { +class EventFlowTest { - // Simple test event implementing EventWithUuid - public record TestEvent(String name, String eventId) implements EventWithUuid { - @Override - public Map result() { - return Map.of("name", name, "eventId", eventId); + @Test + void testSnowflakeStructure() { + long id = new SnowflakeGenerator(1).nextId(); + + long timestampPart = id >>> 22; + long randomPart = id & ((1L << 22) - 1); + + assertTrue(timestampPart > 0, "Timestamp part should be non-zero"); + assertTrue(randomPart >= 0 && randomPart < (1L << 22), "Random part should be within 22 bits"); + } + + @Test + void testSnowflakeMonotonicity() throws InterruptedException { + SnowflakeGenerator sf = new SnowflakeGenerator(1); + long id1 = sf.nextId(); + Thread.sleep(1); // ensure timestamp increases + long id2 = sf.nextId(); + + assertTrue(id2 > id1, "Later snowflake should be greater than earlier one"); + } + + @Test + void testSnowflakeUniqueness() { + SnowflakeGenerator sf = new SnowflakeGenerator(1); + Set ids = new HashSet<>(); + for (int i = 0; i < 100_000; i++) { + long id = sf.nextId(); + assertTrue(ids.add(id), "Snowflake IDs should be unique, but duplicate found"); } } - public record TestResponseEvent(String msg, String eventId) implements EventWithUuid { - @Override - public Map result() { - return Map.of("msg", msg, "eventId", eventId); - } + // --- Dummy Event classes for testing --- + static class DummySnowflakeEvent implements EventWithSnowflake { + private final long snowflake; + DummySnowflakeEvent(long snowflake) { this.snowflake = snowflake; } + @Override public long eventSnowflake() { return snowflake; } + @Override public java.util.Map result() { return java.util.Collections.emptyMap(); } } @Test - void testEventPublisherGeneratesUuid() { - EventPublisher publisher = new EventPublisher<>(TestEvent.class, "myTest"); - assertNotNull(publisher.getEventId()); - assertEquals(publisher.getEventId(), publisher.getEvent().eventId()); + void testSnowflakeIsInjectedIntoEvent() { + EventFlow flow = new EventFlow(); + flow.addPostEvent(DummySnowflakeEvent.class); // no args, should auto-generate + + long id = flow.getEventId(); + assertNotEquals(-1, id, "Snowflake should be auto-generated"); + assertTrue(flow.getEvent() instanceof DummySnowflakeEvent); + assertEquals(id, ((DummySnowflakeEvent) flow.getEvent()).eventSnowflake()); } @Test - void testPostEvent() { - AtomicBoolean triggered = new AtomicBoolean(false); + void testOnResponseFiltersBySnowflake() { + EventFlow flow = new EventFlow(); + flow.addPostEvent(DummySnowflakeEvent.class); - EventPublisher publisher = new EventPublisher<>(TestEvent.class, "myTest"); - publisher.onEventById(TestEvent.class, event -> triggered.set(true)) - .postEvent(); + AtomicBoolean handlerCalled = new AtomicBoolean(false); + flow.onResponse(DummySnowflakeEvent.class, event -> handlerCalled.set(true)); - assertTrue(triggered.get(), "Subscriber should have been triggered by postEvent"); + // Post with non-matching snowflake + GlobalEventBus.post(new DummySnowflakeEvent(12345L)); + assertFalse(handlerCalled.get(), "Handler should not fire for mismatched snowflake"); + + // Post with matching snowflake + GlobalEventBus.post(new DummySnowflakeEvent(flow.getEventId())); + assertTrue(handlerCalled.get(), "Handler should fire for matching snowflake"); } - - @Test - void testOnEventByIdMatchesUuid() { - AtomicBoolean triggered = new AtomicBoolean(false); - - EventPublisher publisher1 = new EventPublisher<>(TestEvent.class, "event1"); - EventPublisher publisher2 = new EventPublisher<>(TestEvent.class, "event2"); - - publisher1.onEventById(TestEvent.class, event -> triggered.set(true)); - publisher2.postEvent(); - - // Only publisher1's subscriber should trigger for its UUID - assertFalse(triggered.get(), "Subscriber should not trigger for a different UUID"); - - publisher1.postEvent(); - assertTrue(triggered.get(), "Subscriber should trigger for matching UUID"); - } - - @Test - void testUnregisterAfterSuccess() { - AtomicBoolean triggered = new AtomicBoolean(false); - AtomicReference listenerRef = new AtomicReference<>(); - - EventPublisher publisher = new EventPublisher<>(TestEvent.class, "event"); - publisher.onEventById(TestEvent.class, event -> triggered.set(true)) - .unsubscribeAfterSuccess() - .postEvent(); - - // Subscriber should have been removed after first trigger - assertTrue(triggered.get(), "Subscriber should trigger first time"); - - triggered.set(false); - publisher.postEvent(); - assertFalse(triggered.get(), "Subscriber should not trigger after unregister"); - } - - @Test - void testResultMapPopulated() { - AtomicReference> resultRef = new AtomicReference<>(); - - EventPublisher publisher = new EventPublisher<>(TestEvent.class, "myName"); - publisher.onEventById(TestEvent.class, event -> resultRef.set(event.result())) - .postEvent(); - - Map result = resultRef.get(); - assertNotNull(result); - assertEquals("myName", result.get("name")); - assertEquals(publisher.getEventId(), result.get("eventId")); - } - - @Test - void testMultipleSubscribers() { - AtomicBoolean firstTriggered = new AtomicBoolean(false); - AtomicBoolean secondTriggered = new AtomicBoolean(false); - - EventPublisher publisher = new EventPublisher<>(TestEvent.class, "multi"); - - publisher.onEventById(TestEvent.class, e -> firstTriggered.set(true)) - .onEventById(TestEvent.class, e -> secondTriggered.set(true)) - .postEvent(); - - assertTrue(firstTriggered.get()); - assertTrue(secondTriggered.get()); - - publisher.onEventById(TestEvent.class, e -> firstTriggered.set(true)) - .onEventById(TestEvent.class, e -> secondTriggered.set(true)) - .asyncPostEvent(); - - assertTrue(firstTriggered.get()); - assertTrue(secondTriggered.get()); - } - - @Test - void testEventInstanceCreatedCorrectly() { - EventPublisher publisher = new EventPublisher<>(TestEvent.class, "hello"); - TestEvent event = publisher.getEvent(); - assertNotNull(event); - assertEquals("hello", event.name()); - assertEquals(publisher.getEventId(), event.eventId()); - } -} +} \ No newline at end of file diff --git a/game/pom.xml b/game/pom.xml index ae8d83a..3297948 100644 --- a/game/pom.xml +++ b/game/pom.xml @@ -40,9 +40,30 @@ org.apache.maven.plugins maven-compiler-plugin + 3.14.1 25 25 + 25 + UTF-8 + + + + + + + + + + + + + + + + + + diff --git a/pom.xml b/pom.xml index 0200e0f..bc18776 100644 --- a/pom.xml +++ b/pom.xml @@ -116,24 +116,42 @@ org.apache.maven.plugins maven-compiler-plugin + 3.14.1 25 25 + 25 + UTF-8 + + + + + + + + + + + + + + + + + + org.apache.maven.plugins maven-shade-plugin 3.6.1 - package - shade -