vendor/shopware/core/Framework/MessageQueue/Monitoring/MonitoringBusDecorator.php line 43

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Framework\MessageQueue\Monitoring;
  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\Messenger\Envelope;
  9. use Symfony\Component\Messenger\MessageBusInterface;
  10. use Symfony\Component\Messenger\Stamp\ReceivedStamp;
  11. use Symfony\Component\Messenger\Stamp\SentStamp;
  13. /**
  14.  * @deprecated tag:v6.5.0 - reason:remove-decorator - will be removed, as we use default symfony messenger
  15.  */
  16. #[Package('core')]
  17. class MonitoringBusDecorator implements MessageBusInterface
  18. {
  19.     private MessageBusInterface $innerBus;
  21.     private string $defaultTransportName;
  23.     private IncrementGatewayRegistry $gatewayRegistry;
  25.     /**
  26.      * @internal
  27.      */
  28.     public function __construct(
  29.         MessageBusInterface $inner,
  30.         string $defaultTransportName,
  31.         IncrementGatewayRegistry $gatewayRegistry
  32.     ) {
  33.         $this->innerBus = $inner;
  34.         $this->defaultTransportName = $defaultTransportName;
  35.         $this->gatewayRegistry = $gatewayRegistry;
  36.     }
  38.     /**
  39.      * Dispatches the given message to the inner Bus and Logs it.
  40.      *
  41.      * @param object|Envelope $message
  42.      */
  43.     public function dispatch($message, array $stamps = []): Envelope
  44.     {
  45.         $message = $this->innerBus->dispatch(Envelope::wrap($message, $stamps), $stamps);
  47.         if ($this->wasSentToDefaultTransport($message)) {
  48.             $this->incrementMessageQueueSize($message);
  49.         }
  51.         if ($this->wasReceivedByDefaultTransport($message)) {
  52.             $this->decrementMessageQueueSize($message);
  53.         }
  55.         return $message;
  56.     }
  58.     private function incrementMessageQueueSize(Envelope $message): void
  59.     {
  60.         try {
  61.             $gateway = $this->gatewayRegistry->get(IncrementGatewayRegistry::MESSAGE_QUEUE_POOL);
  62.         } catch (IncrementGatewayNotFoundException $exception) {
  63.             // In case message_queue pool is disabled
  64.             return;
  65.         }
  67.         $gateway->increment('message_queue_stats', \get_class($message->getMessage()));
  68.     }
  70.     private function decrementMessageQueueSize(Envelope $message): void
  71.     {
  72.         try {
  73.             $gateway = $this->gatewayRegistry->get(IncrementGatewayRegistry::MESSAGE_QUEUE_POOL);
  74.         } catch (IncrementGatewayNotFoundException $exception) {
  75.             // In case message_queue pool is disabled
  76.             return;
  77.         }
  79.         $gateway->decrement('message_queue_stats', \get_class($message->getMessage()));
  80.     }
  82.     private function wasSentToDefaultTransport(Envelope $message): bool
  83.     {
  84.         foreach ($message->all(SentStamp::class) as $stamp) {
  85.             if ($stamp instanceof SentStamp && $stamp->getSenderAlias() === $this->defaultTransportName) {
  86.                 return true;
  87.             }
  88.         }
  90.         return false;
  91.     }
  93.     private function wasReceivedByDefaultTransport(Envelope $message): bool
  94.     {
  95.         foreach ($message->all(ReceivedStamp::class) as $stamp) {
  96.             if ($stamp instanceof ReceivedStamp && $stamp->getTransportName() === $this->defaultTransportName) {
  97.                 return true;
  98.             }
  99.         }
  101.         return false;
  102.     }
  103. }