> */ private array $deferred = []; private ?string $activeExecution = null; private int $dispatchDepth = 0; public function __construct( private readonly EventListenerRegistry $registry, private readonly ContainerInterface $container, private readonly LoggerInterface $logger, private readonly float $deferredProcessingTimeoutSeconds = self::DEFAULT_DEFERRED_PROCESSING_TIMEOUT_SECONDS, private readonly int $maxDeferredEvents = self::DEFAULT_MAX_DEFERRED_EVENTS, private readonly int $maxDeferredListenerInvocations = self::DEFAULT_MAX_DEFERRED_LISTENER_INVOCATIONS, ) { if ($this->deferredProcessingTimeoutSeconds <= 0) { throw new \InvalidArgumentException('The deferred processing timeout must be greater than zero.'); } if ($this->maxDeferredEvents <= 0) { throw new \InvalidArgumentException('The deferred event limit must be greater than zero.'); } if ($this->maxDeferredListenerInvocations <= 0) { throw new \InvalidArgumentException('The deferred listener invocation limit must be greater than zero.'); } } public function dispatch(Event $event): void { if (++$this->dispatchDepth > 32) { --$this->dispatchDepth; throw new \RuntimeException('Event dispatch recursion limit exceeded.'); } try { $this->invoke($event, DeliveryMode::Immediate); if ($this->registry->listeners($event->label(), DeliveryMode::Deferred) !== []) { if ($this->activeExecution === null) { throw new \LogicException('Deferred events require an active execution scope.'); } $this->deferred[$this->activeExecution][] = $event; } } finally { --$this->dispatchDepth; } } public function beginExecution(string $executionId): void { if ($this->activeExecution !== null) { throw new \LogicException('An event execution scope is already active.'); } $this->activeExecution = $executionId; $this->deferred[$executionId] = []; } public function processDeferred(string $executionId): DeferredProcessingResult { if ($this->activeExecution !== $executionId) { throw new \LogicException('Cannot process deferred events for an inactive execution.'); } try { $processedEvents = 0; $listenerInvocations = 0; $deadline = microtime(true) + $this->deferredProcessingTimeoutSeconds; $deadlineExceeded = false; $eventLimitExceeded = false; $listenerInvocationLimitExceeded = false; while (($event = array_shift($this->deferred[$executionId])) !== null) { if ($processedEvents >= $this->maxDeferredEvents) { $eventLimitExceeded = true; array_unshift($this->deferred[$executionId], $event); break; } if (microtime(true) >= $deadline) { $deadlineExceeded = true; array_unshift($this->deferred[$executionId], $event); break; } $eventListenerCount = count($this->registry->listeners( $event->label(), DeliveryMode::Deferred, )); if ($listenerInvocations + $eventListenerCount > $this->maxDeferredListenerInvocations) { $listenerInvocationLimitExceeded = true; array_unshift($this->deferred[$executionId], $event); break; } $listenerInvocations += $this->invoke($event, DeliveryMode::Deferred); $processedEvents++; } return new DeferredProcessingResult( processed: $processedEvents, remaining: count($this->deferred[$executionId]), deadlineExceeded: $deadlineExceeded, limitExceeded: $eventLimitExceeded || $listenerInvocationLimitExceeded, listenerInvocations: $listenerInvocations, eventLimitExceeded: $eventLimitExceeded, listenerInvocationLimitExceeded: $listenerInvocationLimitExceeded, ); } finally { $this->discardDeferred($executionId); } } public function discardDeferred(string $executionId): void { unset($this->deferred[$executionId]); if ($this->activeExecution === $executionId) { $this->activeExecution = null; } } private function invoke(Event $event, DeliveryMode $delivery): int { $processed = 0; foreach ($this->registry->listeners($event->label(), $delivery) as $listener) { if ($event->isPropagationStopped()) { break; } $processed++; try { $service = $this->container->get($listener->service); $service->{$listener->method}($event); } catch (\Throwable $error) { $this->logger->error('Event listener failed.', [ 'event' => $event->label(), 'module' => $listener->module, 'listener' => $listener->service . '::' . $listener->method, 'exception' => $error, ]); if ($listener->failurePolicy === FailurePolicy::Propagate) { throw $error; } } } return $processed; } }