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

Open in your IDE?
  1. <?php
  2. /*
  3.  * This file is part of the Symfony package.
  4.  *
  5.  * (c) Fabien Potencier <[email protected]>
  6.  *
  7.  * For the full copyright and license information, please view the LICENSE
  8.  * file that was distributed with this source code.
  9.  */
  10. namespace Symfony\Component\Messenger\EventListener;
  11. use Psr\Container\ContainerInterface;
  12. use Psr\Log\LoggerInterface;
  13. use Psr\Log\LogLevel;
  14. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  15. use Symfony\Component\Messenger\Envelope;
  16. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  17. use Symfony\Component\Messenger\Event\WorkerMessageRetriedEvent;
  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\Stamp\StampInterface;
  26. use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
  27. use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
  28. /**
  29.  * @author Tobias Schultze <http://tobion.de>
  30.  */
  31. class SendFailedMessageForRetryListener implements EventSubscriberInterface
  32. {
  33.     private $sendersLocator;
  34.     private $retryStrategyLocator;
  35.     private $logger;
  36.     private $eventDispatcher;
  37.     private $historySize;
  38.     public function __construct(ContainerInterface $sendersLocatorContainerInterface $retryStrategyLocatorLoggerInterface $logger nullEventDispatcherInterface $eventDispatcher nullint $historySize 10)
  39.     {
  40.         $this->sendersLocator $sendersLocator;
  41.         $this->retryStrategyLocator $retryStrategyLocator;
  42.         $this->logger $logger;
  43.         $this->eventDispatcher $eventDispatcher;
  44.         $this->historySize $historySize;
  45.     }
  46.     public function onMessageFailed(WorkerMessageFailedEvent $event)
  47.     {
  48.         $retryStrategy $this->getRetryStrategyForTransport($event->getReceiverName());
  49.         $envelope $event->getEnvelope();
  50.         $throwable $event->getThrowable();
  51.         $message $envelope->getMessage();
  52.         $context = [
  53.             'message' => $message,
  54.             'class' => \get_class($message),
  55.         ];
  56.         $shouldRetry $retryStrategy && $this->shouldRetry($throwable$envelope$retryStrategy);
  57.         $retryCount RedeliveryStamp::getRetryCountFromEnvelope($envelope);
  58.         if ($shouldRetry) {
  59.             $event->setForRetry();
  60.             ++$retryCount;
  61.             $delay $retryStrategy->getWaitingTime($envelope$throwable);
  62.             if (null !== $this->logger) {
  63.                 $logLevel LogLevel::ERROR;
  64.                 if ($throwable instanceof RecoverableExceptionInterface) {
  65.                     $logLevel LogLevel::WARNING;
  66.                 } elseif ($throwable instanceof HandlerFailedException) {
  67.                     foreach ($throwable->getNestedExceptions() as $nestedException) {
  68.                         if ($nestedException instanceof RecoverableExceptionInterface) {
  69.                             $logLevel LogLevel::WARNING;
  70.                             break;
  71.                         }
  72.                     }
  73.                 }
  74.                 $this->logger->log($logLevel'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]);
  75.             }
  76.             // add the delay and retry stamp info
  77.             $retryEnvelope $this->withLimitedHistory($envelope, new DelayStamp($delay), new RedeliveryStamp($retryCount));
  78.             // re-send the message for retry
  79.             $this->getSenderForTransport($event->getReceiverName())->send($retryEnvelope);
  80.             if (null !== $this->eventDispatcher) {
  81.                 $this->eventDispatcher->dispatch(new WorkerMessageRetriedEvent($retryEnvelope$event->getReceiverName()));
  82.             }
  83.         } else {
  84.             if (null !== $this->logger) {
  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.         }
  88.     }
  89.     /**
  90.      * Adds stamps to the envelope by keeping only the First + Last N stamps.
  91.      */
  92.     private function withLimitedHistory(Envelope $envelopeStampInterface ...$stamps): Envelope
  93.     {
  94.         foreach ($stamps as $stamp) {
  95.             $history $envelope->all(\get_class($stamp));
  96.             if (\count($history) < $this->historySize) {
  97.                 $envelope $envelope->with($stamp);
  98.                 continue;
  99.             }
  100.             $history array_merge(
  101.                 [$history[0]],
  102.                 \array_slice($history, -$this->historySize 2),
  103.                 [$stamp]
  104.             );
  105.             $envelope $envelope->withoutAll(\get_class($stamp))->with(...$history);
  106.         }
  107.         return $envelope;
  108.     }
  109.     public static function getSubscribedEvents()
  110.     {
  111.         return [
  112.             // must have higher priority than SendFailedMessageToFailureTransportListener
  113.             WorkerMessageFailedEvent::class => ['onMessageFailed'100],
  114.         ];
  115.     }
  116.     private function shouldRetry(\Throwable $eEnvelope $envelopeRetryStrategyInterface $retryStrategy): bool
  117.     {
  118.         if ($e instanceof RecoverableExceptionInterface) {
  119.             return true;
  120.         }
  121.         // if one or more nested Exceptions is an instance of RecoverableExceptionInterface we should retry
  122.         // if ALL nested Exceptions are an instance of UnrecoverableExceptionInterface we should not retry
  123.         if ($e instanceof HandlerFailedException) {
  124.             $shouldNotRetry true;
  125.             foreach ($e->getNestedExceptions() as $nestedException) {
  126.                 if ($nestedException instanceof RecoverableExceptionInterface) {
  127.                     return true;
  128.                 }
  129.                 if (!$nestedException instanceof UnrecoverableExceptionInterface) {
  130.                     $shouldNotRetry false;
  131.                     break;
  132.                 }
  133.             }
  134.             if ($shouldNotRetry) {
  135.                 return false;
  136.             }
  137.         }
  138.         if ($e instanceof UnrecoverableExceptionInterface) {
  139.             return false;
  140.         }
  141.         return $retryStrategy->isRetryable($envelope$e);
  142.     }
  143.     private function getRetryStrategyForTransport(string $alias): ?RetryStrategyInterface
  144.     {
  145.         if ($this->retryStrategyLocator->has($alias)) {
  146.             return $this->retryStrategyLocator->get($alias);
  147.         }
  148.         return null;
  149.     }
  150.     private function getSenderForTransport(string $alias): SenderInterface
  151.     {
  152.         if ($this->sendersLocator->has($alias)) {
  153.             return $this->sendersLocator->get($alias);
  154.         }
  155.         throw new RuntimeException(sprintf('Could not find sender "%s" based on the same receiver to send the failed message to for retry.'$alias));
  156.     }
  157. }