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. }