vendor/shopware/core/Framework/MessageQueue/DeadMessage/RequeueDeadMessagesService.php line 49

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Framework\MessageQueue\DeadMessage;
  5. use Psr\Log\LoggerInterface;
  6. use Shopware\Core\Framework\Context;
  7. use Shopware\Core\Framework\DataAbstractionLayer\EntityRepositoryInterface;
  8. use Shopware\Core\Framework\DataAbstractionLayer\Search\Criteria;
  9. use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\EqualsFilter;
  10. use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\RangeFilter;
  11. use Shopware\Core\Framework\Feature;
  12. use Shopware\Core\Framework\Log\Package;
  13. use Shopware\Core\Framework\MessageQueue\Message\RetryMessage;
  14. use Symfony\Component\Messenger\MessageBusInterface;
  16. /***
  17.  * @package core
  18.  * @deprecated tag:v6.5.0 - Will be removed, as we use the default symfony retry mechanism
  19.  */
  21. #[Package('core')]
  22. class RequeueDeadMessagesService
  23. {
  24.     private const MAX_RETRIES = 3;
  26.     private EntityRepositoryInterface $deadMessageRepository;
  28.     private MessageBusInterface $bus;
  30.     private MessageBusInterface $encryptedBus;
  32.     private LoggerInterface $logger;
  34.     /**
  35.      * @internal
  36.      */
  37.     public function __construct(
  38.         EntityRepositoryInterface $deadMessageRepository,
  39.         MessageBusInterface $bus,
  40.         MessageBusInterface $encryptedBus,
  41.         LoggerInterface $logger
  42.     ) {
  43.         $this->deadMessageRepository = $deadMessageRepository;
  44.         $this->bus = $bus;
  45.         $this->encryptedBus = $encryptedBus;
  46.         $this->logger = $logger;
  47.     }
  49.     public function requeue(?string $messageClass = null): void
  50.     {
  51.         Feature::triggerDeprecationOrThrow(
  52.             'v6.5.0.0',
  53.             Feature::deprecatedClassMessage(__CLASS__, 'v6.5.0.0')
  54.         );
  56.         $criteria = $this->buildCriteria($messageClass);
  57.         $context = Context::createDefaultContext();
  58.         $messages = $this->deadMessageRepository->search($criteria, $context)->getEntities();
  60.         $notFoundDeadMessages = [];
  62.         /** @var DeadMessageEntity $message */
  63.         foreach ($messages as $message) {
  64.             if (!class_exists($message->getOriginalMessageClass())) {
  65.                 $notFoundDeadMessages[] = ['id' => $message->getId()];
  67.                 continue;
  68.             }
  70.             if ($message->getErrorCount() > self::MAX_RETRIES) {
  71.                 $this->logger->warning(sprintf('Dropped the message %s after %d retries', $message->getOriginalMessageClass(), self::MAX_RETRIES), [
  72.                     'exception' => $message->getException(),
  73.                     'exceptionFile' => $message->getExceptionFile(),
  74.                     'exceptionLine' => $message->getExceptionLine(),
  75.                     'exceptionMessage' => $message->getExceptionMessage(),
  76.                 ]);
  77.                 $notFoundDeadMessages[] = ['id' => $message->getId()];
  79.                 continue;
  80.             }
  82.             $this->dispatchRetryMessage($message);
  83.         }
  85.         if (!empty($notFoundDeadMessages)) {
  86.             $this->deadMessageRepository->delete($notFoundDeadMessages, $context);
  87.         }
  88.     }
  90.     private function buildCriteria(?string $messageClass): Criteria
  91.     {
  92.         $criteria = new Criteria();
  93.         $criteria->addFilter(new RangeFilter(
  94.             'nextExecutionTime',
  95.             [
  96.                 RangeFilter::LT => (new \DateTime())->format(\DATE_ATOM),
  97.             ]
  98.         ));
  100.         if ($messageClass) {
  101.             $criteria->addFilter(new EqualsFilter('originalMessageClass', $messageClass));
  102.         }
  104.         return $criteria;
  105.     }
  107.     private function dispatchRetryMessage(DeadMessageEntity $message): void
  108.     {
  109.         $retryMessage = new RetryMessage($message->getId());
  111.         if ($message->isEncrypted()) {
  112.             $this->encryptedBus->dispatch($retryMessage);
  113.         } else {
  114.             $this->bus->dispatch($retryMessage);
  115.         }
  116.     }
  117. }