[AI-3] Add parallel async subscribers in EventBroker

README.md

@@ -295,6 +295,8 @@ async def context():
   await handle_event(product_event)
 ```
 
+> **Note**: Async event handlers are executed in parallel using `asyncio.gather`. Sync handlers are called sequentially.
+
 ### Stories
 
 Stories provide a pattern for defining sequential business operations with optional hooks for execution tracking,

src/dddkit/dataclasses/events.py

@@ -1,3 +1,4 @@
+import asyncio
 from asyncio import get_running_loop
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
@@ -84,11 +85,15 @@
             handler(event)
 
     async def async_publish(self, event: DomainEvent) -> None:
-        for handler in self._get_subscribers(event):
+        handlers = self._get_subscribers(event)
+        async_handlers = []
+        for handler in handlers:
             if iscoroutinefunction(handler):
-                await handler(event)
+                async_handlers.append(handler(event))
             else:
                 handler(event)
+        if async_handlers:
+            await asyncio.gather(*async_handlers)
 
     def instance(self, obj_type: type[ET] | tuple[type[ET], ...] | None) -> Callable[[HandlerEvent], HandlerEvent]:
         _type = obj_type if obj_type is not None else type(None)

src/dddkit/pydantic/events.py

@@ -1,3 +1,4 @@
+import asyncio
 from asyncio import get_running_loop
 from collections.abc import Awaitable, Callable
 from datetime import datetime
@@ -86,11 +87,15 @@
             handler(event)
 
     async def async_publish(self, event: DomainEvent) -> None:
-        for handler in self._get_subscribers(event):
+        handlers = self._get_subscribers(event)
+        async_handlers = []
+        for handler in handlers:
             if iscoroutinefunction(handler):
-                await handler(event)
+                async_handlers.append(handler(event))
             else:
                 handler(event)
+        if async_handlers:
+            await asyncio.gather(*async_handlers)
 
     def instance(self, obj_type: type[ET] | tuple[type[ET], ...] | None) -> Callable[[HandlerEvent], HandlerEvent]:
         _type = obj_type if obj_type is not None else type(None)