Fix event processing documentation

Author: Вадим Козыревский

docs/event_handler/event_flow.md

@@ -25,24 +25,31 @@ sequenceDiagram
     participant Mediator
     participant Handler as Command Handler
     participant Events as Events Collection
-    participant Dispatcher as Event Dispatcher
-    participant Handlers as Event Handlers
+    participant Processor as Event Processor
     participant Emitter as Event Emitter
+    participant Handlers as Event Handlers
+    participant Broker as Message Broker
 
     Client->>Mediator: 1. Send Command
     Mediator->>Handler: 2. Execute Handler
     Handler->>Handler: 3. Business Logic
     Handler->>Events: 4. Collect Events
     Handler-->>Mediator: 5. Return Response
     
-    Mediator->>Dispatcher: 6. Process Events
-    Dispatcher->>Handlers: 7. Execute Handlers
-    Handlers-->>Dispatcher: 8. Complete
+    Mediator->>Processor: 6. Emit Events
+    Processor->>Emitter: 7. Emit Each Event
     
-    Mediator->>Emitter: 9. Emit Events
-    Emitter->>Emitter: 10. Send to Broker/Handlers
+    alt DomainEvent
+        Emitter->>Handlers: 8. Execute Event Handlers
+        Handlers-->>Emitter: 9. Complete
+    else NotificationEvent
+        Emitter->>Broker: 8. Send to Message Broker
+        Broker-->>Emitter: 9. Complete
+    end
     
-    Mediator-->>Client: 11. Return Response
+    Emitter-->>Processor: 10. Complete
+    Processor-->>Mediator: 11. Complete
+    Mediator-->>Client: 12. Return Response
 ```
 
 ### Detailed Event Processing Flow
@@ -54,27 +61,34 @@ graph TD
     C -->|Has Events?| D{Events Exist?}
     
     D -->|No| E[Return Response]
-    D -->|Yes| F[Process Events Parallel]
+    D -->|Yes| F[EventProcessor.emit_events]
+    
+    F -->|For Each Event| G{Parallel Enabled?}
+    
+    G -->|No| H[Sequential: EventEmitter.emit]
+    G -->|Yes| I[Parallel: Create Task with Semaphore]
+    I --> J[EventEmitter.emit]
     
-    F -->|For Each Event| G[EventDispatcher]
-    G -->|Find Handlers| H[EventMap Lookup]
-    H -->|Resolve Handler| I[DI Container]
-    I -->|Execute| J[Event Handler]
-    J -->|Complete| K{More Events?}
+    H --> K{Event Type?}
+    J --> K
     
-    K -->|Yes| F
-    K -->|No| L[Emit Events]
+    K -->|DomainEvent| L[EventEmitter: Find Handlers]
+    K -->|NotificationEvent| M[EventEmitter: Send to Broker]
     
-    L -->|DomainEvent| M[Process via Handlers]
-    L -->|NotificationEvent| N[Send to Broker]
+    L --> N[EventMap Lookup]
+    N --> O[Resolve Handler from DI]
+    O --> P[Execute Event Handler]
+    P --> Q{More Events?}
     
-    M --> E
-    N --> E
+    M --> Q
+    Q -->|Yes| F
+    Q -->|No| E
     
     style A fill:#e1f5ff
     style B fill:#fff3e0
-    style G fill:#c8e6c9
-    style J fill:#c8e6c9
+    style F fill:#c8e6c9
+    style L fill:#c8e6c9
+    style P fill:#c8e6c9
     style E fill:#f3e5f5
 ```
 
@@ -102,42 +116,54 @@ class JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
         )
 ```
 
-### 2. Event Dispatch
+### 2. Event Emission
 
-After the command handler completes, the mediator collects events and dispatches them:
+After the command handler completes, the mediator collects events and emits them through EventProcessor:
 
 ```python
 dispatch_result = await self._dispatcher.dispatch(request)
 
-if dispatch_result.events:
-    # Process events (parallel or sequential)
-    await self._process_events_parallel(dispatch_result.events.copy())
-    # Emit events to broker or handlers
-    await self._send_events(dispatch_result.events.copy())
+# Events are emitted through EventProcessor
+# EventProcessor uses EventEmitter which handles:
+# - DomainEvent: processes via event handlers (in-process)
+# - NotificationEvent: sends to message broker
+await self._event_processor.emit_events(dispatch_result.events)
 ```
 
-### 3. Event Processing
+The `EventProcessor` handles parallel or sequential processing based on configuration, and `EventEmitter` routes events to appropriate handlers or message brokers.
 
-Events are processed through `EventDispatcher`, which finds registered handlers and executes them:
+### 3. Event Processing via EventEmitter
+
+Events are processed through `EventEmitter`, which routes them based on event type:
 
 ```mermaid
 graph TD
