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

Open in your IDE?
  1. <?php
  2. /*
  3.  * This file is part of the Symfony package.
  4.  *
  5.  * (c) Fabien Potencier <fabien@symfony.com>
  6.  *
  7.  * For the full copyright and license information, please view the LICENSE
  8.  * file that was distributed with this source code.
  9.  */
  11. namespace Symfony\Component\Messenger\EventListener;
  13. use Psr\Container\ContainerInterface;
  14. use Psr\Log\LoggerInterface;
  15. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  16. use Symfony\Component\Messenger\Envelope;
  17. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  18. use Symfony\Component\Messenger\Exception\HandlerFailedException;
  19. use Symfony\Component\Messenger\Exception\RecoverableExceptionInterface;
  20. use Symfony\Component\Messenger\Exception\RuntimeException;
  21. use Symfony\Component\Messenger\Exception\UnrecoverableExceptionInterface;
  22. use Symfony\Component\Messenger\Retry\RetryStrategyInterface;
  23. use Symfony\Component\Messenger\Stamp\DelayStamp;
  24. use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
  25. use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
  27. /**
  28.  * @author Tobias Schultze <http://tobion.de>
  29.  */
  30. class SendFailedMessageForRetryListener implements EventSubscriberInterface
  31. {
  32.     private $sendersLocator;
  33.     private $retryStrategyLocator;
  34.     private $logger;
  36.     public function __construct(ContainerInterface $sendersLocator, ContainerInterface $retryStrategyLocator, LoggerInterface $logger = null)
  37.     {
  38.         $this->sendersLocator = $sendersLocator;
  39.         $this->retryStrategyLocator = $retryStrategyLocator;
  40.         $this->logger = $logger;
  41.     }
  43.     public function onMessageFailed(WorkerMessageFailedEvent $event)
  44.     {
  45.         $retryStrategy = $this->getRetryStrategyForTransport($event->getReceiverName());
  46.         $envelope = $event->getEnvelope();
  47.         $throwable = $event->getThrowable();
  49.         $message = $envelope->getMessage();
  50.         $context = [
  51.             'message' => $message,
  52.             'class' => \get_class($message),
  53.         ];
  55.         $shouldRetry = $retryStrategy && $this->shouldRetry($throwable, $envelope, $retryStrategy);
  57.         $retryCount = RedeliveryStamp::getRetryCountFromEnvelope($envelope);
  58.         if ($shouldRetry) {
  59.             $event->setForRetry();
  61.             ++$retryCount;
  63.             $delay = $retryStrategy->getWaitingTime($envelope, $throwable);
  65.             if (null !== $this->logger) {
  66.                 $this->logger->error('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]);
  67.             }
  69.             // add the delay and retry stamp info
  70.             $retryEnvelope = $envelope->with(new DelayStamp($delay), new RedeliveryStamp($retryCount));
  72.             // re-send the message for retry
  73.             $this->getSenderForTransport($event->getReceiverName())->send($retryEnvelope);
  74.         } else {
  75.             if (null !== $this->logger) {
  76.                 $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]);
  77.             }
  78.         }
  79.     }
  81.     public static function getSubscribedEvents()
  82.     {
  83.         return [
  84.             // must have higher priority than SendFailedMessageToFailureTransportListener
  85.             WorkerMessageFailedEvent::class => ['onMessageFailed', 100],
  86.         ];
  87.     }
  89.     private function shouldRetry(\Throwable $e, Envelope $envelope, RetryStrategyInterface $retryStrategy): bool
  90.     {
  91.         if ($e instanceof RecoverableExceptionInterface) {
  92.             return true;
  93.         }
  95.         // if one or more nested Exceptions is an instance of RecoverableExceptionInterface we should retry
  96.         // if ALL nested Exceptions are an instance of UnrecoverableExceptionInterface we should not retry
  97.         if ($e instanceof HandlerFailedException) {
  98.             $shouldNotRetry = true;
  99.             foreach ($e->getNestedExceptions() as $nestedException) {
  100.                 if ($nestedException instanceof RecoverableExceptionInterface) {
  101.                     return true;
  102.                 }
  104.                 if (!$nestedException instanceof UnrecoverableExceptionInterface) {
  105.                     $shouldNotRetry = false;
  106.                     break;
  107.                 }
  108.             }
  109.             if ($shouldNotRetry) {
  110.                 return false;
  111.             }
  112.         }
  114.         if ($e instanceof UnrecoverableExceptionInterface) {
  115.             return false;
  116.         }
  118.         return $retryStrategy->isRetryable($envelope, $e);
  119.     }
  121.     private function getRetryStrategyForTransport(string $alias): ?RetryStrategyInterface
  122.     {
  123.         if ($this->retryStrategyLocator->has($alias)) {
  124.             return $this->retryStrategyLocator->get($alias);
  125.         }
  127.         return null;
  128.     }
  130.     private function getSenderForTransport(string $alias): SenderInterface
  131.     {
  132.         if ($this->sendersLocator->has($alias)) {
  133.             return $this->sendersLocator->get($alias);
  134.         }
  136.         throw new RuntimeException(sprintf('Could not find sender "%s" based on the same receiver to send the failed message to for retry.', $alias));
  137.     }
  138. }