vendor/symfony/messenger/EventListener/SendFailedMessageForRetryListener.php line 54

  1. <?php
  3. /*
  4.  * This file is part of the Symfony package.
  5.  *
  6.  * (c) Fabien Potencier <fabien@symfony.com>
  7.  *
  8.  * For the full copyright and license information, please view the LICENSE
  9.  * file that was distributed with this source code.
  10.  */
  12. namespace Symfony\Component\Messenger\EventListener;
  14. use Psr\Container\ContainerInterface;
  15. use Psr\EventDispatcher\EventDispatcherInterface;
  16. use Psr\Log\LoggerInterface;
  17. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  18. use Symfony\Component\Messenger\Envelope;
  19. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  20. use Symfony\Component\Messenger\Event\WorkerMessageRetriedEvent;
  21. use Symfony\Component\Messenger\Exception\HandlerFailedException;
  22. use Symfony\Component\Messenger\Exception\RecoverableExceptionInterface;
  23. use Symfony\Component\Messenger\Exception\RuntimeException;
  24. use Symfony\Component\Messenger\Exception\UnrecoverableExceptionInterface;
  25. use Symfony\Component\Messenger\Retry\RetryStrategyInterface;
  26. use Symfony\Component\Messenger\Stamp\DelayStamp;
  27. use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
  28. use Symfony\Component\Messenger\Stamp\StampInterface;
  29. use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
  31. /**
  32.  * @author Tobias Schultze <http://tobion.de>
  33.  */
  34. class SendFailedMessageForRetryListener implements EventSubscriberInterface
  35. {
  36.     private ContainerInterface $sendersLocator;
  37.     private ContainerInterface $retryStrategyLocator;
  38.     private ?LoggerInterface $logger;
  39.     private ?EventDispatcherInterface $eventDispatcher;
  40.     private int $historySize;
  42.     public function __construct(ContainerInterface $sendersLocator, ContainerInterface $retryStrategyLocator, ?LoggerInterface $logger = null, ?EventDispatcherInterface $eventDispatcher = null, int $historySize = 10)
  43.     {
  44.         $this->sendersLocator = $sendersLocator;
  45.         $this->retryStrategyLocator = $retryStrategyLocator;
  46.         $this->logger = $logger;
  47.         $this->eventDispatcher = $eventDispatcher;
  48.         $this->historySize = $historySize;
  49.     }
  51.     /**
  52.      * @return void
  53.      */
  54.     public function onMessageFailed(WorkerMessageFailedEvent $event)
  55.     {
  56.         $retryStrategy = $this->getRetryStrategyForTransport($event->getReceiverName());
  57.         $envelope = $event->getEnvelope();
  58.         $throwable = $event->getThrowable();
  60.         $message = $envelope->getMessage();
  61.         $context = [
  62.             'class' => $message::class,
  63.         ];
  65.         $shouldRetry = $retryStrategy && $this->shouldRetry($throwable, $envelope, $retryStrategy);
  67.         $retryCount = RedeliveryStamp::getRetryCountFromEnvelope($envelope);
  68.         if ($shouldRetry) {
  69.             $event->setForRetry();
  71.             ++$retryCount;
  73.             $delay = $retryStrategy->getWaitingTime($envelope, $throwable);
  75.             $this->logger?->warning('Error thrown while handling message {class}. Sending for retry #{retryCount} using {delay} ms delay. Error: "{error}"', $context + ['retryCount' => $retryCount, 'delay' => $delay, 'error' => $throwable->getMessage(), 'exception' => $throwable]);
  77.             // add the delay and retry stamp info
  78.             $retryEnvelope = $this->withLimitedHistory($envelope, new DelayStamp($delay), new RedeliveryStamp($retryCount));
  80.             // re-send the message for retry
  81.             $retryEnvelope = $this->getSenderForTransport($event->getReceiverName())->send($retryEnvelope);
  83.             $this->eventDispatcher?->dispatch(new WorkerMessageRetriedEvent($retryEnvelope, $event->getReceiverName()));
  84.         } else {
  85.             $this->logger?->critical('Error thrown while handling message {class}. Removing from transport after {retryCount} retries. Error: "{error}"', $context + ['retryCount' => $retryCount, 'error' => $throwable->getMessage(), 'exception' => $throwable]);
  86.         }
  87.     }
  89.     /**
  90.      * Adds stamps to the envelope by keeping only the First + Last N stamps.
  91.      */
  92.     private function withLimitedHistory(Envelope $envelope, StampInterface ...$stamps): Envelope
  93.     {
  94.         foreach ($stamps as $stamp) {
  95.             $history = $envelope->all($stamp::class);
  96.             if (\count($history) < $this->historySize) {
  97.                 $envelope = $envelope->with($stamp);
  98.                 continue;
  99.             }
  101.             $history = array_merge(
  102.                 [$history[0]],
  103.                 \array_slice($history, -$this->historySize + 2),
  104.                 [$stamp]
  105.             );
  107.             $envelope = $envelope->withoutAll($stamp::class)->with(...$history);
  108.         }
  110.         return $envelope;
  111.     }
  113.     public static function getSubscribedEvents(): array
  114.     {
  115.         return [
  116.             // must have higher priority than SendFailedMessageToFailureTransportListener
  117.             WorkerMessageFailedEvent::class => ['onMessageFailed', 100],
  118.         ];
  119.     }
  121.     private function shouldRetry(\Throwable $e, Envelope $envelope, RetryStrategyInterface $retryStrategy): bool
  122.     {
  123.         if ($e instanceof RecoverableExceptionInterface) {
  124.             return true;
  125.         }
  127.         // if one or more nested Exceptions is an instance of RecoverableExceptionInterface we should retry
  128.         // if ALL nested Exceptions are an instance of UnrecoverableExceptionInterface we should not retry
  129.         if ($e instanceof HandlerFailedException) {
  130.             $shouldNotRetry = true;
  131.             foreach ($e->getWrappedExceptions() as $nestedException) {
  132.                 if ($nestedException instanceof RecoverableExceptionInterface) {
  133.                     return true;
  134.                 }
  136.                 if (!$nestedException instanceof UnrecoverableExceptionInterface) {
  137.                     $shouldNotRetry = false;
  138.                     break;
  139.                 }
  140.             }
  141.             if ($shouldNotRetry) {
  142.                 return false;
  143.             }
  144.         }
  146.         if ($e instanceof UnrecoverableExceptionInterface) {
  147.             return false;
  148.         }
  150.         return $retryStrategy->isRetryable($envelope, $e);
  151.     }
  153.     private function getRetryStrategyForTransport(string $alias): ?RetryStrategyInterface
  154.     {
  155.         if ($this->retryStrategyLocator->has($alias)) {
  156.             return $this->retryStrategyLocator->get($alias);
  157.         }
  159.         return null;
  160.     }
  162.     private function getSenderForTransport(string $alias): SenderInterface
  163.     {
  164.         if ($this->sendersLocator->has($alias)) {
  165.             return $this->sendersLocator->get($alias);
  166.         }
  168.         throw new RuntimeException(sprintf('Could not find sender "%s" based on the same receiver to send the failed message to for retry.', $alias));
  169.     }
  170. }