Refactor style

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

docs/event_handler/best_practices.md

@@ -23,6 +23,7 @@
 | **Limit concurrency** | Set appropriate `max_concurrent_event_handlers` based on resources | Resource management |
 | **Idempotency** | Make event handlers idempotent when possible | Reliability |
 | **Logging** | Log important events for debugging and monitoring | Observability |
+| **Follow-up events** | Use `handler.events` for multi-level chains; follow-ups run in the same pipeline (BFS or parallel) | Design clarity |
 
 !!! warning "Performance Considerations"
     Event handlers execute synchronously in the request context. Keep them fast to avoid blocking the main request flow.
@@ -35,6 +36,7 @@ Event handling in `python-cqrs`:
 
 - **Runtime Processing** — Events are processed synchronously in the same request context
 - **Automatic Dispatch** — Events are automatically dispatched to registered handlers
+- **Event Propagation** — Handlers can return follow-up events via `events`; they are processed in the same pipeline (BFS or parallel with semaphore)
 - **Parallel Support** — Multiple events can be processed in parallel with configurable limits
 - **Two Types** — DomainEvent (in-process) and NotificationEvent (message broker)
 - **Side Effects** — Event handlers perform side effects without blocking command execution

docs/event_handler/event_flow.md

@@ -41,17 +41,22 @@ sequenceDiagram
     
     alt DomainEvent
         Emitter->>Handlers: 8. Execute Event Handlers
-        Handlers-->>Emitter: 9. Complete
+        Handlers-->>Emitter: 9. handler.events (follow-ups)
+        Emitter-->>Processor: 10. Return follow-ups
+        Note over Processor: Process follow-ups in same pipeline (BFS or parallel)
     else NotificationEvent
         Emitter->>Broker: 8. Send to Message Broker
         Broker-->>Emitter: 9. Complete
+        Emitter-->>Processor: 10. No follow-ups
     end
     
-    Emitter-->>Processor: 10. Complete
     Processor-->>Mediator: 11. Complete
     Mediator-->>Client: 12. Return Response
 ```
 
+!!! note "Follow-up events"
+    For domain events, handlers can return follow-up events via the `events` property. The processor continues emitting these in the same pipeline (sequential BFS or parallel with semaphore) until the queue is empty.
+
 ### Detailed Event Processing Flow
 
 ```mermaid
@@ -63,32 +68,35 @@ graph TD
     D -->|No| E[Return Response]
     D -->|Yes| F[EventProcessor.emit_events]
     
-    F -->|For Each Event| G{Parallel Enabled?}
+    F -->|Queue: initial events| G{Parallel Enabled?}
+    
+    G -->|No| H[Sequential: BFS]
+    G -->|Yes| I[Parallel: Semaphore + FIRST_COMPLETED]
     
-    G -->|No| H[Sequential: EventEmitter.emit]
-    G -->|Yes| I[Parallel: Create Task with Semaphore]
-    I --> J[EventEmitter.emit]
+    H --> J[Pop Event from Queue]
+    I --> J
     
-    H --> K{Event Type?}
-    J --> K
+    J --> K[EventEmitter.emit]
+    K --> L{Event Type?}
     
-    K -->|DomainEvent| L[EventEmitter: Find Handlers]
-    K -->|NotificationEvent| M[EventEmitter: Send to Broker]
+    L -->|DomainEvent| M[EventMap Lookup]
+    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 --> O[Resolve Handler from DI]
+    O --> P[Execute handler.handle]
+    P --> Q[Collect handler.events - follow-ups]
+    Q --> R[Return follow-ups to Processor]
     
-    M --> Q
-    Q -->|Yes| F
-    Q -->|No| E
+    N --> R
+    R --> S{More in Queue?}
+    S -->|Yes| G
+    S -->|No| E
     
     style A fill:#e1f5ff
     style B fill:#fff3e0
     style F fill:#c8e6c9
-    style L fill:#c8e6c9
     style P fill:#c8e6c9
+    style Q fill:#fff9c4
     style E fill:#f3e5f5
 ```
 
@@ -134,7 +142,7 @@ The `EventProcessor` handles parallel or sequential processing based on configur
 
 ### 3. Event Processing via EventEmitter
 
-Events are processed through `EventEmitter`, which routes them based on event type:
+Events are processed through `EventEmitter`, which routes them based on event type. For domain events, after each handler runs, follow-up events from `handler.events` are collected and returned; the processor then continues with these in the same pipeline (BFS in sequential mode, or under the same semaphore in parallel mode).
 
 ```mermaid
 graph TD
