fix: #367 add immediate parameter to EventQueue.close() for forced queue shutdown

- Add immediate parameter to EventQueue.close() method to allow immediate
  queue closure by discarding all pending events
- Update EventConsumer.consume_all() to use close(immediate=True) when
  encountering the first final event to prevent blocking on subsequent
  final events
- This resolves the issue where multiple final events in queue would
  cause the consumer to block after processing the first final event

The immediate parameter provides a way to force close the queue without
waiting for all events to be processed, which is useful in scenarios
where the consumer needs to exit immediately upon encountering a final
event, regardless of remaining events in the queue.

Author: 严骏驰

src/a2a/server/events/event_consumer.py

@@ -125,8 +125,7 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                 # other part is waiting for an event or a closed queue.
                 if is_final_event:
                     logger.debug('Stopping event consumption in consume_all.')
-                    await self.queue.clear_events()
-                    await self.queue.close()
+                    await self.queue.close(True)
                     yield event
                     break
                 yield event

src/a2a/server/events/event_queue.py

@@ -127,7 +127,7 @@ def tap(self) -> 'EventQueue':
         self._children.append(queue)
         return queue
 
-    async def close(self) -> None:
+    async def close(self, immediate: bool = False) -> None:
         """Closes the queue for future push events.
 
         Once closed, `dequeue_event` will eventually raise `asyncio.QueueShutDown`
@@ -136,16 +136,20 @@ async def close(self) -> None:
         logger.debug('Closing EventQueue.')
         async with self._lock:
             # If already closed, just return.
-            if self._is_closed:
+            if self._is_closed and not immediate:
                 return
-            self._is_closed = True
+            if not self._is_closed:
+                self._is_closed = True
         # If using python 3.13 or higher, use the shutdown method
         if sys.version_info >= (3, 13):
-            self.queue.shutdown()
+            self.queue.shutdown(immediate)
             for child in self._children:
-                await child.close()
+                await child.close(immediate)
         # Otherwise, join the queue
         else:
+            if immediate:
+                await self.clear_events(True)
+                return
             tasks = [asyncio.create_task(self.queue.join())]
             for child in self._children:
                 tasks.append(asyncio.create_task(child.close()))