platform/src/Core/Framework/MessageQueue/Subscriber/MessageFailedHandler.php line 32

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.     public function __construct(IncrementGatewayRegistry $gatewayRegistry, string $defaultTransportName)
  19.     {
  20.         $this->defaultTransportName = $defaultTransportName;
  21.         $this->gatewayRegistry = $gatewayRegistry;
  22.     }
  24.     public static function getSubscribedEvents(): array
  25.     {
  26.         return [
  27.             // must have higher priority than SendFailedMessageToFailureTransportListener
  28.             WorkerMessageFailedEvent::class => ['onMessageFailed', 99],
  29.         ];
  30.     }
  32.     public function onMessageFailed(WorkerMessageFailedEvent $event): void
  33.     {
  34.         if ($event->willRetry()) {
  35.             return;
  36.         }
  38.         $message = $event->getEnvelope();
  40.         if (!$this->wasReceivedByDefaultTransport($message)) {
  41.             return;
  42.         }
  44.         $name = \get_class($message->getMessage());
  46.         try {
  47.             $gateway = $this->gatewayRegistry->get(IncrementGatewayRegistry::MESSAGE_QUEUE_POOL);
  48.         } catch (IncrementGatewayNotFoundException $exception) {
  49.             return;
  50.         }
  52.         $gateway->decrement('message_queue_stats', $name);
  53.     }
  55.     private function wasReceivedByDefaultTransport(Envelope $message): bool
  56.     {
  57.         foreach ($message->all(ReceivedStamp::class) as $stamp) {
  58.             if ($stamp instanceof ReceivedStamp && $stamp->getTransportName() === $this->defaultTransportName) {
  59.                 return true;
  60.             }
  61.         }
  63.         return false;
  64.     }
  65. }