src/Core/Framework/MessageQueue/Subscriber/MessageQueueStatsSubscriber.php line 47

  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Framework\MessageQueue\Subscriber;
  5. use Shopware\Core\Framework\Increment\IncrementGatewayRegistry;
  6. use Shopware\Core\Framework\Log\Package;
  7. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  8. use Symfony\Component\Messenger\Envelope;
  9. use Symfony\Component\Messenger\Event\SendMessageToTransportsEvent;
  10. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  11. use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
  13. /**
  14.  * @internal
  15.  */
  16. #[Package('system-settings')]
  17. class MessageQueueStatsSubscriber implements EventSubscriberInterface
  18. {
  19.     /**
  20.      * @internal
  21.      */
  22.     public function __construct(private readonly IncrementGatewayRegistry $gatewayRegistry)
  23.     {
  24.     }
  26.     public static function getSubscribedEvents(): array
  27.     {
  28.         return [
  29.             // must have higher priority than SendFailedMessageToFailureTransportListener
  30.             WorkerMessageFailedEvent::class => ['onMessageFailed', 99],
  31.             WorkerMessageHandledEvent::class => 'onMessageHandled',
  32.             SendMessageToTransportsEvent::class => ['onMessageSent', 99],
  33.         ];
  34.     }
  36.     public function onMessageFailed(WorkerMessageFailedEvent $event): void
  37.     {
  38.         if ($event->willRetry()) {
  39.             return;
  40.         }
  42.         $this->handle($event->getEnvelope(), false);
  43.     }
  45.     public function onMessageHandled(WorkerMessageHandledEvent $event): void
  46.     {
  47.         $this->handle($event->getEnvelope(), false);
  48.     }
  50.     public function onMessageSent(SendMessageToTransportsEvent $event): void
  51.     {
  52.         $this->handle($event->getEnvelope(), true);
  53.     }
  55.     private function handle(Envelope $envelope, bool $increment): void
  56.     {
  57.         $name = $envelope->getMessage()::class;
  59.         $gateway = $this->gatewayRegistry->get(IncrementGatewayRegistry::MESSAGE_QUEUE_POOL);
  61.         if ($increment) {
  62.             $gateway->increment('message_queue_stats', $name);
  64.             return;
  65.         }
  67.         $gateway->decrement('message_queue_stats', $name);
  68.     }
  69. }