Move sepaphore into MainQueue

Author: kabir

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

@@ -21,7 +21,6 @@ public abstract class EventQueue implements AutoCloseable {
     public static final int DEFAULT_QUEUE_SIZE = 1000;
 
     private final int queueSize;
-    protected final Semaphore semaphore;
     private volatile boolean closed = false;
 
     protected EventQueue() {
@@ -33,7 +32,6 @@ protected EventQueue(int queueSize) {
             throw new IllegalArgumentException("Queue size must be greater than 0");
         }
         this.queueSize = queueSize;
-        this.semaphore = new Semaphore(queueSize, true);
         LOGGER.trace("Creating {} with queue size: {}", this, queueSize);
     }
 
@@ -179,6 +177,7 @@ protected void doClose(boolean immediate) {
 
     static class MainQueue extends EventQueue {
         private final List<ChildQueue> children = new CopyOnWriteArrayList<>();
+        protected final Semaphore semaphore;
         private final CountDownLatch pollingStartedLatch = new CountDownLatch(1);
         private final AtomicBoolean pollingStarted = new AtomicBoolean(false);
         private final EventEnqueueHook enqueueHook;
@@ -189,6 +188,7 @@ static class MainQueue extends EventQueue {
 
         MainQueue(int queueSize, EventEnqueueHook hook, String taskId, List<Runnable> onCloseCallbacks, TaskStateProvider taskStateProvider, MainEventBus mainEventBus) {
             super(queueSize);
+            this.semaphore = new Semaphore(queueSize, true);
             this.enqueueHook = hook;
             this.taskId = taskId;
             this.onCloseCallbacks = List.copyOf(onCloseCallbacks);  // Defensive copy
@@ -406,8 +406,12 @@ private void internalEnqueueItem(EventQueueItem item) {
                 LOGGER.warn("ChildQueue is closed. Event will not be enqueued. {} {}", this, event);
                 return;
             }
-            queue.add(item);
-            LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this);
+            if (!queue.offer(item)) {
+                  LOGGER.warn("ChildQueue {} is full. Closing immediately.", this);
+                  close(true); // immediate close
+            } else {
+                LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this);
+            }
         }
 
         @Override
@@ -422,7 +426,6 @@ public EventQueueItem dequeueEventItem(int waitMilliSeconds) throws EventQueueCl
                     if (item != null) {
                         Event event = item.getEvent();
                         LOGGER.debug("Dequeued event item (no wait) {} {}", this, event instanceof Throwable ? event.toString() : event);
-                        // Note: Semaphore is managed by parent MainQueue, not released here
                     }
                     return item;
                 }
@@ -432,7 +435,6 @@ public EventQueueItem dequeueEventItem(int waitMilliSeconds) throws EventQueueCl
                     if (item != null) {
                         Event event = item.getEvent();
                         LOGGER.debug("Dequeued event item (waiting) {} {}", this, event instanceof Throwable ? event.toString() : event);
-                        // Note: Semaphore is managed by parent MainQueue, not released here
                     } else {
                         LOGGER.trace("Dequeue timeout (null) from ChildQueue {}", System.identityHashCode(this));
                     }
@@ -475,8 +477,6 @@ protected void doClose(boolean immediate) {
                 // Immediate close: clear pending events from local queue
                 int clearedCount = queue.size();
                 queue.clear();
-                // Release semaphore permits for cleared events to prevent deadlock
-                semaphore.release(clearedCount);
                 LOGGER.debug("Cleared {} events from ChildQueue for immediate close: {}", clearedCount, this);
             }
             // For graceful close, let the queue drain naturally through normal consumption