-    A[EventDispatcher.dispatch] -->|1. Get Event Type| B[EventMap.get]
-    B -->|2. Find Handlers| C{Handlers Found?}
-    C -->|No| D[Log Warning]
-    C -->|Yes| E[Loop Through Handlers]
-    E -->|3. Resolve Handler| F[DI Container]
-    F -->|4. Execute Handler| G[Handler.handle]
-    G -->|5. Process Side Effects| H[Complete]
+    A[EventEmitter.emit] -->|1. Get Event Type| B{Event Type?}
+    
+    B -->|DomainEvent| C[EventMap.get]
+    C -->|2. Find Handlers| D{Handlers Found?}
+    D -->|No| E[Log Warning]
+    D -->|Yes| F[Loop Through Handlers]
+    F -->|3. Resolve Handler| G[DI Container]
+    G -->|4. Execute Handler| H[Handler.handle]
+    H -->|5. Process Side Effects| I[Complete]
+    
+    B -->|NotificationEvent| J{Message Broker?}
+    J -->|No| K[Raise RuntimeError]
+    J -->|Yes| L[Send to Message Broker]
+    L --> I
     
     style A fill:#e1f5ff
-    style G fill:#c8e6c9
-    style H fill:#fff3e0
+    style H fill:#c8e6c9
+    style I fill:#fff3e0
 ```
 
-### 4. Event Emission
+### 4. Event Routing
+
+`EventEmitter` automatically routes events based on their type:
 
-After processing, events are emitted through `EventEmitter`:
+- **DomainEvent** — Processed by event handlers registered in EventMap (in-process, synchronous)
+- **NotificationEvent** — Sent to message broker (Kafka, RabbitMQ, etc.) for asynchronous processing
 
-- **DomainEvent** — Processed by event handlers (in-process)
-- **NotificationEvent** — Sent to message broker (Kafka, RabbitMQ, etc.)
+!!! important "Single Processing"
+    Events are processed **only once** through EventEmitter. There is no duplicate processing - DomainEvents are handled by event handlers, and NotificationEvents are sent to message brokers.

docs/event_handler/parallel_processing.md

@@ -24,22 +24,32 @@ Events can be processed in parallel to improve performance. This is controlled b
 
 ```mermaid
 graph TD
-    Start[Process Events] --> CheckEnable{Parallel Enabled?}
+    Start[EventProcessor.emit_events] --> CheckEnable{Parallel Enabled?}
     
     CheckEnable -->|No| Sequential[Sequential Processing]
     Sequential --> LoopSeq[For Each Event]
-    LoopSeq --> ProcessSeq[Process Event]
-    ProcessSeq --> NextSeq{More Events?}
+    LoopSeq --> EmitSeq[EventEmitter.emit]
+    EmitSeq --> RouteSeq{Route Event}
+    RouteSeq -->|DomainEvent| HandlerSeq[Execute Handlers]
+    RouteSeq -->|NotificationEvent| BrokerSeq[Send to Broker]
+    HandlerSeq --> NextSeq{More Events?}
+    BrokerSeq --> NextSeq
     NextSeq -->|Yes| LoopSeq
     NextSeq -->|No| End1[End]
     
     CheckEnable -->|Yes| Parallel[Parallel Processing]
-    Parallel --> CreateTasks[Create Tasks for Each Event]
-    CreateTasks --> Semaphore[Acquire Semaphore]
-    Semaphore --> ProcessPar[Process Event]
-    ProcessPar --> ReleaseSem[Release Semaphore]
-    ReleaseSem --> Gather[asyncio.gather All Tasks]
-    Gather --> End2[End]
+    Parallel --> LoopPar[For Each Event]
+    LoopPar --> CreateTask[Create Task]
+    CreateTask --> Semaphore[Acquire Semaphore]
+    Semaphore --> EmitPar[EventEmitter.emit]
+    EmitPar --> RoutePar{Route Event}
+    RoutePar -->|DomainEvent| HandlerPar[Execute Handlers]
+    RoutePar -->|NotificationEvent| BrokerPar[Send to Broker]
+    HandlerPar --> ReleaseSem[Release Semaphore]
+    BrokerPar --> ReleaseSem
+    ReleaseSem --> NextPar{More Events?}
+    NextPar -->|Yes| LoopPar
+    NextPar -->|No| End2[End]
     
     style Start fill:#e1f5ff
     style Sequential fill:#fff3e0
@@ -49,43 +59,44 @@ graph TD
 
 ### Implementation
 
