refactor

framework/src/test/eventbus/EventPublisherSpeedTest.java

@@ -0,0 +1,89 @@
+package org.toop.eventbus;
+
+import org.junit.jupiter.api.Test;
+import org.toop.framework.eventbus.GlobalEventBus;
+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<String, Object> 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");
+    }
+}

framework/src/test/eventbus/EventPublisherStressTest.java

@@ -0,0 +1,247 @@
+package org.toop.eventbus;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.toop.framework.eventbus.events.EventWithUuid;
+
+import java.math.BigInteger;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.LongAdder;
+
+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 {
+        @Override
+        public java.util.Map<String, Object> result() {
+            return java.util.Map.of("payload", payload, "eventId", eventId);
+        }
+
+        @Override
+        public String eventId() {
+            return eventId;
+        }
+    }
+
+    public record HeavyEventSuccess(String payload, String eventId) implements EventWithUuid {
+        @Override
+        public java.util.Map<String, Object> result() {
+            return java.util.Map.of("payload", payload, "eventId", eventId);
+        }
+
+        @Override
+        public String eventId() {
+            return eventId;
+        }
+    }
+
+    private static final int THREADS = 16;
+    private static final long EVENTS_PER_THREAD = 1_000_000_000;
+
+    @Tag("stress")
+    @Test
+    void extremeConcurrencySendTest_progressWithMemory() throws InterruptedException {
+        LongAdder counter = new LongAdder();
+        ExecutorService executor = Executors.newFixedThreadPool(THREADS);
+
+        BigInteger totalEvents = BigInteger.valueOf(THREADS)
+                .multiply(BigInteger.valueOf(EVENTS_PER_THREAD));
+
+        long startTime = System.currentTimeMillis();
+
+        // Monitor thread for EPS and memory
+        Thread monitor = new Thread(() -> {
+            long lastCount = 0;
+            long lastTime = System.currentTimeMillis();
+            Runtime runtime = Runtime.getRuntime();
+
+            while (counter.sum() < totalEvents.longValue()) {
+                try { Thread.sleep(200); } catch (InterruptedException ignored) {}
+
+                long now = System.currentTimeMillis();
+                long completed = counter.sum();
+                long eventsThisPeriod = completed - lastCount;
+                double eps = eventsThisPeriod / ((now - lastTime) / 1000.0);
+
+                long usedMemory = runtime.totalMemory() - runtime.freeMemory();
+                double usedPercent = usedMemory * 100.0 / runtime.maxMemory();
+
+                System.out.printf(
+                        "Progress: %d/%d (%.2f%%), EPS: %.0f, Memory Used: %.2f MB (%.2f%%)%n",
+                        completed,
+                        totalEvents.longValue(),
+                        completed * 100.0 / totalEvents.doubleValue(),
+                        eps,
+                        usedMemory / 1024.0 / 1024.0,
+                        usedPercent
+                );
+
+                lastCount = completed;
+                lastTime = now;
+            }
+        });
+        monitor.setDaemon(true);
+        monitor.start();
+
+        var listener = new EventPublisher<>(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)
+                            .asyncPostEvent();
+                }
+            });
+        }
+
+        executor.shutdown();
+        executor.awaitTermination(10, TimeUnit.MINUTES);
+
+        listener.getResult();
+
+        long endTime = System.currentTimeMillis();
+        double durationSeconds = (endTime - startTime) / 1000.0;
+
+        System.out.println("Posted " + totalEvents + " events in " + durationSeconds + " seconds");
+        double averageEps = totalEvents.doubleValue() / durationSeconds;
+        System.out.printf("Average EPS: %.0f%n", averageEps);
+
+        assertEquals(totalEvents.longValue(), counter.sum());
+    }
+
+    @Tag("stress")
+    @Test
+    void extremeConcurrencySendAndReturnTest_progressWithMemory() throws InterruptedException {
+        LongAdder counter = new LongAdder();
+        ExecutorService executor = Executors.newFixedThreadPool(THREADS);
+
+        BigInteger totalEvents = BigInteger.valueOf(THREADS)
+                .multiply(BigInteger.valueOf(EVENTS_PER_THREAD));
+
+        long startTime = System.currentTimeMillis();
+
+        // Monitor thread for EPS and memory
+        Thread monitor = new Thread(() -> {
+            long lastCount = 0;
+            long lastTime = System.currentTimeMillis();
+            Runtime runtime = Runtime.getRuntime();
+
+            while (counter.sum() < totalEvents.longValue()) {
+                try { Thread.sleep(200); } catch (InterruptedException ignored) {}
+
+                long now = System.currentTimeMillis();
+                long completed = counter.sum();
+                long eventsThisPeriod = completed - lastCount;
+                double eps = eventsThisPeriod / ((now - lastTime) / 1000.0);
+
+                long usedMemory = runtime.totalMemory() - runtime.freeMemory();
+                double usedPercent = usedMemory * 100.0 / runtime.maxMemory();
+
+                System.out.printf(
+                        "Progress: %d/%d (%.2f%%), EPS: %.0f, Memory Used: %.2f MB (%.2f%%)%n",
+                        completed,
+                        totalEvents.longValue(),
+                        completed * 100.0 / totalEvents.doubleValue(),
+                        eps,
+                        usedMemory / 1024.0 / 1024.0,
+                        usedPercent
+                );
+
+                lastCount = completed;
+                lastTime = now;
+            }
+        });
+        monitor.setDaemon(true);
+        monitor.start();
+
+        // Submit events asynchronously
+        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())
+                            .unsubscribeAfterSuccess()
+                            .asyncPostEvent();
+
+                    new EventPublisher<>(HeavyEventSuccess.class, "payload-" + i, a.getEventId())
+                            .asyncPostEvent();
+                }
+            });
+        }
+
+        executor.shutdown();
+        executor.awaitTermination(10, TimeUnit.MINUTES);
+
+        long endTime = System.currentTimeMillis();
+        double durationSeconds = (endTime - startTime) / 1000.0;
+
+        System.out.println("Posted " + totalEvents + " events in " + durationSeconds + " seconds");
+        double averageEps = totalEvents.doubleValue() / durationSeconds;
+        System.out.printf("Average EPS: %.0f%n", averageEps);
+
+        assertEquals(totalEvents.longValue(), counter.sum());
+    }
+
+
+    @Tag("stress")
+    @Test
+    void efficientExtremeConcurrencyTest() throws InterruptedException {
+        final int THREADS = Runtime.getRuntime().availableProcessors();
+        final int EVENTS_PER_THREAD = 5000;
+
+        ExecutorService executor = Executors.newFixedThreadPool(THREADS);
+        ConcurrentLinkedQueue<HeavyEvent> processedEvents = new ConcurrentLinkedQueue<>();
+
+        long start = System.nanoTime();
+
+        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)
+                            .postEvent();
+                }
+            });
+        }
+
+        executor.shutdown();
+        executor.awaitTermination(10, TimeUnit.MINUTES);
+
+        long end = System.nanoTime();
+        double durationSeconds = (end - start) / 1_000_000_000.0;
+
+        BigInteger totalEvents = BigInteger.valueOf((long) THREADS).multiply(BigInteger.valueOf(EVENTS_PER_THREAD));
+        double eps = totalEvents.doubleValue() / durationSeconds;
+
+        System.out.printf("Posted %s events in %.3f seconds%n", totalEvents, durationSeconds);
+        System.out.printf("Throughput: %.0f events/sec%n", eps);
+
+        Runtime rt = Runtime.getRuntime();
+        System.out.printf("Used memory: %.2f MB%n", (rt.totalMemory() - rt.freeMemory()) / 1024.0 / 1024.0);
+
+        assertEquals(totalEvents.intValue(), processedEvents.size());
+    }
+
+    @Tag("stress")
+    @Test
+    void constructorCacheVsReflection() throws Throwable {
+        int iterations = 1_000_000;
+        long startReflect = System.nanoTime();
+        for (int i = 0; i < iterations; i++) {
+            HeavyEvent.class.getDeclaredConstructors()[0].newInstance("payload", "uuid-" + i);
+        }
+        long endReflect = System.nanoTime();
+
+        long startHandle = System.nanoTime();
+        for (int i = 0; i < iterations; i++) {
+            EventPublisher<HeavyEvent> ep = new EventPublisher<>(HeavyEvent.class, "payload-" + i);
+        }
+        long endHandle = System.nanoTime();
+
+        System.out.println("Reflection: " + (endReflect - startReflect) / 1_000_000 + " ms");
+        System.out.println("MethodHandle Cache: " + (endHandle - startHandle) / 1_000_000 + " ms");
+    }
+}