@@ -145,25 +153,54 @@ graph TD
     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]
+    G -->|4. Execute handler.handle| H[Handler.handle]
+    H -->|5. Collect handler.events| I[Follow-up events]
+    I --> J[Return follow-ups to Processor]
     
-    B -->|NotificationEvent| J{Message Broker?}
-    J -->|No| K[Raise RuntimeError]
-    J -->|Yes| L[Send to Message Broker]
-    L --> I
+    B -->|NotificationEvent| K{Message Broker?}
+    K -->|No| L[Raise RuntimeError]
+    K -->|Yes| M[Send to Message Broker]
+    M --> N[Return empty - no follow-ups]
     
     style A fill:#e1f5ff
     style H fill:#c8e6c9
-    style I fill:#fff3e0
+    style I fill:#fff9c4
+    style J fill:#fff3e0
+```
+
+### 3.1. Follow-up events from event handlers (event propagation)
+
+Event handlers can produce **follow-up events** by implementing the `events` property. After `handle()` is called, the emitter reads `handler.events` and returns them to the processor. These follow-ups are processed in the **same pipeline**:
+
+| Mode | Behavior |
+|------|----------|
+| **Sequential** (`concurrent_event_handle_enable=False`) | Events and follow-ups are processed in **BFS order**: one event at a time, then its follow-ups are appended to the queue. |
+| **Parallel** (`concurrent_event_handle_enable=True`) | Events are processed under a semaphore; as soon as any task completes, its follow-ups are queued and started (FIRST_COMPLETED), without waiting for sibling events. |
+
+This allows **multi-level event chains**: e.g. `OrderCreated` → handler emits `InventoryReserved` → handler emits `NotificationScheduled`, all in one run.
+
+Example: handler that produces a follow-up event:
+
+```python
+class OrderCreatedEventHandler(cqrs.EventHandler[OrderCreatedEvent]):
+    def __init__(self) -> None:
+        self._follow_ups: list[cqrs.IEvent] = []
+
+    @property
+    def events(self) -> typing.Sequence[cqrs.IEvent]:
+        return tuple(self._follow_ups)
+
+    async def handle(self, event: OrderCreatedEvent) -> None:
+        # Side effects...
+        self._follow_ups.append(InventoryReservedEvent(order_id=event.order_id))
 ```
 
 ### 4. Event Routing
 
 `EventEmitter` automatically routes events based on their type:
 
-- **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 registered in EventMap (in-process). Handlers may return follow-up events via the `events` property; these are processed in the same pipeline (BFS or parallel with semaphore).
+- **NotificationEvent** — Sent to message broker (Kafka, RabbitMQ, etc.) for asynchronous processing; no follow-ups.
 
 !!! 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.
+    Each event instance is processed **only once** through EventEmitter. Follow-up events returned by handlers are **new** events that are then processed in the same run (same pipeline) until the queue is empty.

docs/event_handler/event_types.md

@@ -22,7 +22,7 @@
 
 ### DomainEvent
 
-Domain events represent something that happened in the domain. They are processed by event handlers:
+Domain events represent something that happened in the domain. They are processed by event handlers. Handlers can return **follow-up events** via the `events` property; these are processed in the same pipeline (see [Event Flow](event_flow.md)).
 
 ```python
 class UserJoined(cqrs.DomainEvent, frozen=True):
@@ -32,6 +32,7 @@ class UserJoined(cqrs.DomainEvent, frozen=True):
 class UserJoinedEventHandler(cqrs.EventHandler[UserJoined]):
     async def handle(self, event: UserJoined) -> None:
         # Process domain event
+        # Optionally populate self._follow_ups and return via events property
         ...
 ```
 

docs/event_handler/examples.md

@@ -78,7 +78,68 @@ await mediator.send(JoinMeetingCommand(user_id="123", meeting_id="456"))
 # Flow:
 # 1. Command handler executes
 # 2. UserJoined event is collected
-# 3. EventDispatcher processes event (finds UserJoinedEventHandler)
+# 3. EventProcessor.emit_events runs (EventEmitter finds UserJoinedEventHandler)
 # 4. UserJoinedEventHandler.handle() executes
 # 5. Response is returned
 ```
