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

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 Symfony\Component\EventDispatcher\EventSubscriberInterface;
  8. use Symfony\Component\Messenger\Envelope;
  9. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  10. use Symfony\Component\Messenger\Stamp\ReceivedStamp;
  12. /**
  13.  * @deprecated tag:v6.5.0 - reason:becomes-internal - EventSubscribers will become internal in v6.5.0
  14.  */
  15. class MessageFailedHandler implements EventSubscriberInterface
  16. {
  17.     private string $defaultTransportName;
  19.     private IncrementGatewayRegistry $gatewayRegistry;
  21.     /**
  22.      * @internal
  23.      */
  24.     public function __construct(IncrementGatewayRegistry $gatewayRegistry, string $defaultTransportName)
  25.     {
  26.         $this->defaultTransportName = $defaultTransportName;
  27.         $this->gatewayRegistry = $gatewayRegistry;
  28.     }
  30.     public static function getSubscribedEvents(): array
  31.     {
  32.         return [
  33.             // must have higher priority than SendFailedMessageToFailureTransportListener
  34.             WorkerMessageFailedEvent::class => ['onMessageFailed', 99],
  35.         ];
  36.     }
  38.     public function onMessageFailed(WorkerMessageFailedEvent $event): void
  39.     {
  40.         if ($event->willRetry()) {
  41.             return;
  42.         }
  44.         $message = $event->getEnvelope();
  46.         if (!$this->wasReceivedByDefaultTransport($message)) {
  47.             return;
  48.         }
  50.         $name = \get_class($message->getMessage());
  52.         try {
  53.             $gateway = $this->gatewayRegistry->get(IncrementGatewayRegistry::MESSAGE_QUEUE_POOL);
  54.         } catch (IncrementGatewayNotFoundException $exception) {
  55.             return;
  56.         }
  58.         $gateway->decrement('message_queue_stats', $name);
  59.     }
  61.     private function wasReceivedByDefaultTransport(Envelope $message): bool
  62.     {
  63.         foreach ($message->all(ReceivedStamp::class) as $stamp) {
  64.             if ($stamp instanceof ReceivedStamp && $stamp->getTransportName() === $this->defaultTransportName) {
  65.                 return true;
  66.             }
  67.         }
  69.         return false;
  70.     }
  71. }