framework/src/test/eventbus/EventPublisherTest.java

@@ -0,0 +1,126 @@
+package org.toop.eventbus;
+
+import org.junit.jupiter.api.Test;
+import org.toop.framework.eventbus.events.EventWithUuid;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class EventPublisherTest {
+
+    // Simple test event implementing EventWithUuid
+    public record TestEvent(String name, String eventId) implements EventWithUuid {
+        @Override
+        public Map<String, Object> result() {
+            return Map.of("name", name, "eventId", eventId);
+        }
+    }
+
+    public record TestResponseEvent(String msg, String eventId) implements EventWithUuid {
+        @Override
+        public Map<String, Object> result() {
+            return Map.of("msg", msg, "eventId", eventId);
+        }
+    }
+
+    @Test
+    void testEventPublisherGeneratesUuid() {
+        EventPublisher<TestEvent> publisher = new EventPublisher<>(TestEvent.class, "myTest");
+        assertNotNull(publisher.getEventId());
+        assertEquals(publisher.getEventId(), publisher.getEvent().eventId());
+    }
+
+    @Test
+    void testPostEvent() {
+        AtomicBoolean triggered = new AtomicBoolean(false);
+
+        EventPublisher<TestEvent> publisher = new EventPublisher<>(TestEvent.class, "myTest");
+        publisher.onEventById(TestEvent.class, event -> triggered.set(true))
+                .postEvent();
+
+        assertTrue(triggered.get(), "Subscriber should have been triggered by postEvent");
+    }
+
+    @Test
+    void testOnEventByIdMatchesUuid() {
+        AtomicBoolean triggered = new AtomicBoolean(false);
+
+        EventPublisher<TestEvent> publisher1 = new EventPublisher<>(TestEvent.class, "event1");
+        EventPublisher<TestEvent> 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<Object> listenerRef = new AtomicReference<>();
+
+        EventPublisher<TestEvent> 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<Map<String, Object>> resultRef = new AtomicReference<>();
+
+        EventPublisher<TestEvent> publisher = new EventPublisher<>(TestEvent.class, "myName");
+        publisher.onEventById(TestEvent.class, event -> resultRef.set(event.result()))
+                .postEvent();
+
+        Map<String, Object> 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<TestEvent> 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<TestEvent> publisher = new EventPublisher<>(TestEvent.class, "hello");
+        TestEvent event = publisher.getEvent();
+        assertNotNull(event);
+        assertEquals("hello", event.name());
+        assertEquals(publisher.getEventId(), event.eventId());
+    }
+}

