feat: Rework Event Queues to use central event bus

Author: kabir

extras/queue-manager-replicated/core/src/main/java/io/a2a/extras/queuemanager/replicated/core/ReplicatedQueueManager.java

@@ -17,6 +17,7 @@
 import io.a2a.server.events.EventQueueFactory;
 import io.a2a.server.events.EventQueueItem;
 import io.a2a.server.events.InMemoryQueueManager;
+import io.a2a.server.events.MainEventBus;
 import io.a2a.server.events.QueueManager;
 
 @ApplicationScoped
@@ -32,10 +33,12 @@ public class ReplicatedQueueManager implements QueueManager {
     private TaskStateProvider taskStateProvider;
 
     @Inject
-    public ReplicatedQueueManager(ReplicationStrategy replicationStrategy, TaskStateProvider taskStateProvider) {
+    public ReplicatedQueueManager(ReplicationStrategy replicationStrategy,
+                                    TaskStateProvider taskStateProvider,
+                                    MainEventBus mainEventBus) {
         this.replicationStrategy = replicationStrategy;
         this.taskStateProvider = taskStateProvider;
-        this.delegate = new InMemoryQueueManager(new ReplicatingEventQueueFactory(), taskStateProvider);
+        this.delegate = new InMemoryQueueManager(new ReplicatingEventQueueFactory(), taskStateProvider, mainEventBus);
     }
 
 
@@ -139,12 +142,11 @@ public EventQueue.EventQueueBuilder builder(String taskId) {
             // which sends the QueueClosedEvent after the database transaction commits.
             // This ensures proper ordering and transactional guarantees.
 
-            // Return the builder with callbacks
-            return delegate.getEventQueueBuilder(taskId)
-                    .taskId(taskId)
-                    .hook(new ReplicationHook(taskId))
-                    .addOnCloseCallback(delegate.getCleanupCallback(taskId))
-                    .taskStateProvider(taskStateProvider);
+            // Call createBaseEventQueueBuilder() directly to avoid infinite recursion
+            // (getEventQueueBuilder() would delegate back to this factory, creating a loop)
+            // The base builder already includes: taskId, cleanup callback, taskStateProvider, mainEventBus
+            return delegate.createBaseEventQueueBuilder(taskId)
+                    .hook(new ReplicationHook(taskId));
         }
     }
 

extras/queue-manager-replicated/core/src/test/java/io/a2a/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java

@@ -21,7 +21,12 @@
 import io.a2a.server.events.EventQueueClosedException;
 import io.a2a.server.events.EventQueueItem;
 import io.a2a.server.events.EventQueueTestHelper;
+import io.a2a.server.events.EventQueueUtil;
+import io.a2a.server.events.MainEventBus;
+import io.a2a.server.events.MainEventBusProcessor;
 import io.a2a.server.events.QueueClosedEvent;
+import io.a2a.server.tasks.InMemoryTaskStore;
+import io.a2a.server.tasks.PushNotificationSender;
 import io.a2a.spec.Event;
 import io.a2a.spec.StreamingEventKind;
 import io.a2a.spec.TaskState;
@@ -35,10 +40,24 @@ class ReplicatedQueueManagerTest {
 
     private ReplicatedQueueManager queueManager;
     private StreamingEventKind testEvent;
+    private MainEventBus mainEventBus;
+    private MainEventBusProcessor mainEventBusProcessor;
+    private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = task -> {};
 
     @BeforeEach
     void setUp() {
-        queueManager = new ReplicatedQueueManager(new NoOpReplicationStrategy(), new MockTaskStateProvider(true));
+        // Create MainEventBus and MainEventBusProcessor for tests
+        InMemoryTaskStore taskStore = new InMemoryTaskStore();
+        mainEventBus = new MainEventBus();
+        mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER);
+        EventQueueUtil.start(mainEventBusProcessor);
+
+        queueManager = new ReplicatedQueueManager(
+            new NoOpReplicationStrategy(),
+            new MockTaskStateProvider(true),
+            mainEventBus
+        );
+
         testEvent = new TaskStatusUpdateEvent.Builder()
                 .taskId("test-task")
                 .contextId("test-context")
@@ -50,7 +69,7 @@ void setUp() {
     @Test
     void testReplicationStrategyTriggeredOnNormalEnqueue() throws InterruptedException {
         CountingReplicationStrategy strategy = new CountingReplicationStrategy();
-        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true));
+        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true), mainEventBus);
 
         String taskId = "test-task-1";
         EventQueue queue = queueManager.createOrTap(taskId);