+
+---
+
+## Event handler chain (follow-up events)
+
+Event handlers can produce **follow-up events** via the `events` property. These are processed in the same pipeline (BFS in sequential mode, or under the same semaphore in parallel mode), enabling multi-level chains (e.g. L1 → L2 → L3).
+
+```python
+import typing
+import cqrs
+from cqrs.events.event import IEvent
+
+# Level 1: emitted by command handler
+class EventL1(cqrs.DomainEvent, frozen=True):
+    seed: str
+
+# Level 2: emitted by HandlerL1
+class EventL2(cqrs.DomainEvent, frozen=True):
+    seed: str
+
+# Level 3: emitted by HandlerL2 (terminal)
+class EventL3(cqrs.DomainEvent, frozen=True):
+    seed: str
+
+class HandlerL1(cqrs.EventHandler[EventL1]):
+    def __init__(self) -> None:
+        self._follow_ups: list[IEvent] = []
+
+    @property
+    def events(self) -> typing.Sequence[IEvent]:
+        return tuple(self._follow_ups)
+
+    async def handle(self, event: EventL1) -> None:
+        # Side effects...
+        self._follow_ups.append(EventL2(seed=event.seed))
+
+class HandlerL2(cqrs.EventHandler[EventL2]):
+    def __init__(self) -> None:
+        self._follow_ups: list[IEvent] = []
+
+    @property
+    def events(self) -> typing.Sequence[IEvent]:
+        return tuple(self._follow_ups)
+
+    async def handle(self, event: EventL2) -> None:
+        # Side effects...
+        self._follow_ups.append(EventL3(seed=event.seed))
+
+class HandlerL3(cqrs.EventHandler[EventL3]):
+    async def handle(self, event: EventL3) -> None:
+        # Terminal handler — no follow-ups (default events = ())
+        pass
+
+# In events_mapper:
+def domain_events_mapper(mapper: cqrs.EventMap) -> None:
+    mapper.bind(EventL1, HandlerL1)
+    mapper.bind(EventL2, HandlerL2)
+    mapper.bind(EventL3, HandlerL3)
+```
+
+When you emit `EventL1(seed="x")`, the processor runs: L1 → HandlerL1 emits L2 → HandlerL2 emits L3 → HandlerL3 runs. All in the same `emit_events()` call (sequential BFS or parallel with FIRST_COMPLETED).

docs/event_handler/index.md

@@ -44,12 +44,13 @@
 
 Event handlers process domain events that are emitted from command handlers. These events represent something that happened in the domain and trigger side effects like sending notifications, updating read models, or triggering other workflows.
 
-When a command handler processes a request, it can emit domain events through the `events` property. These events are automatically collected and processed by event handlers registered in the system.
+When a command handler processes a request, it can emit domain events through the `events` property. These events are automatically collected and processed by event handlers registered in the system. **Event handlers** can in turn produce **follow-up events** via their own `events` property; these follow-ups are processed in the same pipeline (sequential BFS or parallel with semaphore), enabling multi-level event chains.
 
 | Aspect | Description |
 |--------|-------------|
 | **Runtime Processing** | Events are processed synchronously in the same request context, not asynchronously |
 | **Automatic Dispatch** | Events are automatically dispatched to registered handlers after command execution |
+| **Event Propagation** | Handlers can return follow-up events via `events`; they are processed in the same run (BFS or parallel) |
 | **Parallel Support** | Multiple events can be processed in parallel with configurable concurrency limits |
 | **Side Effects** | Event handlers perform side effects without blocking the main command flow |
 

docs/event_handler/parallel_processing.md

@@ -22,34 +22,39 @@ Events can be processed in parallel to improve performance. This is controlled b
 
 ### How Parallel Processing Works
 