+The `EventProcessor` handles parallel or sequential event emission:
+
 ```python
-class RequestMediator:
+class EventProcessor:
     def __init__(
         self,
+        event_map: EventMap,
+        event_emitter: EventEmitter | None = None,
         max_concurrent_event_handlers: int = 1,
         concurrent_event_handle_enable: bool = True,
     ):
-        # Create semaphore to limit concurrency
-        self._event_semaphore = asyncio.Semaphore(max_concurrent_event_handlers)
+        self._event_emitter = event_emitter
+        self._max_concurrent_event_handlers = max_concurrent_event_handlers
         self._concurrent_event_handle_enable = concurrent_event_handle_enable
+        self._event_semaphore = asyncio.Semaphore(max_concurrent_event_handlers)
 
-    async def _process_event_with_semaphore(self, event: Event) -> None:
-        """Process a single event with semaphore limit."""
-        async with self._event_semaphore:
-            await self._event_dispatcher.dispatch(event)
-    
-    async def _process_events_parallel(
-        self,
-        events: List[Event],
-    ) -> None:
-        """Process events in parallel with semaphore limit or sequentially."""
-        if not events:
+    async def emit_events(self, events: List[Event]) -> None:
+        """Emit events via event emitter (parallel or sequential)."""
+        if not events or not self._event_emitter:
             return
 
         if not self._concurrent_event_handle_enable:
             # Sequential processing
             for event in events:
-                await self._event_dispatcher.dispatch(event)
+                await self._event_emitter.emit(event)
         else:
-            # Parallel processing with semaphore limit
-            tasks = [
-                self._process_event_with_semaphore(event) 
-                for event in events
-            ]
-            await asyncio.gather(*tasks)
+            # Parallel processing with semaphore limit (fire-and-forget)
+            for event in events:
+                asyncio.create_task(self._emit_event_with_semaphore(event))
+    
+    async def _emit_event_with_semaphore(self, event: Event) -> None:
+        """Emit a single event with semaphore limit."""
+        async with self._event_semaphore:
+            await self._event_emitter.emit(event)
 ```
 
+The `EventEmitter` then routes events to handlers or message brokers based on event type.
+
 ### Configuration
 
 ```python
@@ -129,7 +140,9 @@ class ProcessOrderCommandHandler(RequestHandler[ProcessOrderCommand, None]):
         self._events.append(EmailNotificationEvent(...))
 
 # With max_concurrent_event_handlers=3:
-# - Events 1-3 process in parallel
-# - Event 4 waits for a slot
-# - All events complete before response is returned
+# - Events 1-3 emit in parallel (fire-and-forget tasks)
+# - Event 4 waits for a semaphore slot
+# - Each event is routed by EventEmitter:
+#   - DomainEvents → processed by handlers
+#   - NotificationEvents → sent to message broker
 ```

docs/event_handler/runtime_processing.md

@@ -40,17 +40,19 @@ await mediator.send(JoinMeetingCommand(user_id="123", meeting_id="456"))
 # What happens:
 # 1. Command handler executes (synchronously)
 # 2. Events are collected from handler.events
-# 3. Events are processed by handlers (synchronously, in parallel if enabled)
-# 4. Events are emitted (synchronously)
-# 5. Response is returned
+# 3. Events are emitted through EventProcessor (synchronously, in parallel if enabled)
+#    - EventProcessor uses EventEmitter which routes events:
+#      - DomainEvent → processed by event handlers (in-process)
+#      - NotificationEvent → sent to message broker
+# 4. Response is returned
 ```
 
 
-The `EventDispatcher` is responsible for routing events to their handlers:
+The `EventEmitter` routes events to their handlers or message brokers. For DomainEvents, it uses EventDispatcher logic internally:
 
 ```mermaid
 graph TD
-    Start[EventDispatcher.dispatch] --> GetType[Get Event Type]
+    Start[EventEmitter.emit DomainEvent] --> GetType[Get Event Type]
     GetType --> Lookup[Lookup in EventMap]
     Lookup --> Found{Handlers Found?}
     
@@ -71,29 +73,29 @@ graph TD
     style Execute fill:#c8e6c9
 ```
 
-### Dispatcher Implementation
+### EventEmitter Implementation for DomainEvents
 
 ```python
-class EventDispatcher:
-    async def dispatch(self, event: Event) -> None:
+class EventEmitter:
+    @emit.register
+    async def _(self, event: DomainEvent) -> None:
         # 1. Find handlers for event type
-        handler_types = self._event_map.get(type(event), [])
+        handlers_types = self._event_map.get(type(event), [])
 
-        if not handler_types:
+        if not handlers_types:
             logger.warning(f"Handlers for event {type(event).__name__} not found")
             return
 
         # 2. Process each handler
-        for handler_type in handler_types:
+        for handler_type in handlers_types:
             # 3. Resolve handler from DI container
             handler = await self._container.resolve(handler_type)
 
             # 4. Execute handler
             await handler.handle(event)
 ```
 
-
-The `EventEmitter` is responsible for emitting events after processing:
+The `EventEmitter` is responsible for emitting events and routing them based on type:
 
 ```mermaid
 graph TD