@@ -65,7 +84,7 @@ void testReplicationStrategyTriggeredOnNormalEnqueue() throws InterruptedExcepti
     @Test
     void testReplicationStrategyNotTriggeredOnReplicatedEvent() throws InterruptedException {
         CountingReplicationStrategy strategy = new CountingReplicationStrategy();
-        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true));
+        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true), mainEventBus);
 
         String taskId = "test-task-2";
         EventQueue queue = queueManager.createOrTap(taskId);
@@ -79,7 +98,7 @@ void testReplicationStrategyNotTriggeredOnReplicatedEvent() throws InterruptedEx
     @Test
     void testReplicationStrategyWithCountingImplementation() throws InterruptedException {
         CountingReplicationStrategy countingStrategy = new CountingReplicationStrategy();
-        queueManager = new ReplicatedQueueManager(countingStrategy, new MockTaskStateProvider(true));
+        queueManager = new ReplicatedQueueManager(countingStrategy, new MockTaskStateProvider(true), mainEventBus);
 
         String taskId = "test-task-3";
         EventQueue queue = queueManager.createOrTap(taskId);
@@ -170,7 +189,7 @@ void testBasicQueueManagerFunctionality() throws InterruptedException {
     void testQueueToTaskIdMappingMaintained() throws InterruptedException {
         String taskId = "test-task-6";
         CountingReplicationStrategy countingStrategy = new CountingReplicationStrategy();
-        queueManager = new ReplicatedQueueManager(countingStrategy, new MockTaskStateProvider(true));
+        queueManager = new ReplicatedQueueManager(countingStrategy, new MockTaskStateProvider(true), mainEventBus);
 
         EventQueue queue = queueManager.createOrTap(taskId);
         queue.enqueueEvent(testEvent);
@@ -217,7 +236,7 @@ void testReplicatedEventJsonSerialization() throws Exception {
     @Test
     void testParallelReplicationBehavior() throws InterruptedException {
         CountingReplicationStrategy strategy = new CountingReplicationStrategy();
-        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true));
+        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true), mainEventBus);
 
         String taskId = "parallel-test-task";
         EventQueue queue = queueManager.createOrTap(taskId);
@@ -297,7 +316,7 @@ void testParallelReplicationBehavior() throws InterruptedException {
     void testReplicatedEventSkippedWhenTaskInactive() throws InterruptedException {
         // Create a task state provider that returns false (task is inactive)
         MockTaskStateProvider stateProvider = new MockTaskStateProvider(false);
-        queueManager = new ReplicatedQueueManager(new CountingReplicationStrategy(), stateProvider);
+        queueManager = new ReplicatedQueueManager(new CountingReplicationStrategy(), stateProvider, mainEventBus);
 
         String taskId = "inactive-task";
 
@@ -316,7 +335,7 @@ void testReplicatedEventSkippedWhenTaskInactive() throws InterruptedException {
     void testReplicatedEventProcessedWhenTaskActive() throws InterruptedException {
         // Create a task state provider that returns true (task is active)
         MockTaskStateProvider stateProvider = new MockTaskStateProvider(true);
-        queueManager = new ReplicatedQueueManager(new CountingReplicationStrategy(), stateProvider);
+        queueManager = new ReplicatedQueueManager(new CountingReplicationStrategy(), stateProvider, mainEventBus);
 
         String taskId = "active-task";
 
@@ -347,7 +366,7 @@ void testReplicatedEventProcessedWhenTaskActive() throws InterruptedException {
     void testReplicatedEventToExistingQueueWhenTaskBecomesInactive() throws InterruptedException {
         // Create a task state provider that returns true initially
         MockTaskStateProvider stateProvider = new MockTaskStateProvider(true);
-        queueManager = new ReplicatedQueueManager(new CountingReplicationStrategy(), stateProvider);
+        queueManager = new ReplicatedQueueManager(new CountingReplicationStrategy(), stateProvider, mainEventBus);
 
         String taskId = "task-becomes-inactive";
 
@@ -387,7 +406,7 @@ void testReplicatedEventToExistingQueueWhenTaskBecomesInactive() throws Interrup
     @Test
     void testPoisonPillSentViaTransactionAwareEvent() throws InterruptedException {
         CountingReplicationStrategy strategy = new CountingReplicationStrategy();
-        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true));
+        queueManager = new ReplicatedQueueManager(strategy, new MockTaskStateProvider(true), mainEventBus);
 
         String taskId = "poison-pill-test";
         EventQueue queue = queueManager.createOrTap(taskId);

extras/queue-manager-replicated/core/src/test/java/io/a2a/server/events/EventQueueUtil.java

@@ -0,0 +1,16 @@
+package io.a2a.server.events;
+
+public class EventQueueUtil {
+    // Since EventQueue.builder() is package protected, add a method to expose it
+    public static EventQueue.EventQueueBuilder getEventQueueBuilder() {
+        return EventQueue.builder();
+    }
+
+    public static void start(MainEventBusProcessor processor) {
+        processor.start();
+    }
+
+    public static void stop(MainEventBusProcessor processor) {
+        processor.stop();
+    }
+}

reference/jsonrpc/src/test/resources/application.properties

@@ -1 +1,6 @@
 quarkus.arc.selected-alternatives=io.a2a.server.apps.common.TestHttpClient
+
+# Debug logging for event processing and request handling
+quarkus.log.category."io.a2a.server.events".level=DEBUG
+quarkus.log.category."io.a2a.server.requesthandlers".level=DEBUG
+quarkus.log.category."io.a2a.server.tasks".level=DEBUG

server-common/src/main/java/io/a2a/server/events/EventQueue.java

@@ -53,6 +53,7 @@ public static class EventQueueBuilder {
         private String taskId;
         private List<Runnable> onCloseCallbacks = new java.util.ArrayList<>();
         private TaskStateProvider taskStateProvider;
+        private MainEventBus mainEventBus;
 
         public EventQueueBuilder queueSize(int queueSize) {
             this.queueSize = queueSize;
@@ -81,9 +82,14 @@ public EventQueueBuilder taskStateProvider(TaskStateProvider taskStateProvider)
             return this;
         }
 
+        public EventQueueBuilder mainEventBus(MainEventBus mainEventBus) {
+            this.mainEventBus = mainEventBus;
+            return this;
+        }
+
         public EventQueue build() {
-            if (hook != null || !onCloseCallbacks.isEmpty() || taskStateProvider != null) {
-                return new MainQueue(queueSize, hook, taskId, onCloseCallbacks, taskStateProvider);
+            if (hook != null || !onCloseCallbacks.isEmpty() || taskStateProvider != null || mainEventBus != null) {
+                return new MainQueue(queueSize, hook, taskId, onCloseCallbacks, taskStateProvider, mainEventBus);
             } else {
                 return new MainQueue(queueSize);
             }
@@ -222,13 +228,15 @@ static class MainQueue extends EventQueue {
         private final String taskId;
         private final List<Runnable> onCloseCallbacks;
         private final TaskStateProvider taskStateProvider;
+        private final MainEventBus mainEventBus;
 
         MainQueue() {
             super();
             this.enqueueHook = null;
             this.taskId = null;
             this.onCloseCallbacks = List.of();
             this.taskStateProvider = null;
+            this.mainEventBus = null;
         }
 
         MainQueue(int queueSize) {
@@ -237,6 +245,7 @@ static class MainQueue extends EventQueue {
             this.taskId = null;
             this.onCloseCallbacks = List.of();
             this.taskStateProvider = null;
+            this.mainEventBus = null;
         }
 
         MainQueue(EventEnqueueHook hook) {
@@ -245,6 +254,7 @@ static class MainQueue extends EventQueue {
             this.taskId = null;
             this.onCloseCallbacks = List.of();
             this.taskStateProvider = null;
+            this.mainEventBus = null;
         }
 
         MainQueue(int queueSize, EventEnqueueHook hook) {
@@ -253,6 +263,7 @@ static class MainQueue extends EventQueue {
             this.taskId = null;
             this.onCloseCallbacks = List.of();
             this.taskStateProvider = null;
+            this.mainEventBus = null;
         }
 
         MainQueue(int queueSize, EventEnqueueHook hook, String taskId, List<Runnable> onCloseCallbacks, TaskStateProvider taskStateProvider) {
@@ -261,16 +272,36 @@ static class MainQueue extends EventQueue {
             this.taskId = taskId;
             this.onCloseCallbacks = List.copyOf(onCloseCallbacks);  // Defensive copy
             this.taskStateProvider = taskStateProvider;
+            this.mainEventBus = null;
             LOGGER.debug("Created MainQueue for task {} with {} onClose callbacks and TaskStateProvider: {}",
                     taskId, onCloseCallbacks.size(), taskStateProvider != null);
         }
 
+        MainQueue(int queueSize, EventEnqueueHook hook, String taskId, List<Runnable> onCloseCallbacks, TaskStateProvider taskStateProvider, MainEventBus mainEventBus) {
+            super(queueSize);
+            this.enqueueHook = hook;
+            this.taskId = taskId;
+            this.onCloseCallbacks = List.copyOf(onCloseCallbacks);  // Defensive copy
+            this.taskStateProvider = taskStateProvider;
+            this.mainEventBus = mainEventBus;
+            LOGGER.debug("Created MainQueue for task {} with {} onClose callbacks, TaskStateProvider: {}, MainEventBus: {}",
+                    taskId, onCloseCallbacks.size(), taskStateProvider != null, mainEventBus != null);
+        }
+
         public EventQueue tap() {
             ChildQueue child = new ChildQueue(this);
             children.add(child);
             return child;
         }
 
+        /**
+         * Returns the current number of child queues.
+         * Useful for debugging and logging event distribution.
+         */
+        public int getChildCount() {
+            return children.size();
+        }
+
         @Override
         public void enqueueItem(EventQueueItem item) {
             // MainQueue must accept events even when closed to support:
@@ -293,10 +324,15 @@ public void enqueueItem(EventQueueItem item) {
             queue.add(item);
             LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this);
 
-            // Distribute to all ChildQueues (they will receive the event even if MainQueue is closed)
-            children.forEach(eq -> eq.internalEnqueueItem(item));
+            // Submit to MainEventBus for centralized persistence + distribution
+            if (mainEventBus != null && taskId != null) {
+                mainEventBus.submit(taskId, this, item);
+            } else {
+                // This should not happen in properly configured systems
+                LOGGER.error("MainEventBus not configured for task {} - events will NOT be distributed to children!", taskId);
+            }
 
-            // Trigger replication hook if configured
+            // Trigger replication hook if configured (KEEP for inter-process replication)
             if (enqueueHook != null) {
                 enqueueHook.onEnqueue(item);
             }
@@ -350,6 +386,29 @@ void childClosing(ChildQueue child, boolean immediate) {
             this.doClose(immediate);
         }
 
+        /**
+         * Distribute event to all ChildQueues.
+         * Called by MainEventBusProcessor after TaskStore persistence.
+         */
+        void distributeToChildren(EventQueueItem item) {
+            synchronized (children) {
+                int childCount = children.size();
+                if (LOGGER.isDebugEnabled()) {
+                    LOGGER.debug("MainQueue[{}]: Distributing event {} to {} children",
+                                taskId, item.getEvent().getClass().getSimpleName(), childCount);
+                }
+                children.forEach(child -> {
+                    LOGGER.debug("MainQueue[{}]: Enqueueing event {} to child queue",
+                                taskId, item.getEvent().getClass().getSimpleName());
+                    child.internalEnqueueItem(item);
+                });
+                if (LOGGER.isDebugEnabled()) {
+                    LOGGER.debug("MainQueue[{}]: Completed distribution of {} to {} children",
+                                taskId, item.getEvent().getClass().getSimpleName(), childCount);
+                }
+            }
+        }
+
         /**
          * Get the count of active child queues.
          * Used for testing to verify reference counting mechanism.