+In **sequential** mode, events and follow-ups (from `handler.events`) are processed in **BFS order**: one event at a time, then its follow-ups are appended to the queue. In **parallel** mode, events are processed under a semaphore; as soon as any task completes (FIRST_COMPLETED), its follow-up events are queued and started without waiting for sibling events. `emit_events` returns only when all events and follow-ups are done.
+
 ```mermaid
 graph TD
     Start[EventProcessor.emit_events] --> CheckEnable{Parallel Enabled?}
     
-    CheckEnable -->|No| Sequential[Sequential Processing]
-    Sequential --> LoopSeq[For Each Event]
-    LoopSeq --> EmitSeq[EventEmitter.emit]
+    CheckEnable -->|No| Sequential[Sequential: BFS]
+    Sequential --> PopSeq[Pop event from queue]
+    PopSeq --> 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]
+    HandlerSeq --> CollectSeq[Collect handler.events]
+    BrokerSeq --> CollectSeq
+    CollectSeq --> ExtendSeq[Append follow-ups to queue]
+    ExtendSeq --> MoreSeq{Queue empty?}
+    MoreSeq -->|No| PopSeq
+    MoreSeq -->|Yes| End1[End]
     
-    CheckEnable -->|Yes| Parallel[Parallel Processing]
-    Parallel --> LoopPar[For Each Event]
-    LoopPar --> CreateTask[Create Task]
-    CreateTask --> Semaphore[Acquire Semaphore]
+    CheckEnable -->|Yes| Parallel[Parallel: Semaphore + FIRST_COMPLETED]
+    Parallel --> PopPar[Start tasks for queued events]
+    PopPar --> Semaphore[Acquire Semaphore per task]
     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]
+    HandlerPar --> CollectPar[Collect follow-ups]
+    BrokerPar --> CollectPar
+    CollectPar --> QueuePar[Queue follow-ups, start new tasks]
+    QueuePar --> WaitPar[Wait FIRST_COMPLETED]
+    WaitPar --> MorePar{Pending or queue?}
+    MorePar -->|Yes| PopPar
+    MorePar -->|No| End2[End]
     
     style Start fill:#e1f5ff
     style Sequential fill:#fff3e0
@@ -59,7 +64,7 @@ graph TD
 
 ### Implementation
 
-The `EventProcessor` handles parallel or sequential event emission:
+The `EventProcessor` handles parallel or sequential event emission. Follow-up events returned by handlers (via `handler.events`) are processed in the **same pipeline**: BFS in sequential mode, or under the same semaphore with FIRST_COMPLETED in parallel mode. The method returns when all events and follow-ups are done.
 
 ```python
 class EventProcessor:
@@ -75,27 +80,29 @@ class EventProcessor:
         self._concurrent_event_handle_enable = concurrent_event_handle_enable
         self._event_semaphore = asyncio.Semaphore(max_concurrent_event_handlers)
 
-    async def emit_events(self, events: List[Event]) -> None:
-        """Emit events via event emitter (parallel or sequential)."""
+    async def emit_events(self, events: Sequence[IEvent]) -> None:
+        """Emit events and process follow-ups in the same pipeline."""
         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_emitter.emit(event)
+            # Sequential: BFS over events and follow-ups
+            to_process = deque(events)
+            while to_process:
+                event = to_process.popleft()
+                follow_ups = await self._event_emitter.emit(event)
+                to_process.extend(follow_ups)
         else:
-            # Parallel processing with semaphore limit (fire-and-forget)
-            for event in events:
-                asyncio.create_task(self._emit_event_with_semaphore(event))
+            # Parallel: tasks under semaphore; follow-ups queued on FIRST_COMPLETED
+            await self._emit_events_parallel_first_completed(deque(events))
 
-    async def _emit_event_with_semaphore(self, event: Event) -> None:
-        """Emit a single event with semaphore limit."""
+    async def _emit_one_event(self, event: IEvent) -> Sequence[IEvent]:
+        """Emit one event under semaphore; returns follow-ups from handler.events."""
         async with self._event_semaphore:
-            await self._event_emitter.emit(event)
+            return await self._event_emitter.emit(event)
 ```
 
-The `EventEmitter` then routes events to handlers or message brokers based on event type.
+The `EventEmitter.emit()` returns follow-up events from domain event handlers; the processor continues with these until the queue is empty.
 
 ### Configuration
 
@@ -140,9 +147,10 @@ class ProcessOrderCommandHandler(RequestHandler[ProcessOrderCommand, None]):
         self._events.append(EmailNotificationEvent(...))
 
 # With max_concurrent_event_handlers=3:
-# - Events 1-3 emit in parallel (fire-and-forget tasks)
-# - Event 4 waits for a semaphore slot
+# - Up to 3 events (or follow-ups) run at once under the semaphore
+# - When any task completes, its follow-ups (from handler.events) are queued and started (FIRST_COMPLETED)
+# - emit_events() returns only when all events and follow-ups are done
 # - Each event is routed by EventEmitter:
-#   - DomainEvents → processed by handlers
-#   - NotificationEvents → sent to message broker
+#   - DomainEvents → processed by handlers (follow-ups collected and processed in same pipeline)
+#   - NotificationEvents → sent to message broker (no follow-ups)
 ```