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

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. class MessageFailedHandler implements EventSubscriberInterface
  13. {
  14.     private string $defaultTransportName;
  16.     private IncrementGatewayRegistry $gatewayRegistry;
  18.     /**
  19.      * @internal
  20.      */
  21.     public function __construct(IncrementGatewayRegistry $gatewayRegistry, string $defaultTransportName)
  22.     {
  23.         $this->defaultTransportName = $defaultTransportName;
  24.         $this->gatewayRegistry = $gatewayRegistry;
  25.     }
  27.     public static function getSubscribedEvents(): array
  28.     {
  29.         return [
  30.             // must have higher priority than SendFailedMessageToFailureTransportListener
  31.             WorkerMessageFailedEvent::class => ['onMessageFailed', 99],
  32.         ];
  33.     }
  35.     public function onMessageFailed(WorkerMessageFailedEvent $event): void
  36.     {
  37.         if ($event->willRetry()) {
  38.             return;
  39.         }
  41.         $message = $event->getEnvelope();
  43.         if (!$this->wasReceivedByDefaultTransport($message)) {
  44.             return;
  45.         }
  47.         $name = \get_class($message->getMessage());
  49.         try {
  50.             $gateway = $this->gatewayRegistry->get(IncrementGatewayRegistry::MESSAGE_QUEUE_POOL);
  51.         } catch (IncrementGatewayNotFoundException $exception) {
  52.             return;
  53.         }
  55.         $gateway->decrement('message_queue_stats', $name);
  56.     }
  58.     private function wasReceivedByDefaultTransport(Envelope $message): bool
  59.     {
  60.         foreach ($message->all(ReceivedStamp::class) as $stamp) {
  61.             if ($stamp instanceof ReceivedStamp && $stamp->getTransportName() === $this->defaultTransportName) {
  62.                 return true;
  63.             }
  64.         }
  66.         return false;
  67.     }
  68. }