framework/src/test/eventbus/GlobalEventBusTest.java

@@ -0,0 +1,110 @@
+//package org.toop.eventbus;
+//
+//import net.engio.mbassy.bus.publication.SyncAsyncPostCommand;
+//import org.junit.jupiter.api.AfterEach;
+//import org.junit.jupiter.api.Test;
+//import org.toop.eventbus.events.IEvent;
+//
+//import java.util.concurrent.atomic.AtomicBoolean;
+//import java.util.concurrent.atomic.AtomicReference;
+//
+//import static org.junit.jupiter.api.Assertions.*;
+//
+//class GlobalEventBusTest {
+//
+//    // A simple test event
+//    static class TestEvent implements IEvent {
+//        private final String message;
+//
+//        TestEvent(String message) {
+//            this.message = message;
+//        }
+//
+//        String getMessage() {
+//            return message;
+//        }
+//    }
+//
+//    @AfterEach
+//    void tearDown() {
+//        // Reset to avoid leaking subscribers between tests
+//        GlobalEventBus.reset();
+//    }
+//
+//    @Test
+//    void testSubscribeWithType() {
+//        AtomicReference<String> result = new AtomicReference<>();
+//
+//        GlobalEventBus.subscribe(TestEvent.class, e -> result.set(e.getMessage()));
+//
+//        GlobalEventBus.post(new TestEvent("hello"));
+//
+//        assertEquals("hello", result.get());
+//    }
+//
+//    @Test
+//    void testSubscribeWithoutType() {
+//        AtomicReference<String> result = new AtomicReference<>();
+//
+//        GlobalEventBus.subscribe((TestEvent e) -> result.set(e.getMessage()));
+//
+//        GlobalEventBus.post(new TestEvent("world"));
+//
+//        assertEquals("world", result.get());
+//    }
+//
+//    @Test
+//    void testUnsubscribeStopsReceivingEvents() {
+//        AtomicBoolean called = new AtomicBoolean(false);
+//
+//        Object listener = GlobalEventBus.subscribe(TestEvent.class, e -> called.set(true));
+//
+//        // First event should trigger
+//        GlobalEventBus.post(new TestEvent("first"));
+//        assertTrue(called.get());
+//
+//        // Reset flag
+//        called.set(false);
+//
+//        // Unsubscribe and post again
+//        GlobalEventBus.unsubscribe(listener);
+//        GlobalEventBus.post(new TestEvent("second"));
+//
+//        assertFalse(called.get(), "Listener should not be called after unsubscribe");
+//    }
+//
+//    @Test
+//    void testResetClearsListeners() {
+//        AtomicBoolean called = new AtomicBoolean(false);
+//
+//        GlobalEventBus.subscribe(TestEvent.class, e -> called.set(true));
+//
+//        GlobalEventBus.reset(); // should wipe subscriptions
+//
+//        GlobalEventBus.post(new TestEvent("ignored"));
+//
+//        assertFalse(called.get(), "Listener should not survive reset()");
+//    }
+
+//    @Test
+//    void testSetReplacesBus() {
+//        MBassadorMock<IEvent> mockBus = new MBassadorMock<>();
+//        GlobalEventBus.set(mockBus);
+//
+//        TestEvent event = new TestEvent("test");
+//        GlobalEventBus.post(event);
+//
+//        assertEquals(event, mockBus.lastPosted, "Custom bus should receive the event");
+//    }
+//
+//    // Minimal fake MBassador for verifying set()
+//    static class MBassadorMock<T extends IEvent> extends net.engio.mbassy.bus.MBassador<T> {
+//        T lastPosted;
+//
+//        @Override
+//        public SyncAsyncPostCommand<T> post(T message) {
+//            this.lastPosted = message;
+//            return super.post(message);
+//        }
+//    }
+//}