vendor/shopware/core/Framework/MessageQueue/Subscriber/MessageFailedHandler.php line 40

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Framework\MessageQueue\Subscriber;
  5. use Shopware\Core\Framework\Increment\Exception\IncrementGatewayNotFoundException;
  6. use Shopware\Core\Framework\Increment\IncrementGatewayRegistry;
  7. use Shopware\Core\Framework\Log\Package;
  8. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  9. use Symfony\Component\Messenger\Envelope;
  10. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  11. use Symfony\Component\Messenger\Stamp\ReceivedStamp;
  13. /**
  14.  * @deprecated tag:v6.5.0 - reason:becomes-internal - EventSubscribers will become internal in v6.5.0
  15.  */
  16. #[Package('system-settings')]
  17. class MessageFailedHandler implements EventSubscriberInterface
  18. {
  19.     private string $defaultTransportName;
  21.     private IncrementGatewayRegistry $gatewayRegistry;
  23.     /**
  24.      * @internal
  25.      */
  26.     public function __construct(IncrementGatewayRegistry $gatewayRegistry, string $defaultTransportName)
  27.     {
  28.         $this->defaultTransportName = $defaultTransportName;
  29.         $this->gatewayRegistry = $gatewayRegistry;
  30.     }
  32.     public static function getSubscribedEvents(): array
  33.     {
  34.         return [
  35.             // must have higher priority than SendFailedMessageToFailureTransportListener
  36.             WorkerMessageFailedEvent::class => ['onMessageFailed', 99],
  37.         ];
  38.     }
  40.     public function onMessageFailed(WorkerMessageFailedEvent $event): void
  41.     {
  42.         if ($event->willRetry()) {
  43.             return;
  44.         }
  46.         $message = $event->getEnvelope();
  48.         if (!$this->wasReceivedByDefaultTransport($message)) {
  49.             return;
  50.         }
  52.         $name = \get_class($message->getMessage());
  54.         try {
  55.             $gateway = $this->gatewayRegistry->get(IncrementGatewayRegistry::MESSAGE_QUEUE_POOL);
  56.         } catch (IncrementGatewayNotFoundException $exception) {
  57.             return;
  58.         }
  60.         $gateway->decrement('message_queue_stats', $name);
  61.     }
  63.     private function wasReceivedByDefaultTransport(Envelope $message): bool
  64.     {
  65.         foreach ($message->all(ReceivedStamp::class) as $stamp) {
  66.             if ($stamp instanceof ReceivedStamp && $stamp->getTransportName() === $this->defaultTransportName) {
  67.                 return true;
  68.             }
  69.         }
  71.         return false;
  72.     }
  73. }