vendor/shopware/core/Framework/MessageQueue/Api/ConsumeMessagesController.php line 98

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Framework\MessageQueue\Api;
  5. use Shopware\Core\Framework\Log\Package;
  6. use Shopware\Core\Framework\MessageQueue\Subscriber\CountHandledMessagesListener;
  7. use Shopware\Core\Framework\MessageQueue\Subscriber\EarlyReturnMessagesListener;
  8. use Shopware\Core\Framework\Routing\Annotation\RouteScope;
  9. use Shopware\Core\Framework\Routing\Annotation\Since;
  10. use Shopware\Core\Framework\Util\MemorySizeCalculator;
  11. use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
  12. use Symfony\Component\DependencyInjection\ServiceLocator;
  13. use Symfony\Component\EventDispatcher\EventDispatcher;
  14. use Symfony\Component\HttpFoundation\JsonResponse;
  15. use Symfony\Component\HttpFoundation\Request;
  16. use Symfony\Component\Messenger\EventListener\DispatchPcntlSignalListener;
  17. use Symfony\Component\Messenger\EventListener\StopWorkerOnMemoryLimitListener;
  18. use Symfony\Component\Messenger\EventListener\StopWorkerOnRestartSignalListener;
  19. use Symfony\Component\Messenger\EventListener\StopWorkerOnSigtermSignalListener;
  20. use Symfony\Component\Messenger\MessageBusInterface;
  21. use Symfony\Component\Messenger\Worker;
  22. use Symfony\Component\Routing\Annotation\Route;
  24. /**
  25.  * @Route(defaults={"_routeScope"={"api"}})
  26.  */
  27. #[Package('system-settings')]
  28. class ConsumeMessagesController extends AbstractController
  29. {
  30.     /**
  31.      * @var ServiceLocator
  32.      */
  33.     private $receiverLocator;
  35.     /**
  36.      * @var MessageBusInterface
  37.      */
  38.     private $bus;
  40.     /**
  41.      * @var int
  42.      */
  43.     private $pollInterval;
  45.     /**
  46.      * @var StopWorkerOnRestartSignalListener
  47.      */
  48.     private $stopWorkerOnRestartSignalListener;
  50.     /**
  51.      * @var StopWorkerOnSigtermSignalListener
  52.      */
  53.     private $stopWorkerOnSigtermSignalListener;
  55.     /**
  56.      * @var DispatchPcntlSignalListener
  57.      */
  58.     private $dispatchPcntlSignalListener;
  60.     /**
  61.      * @var EarlyReturnMessagesListener
  62.      */
  63.     private $earlyReturnListener;
  65.     private string $defaultTransportName;
  67.     private string $memoryLimit;
  69.     /**
  70.      * @internal
  71.      */
  72.     public function __construct(
  73.         ServiceLocator $receiverLocator,
  74.         MessageBusInterface $bus,
  75.         int $pollInterval,
  76.         StopWorkerOnRestartSignalListener $stopWorkerOnRestartSignalListener,
  77.         StopWorkerOnSigtermSignalListener $stopWorkerOnSigtermSignalListener,
  78.         DispatchPcntlSignalListener $dispatchPcntlSignalListener,
  79.         EarlyReturnMessagesListener $earlyReturnListener,
  80.         string $defaultTransportName,
  81.         string $memoryLimit
  82.     ) {
  83.         $this->receiverLocator = $receiverLocator;
  84.         $this->bus = $bus;
  85.         $this->pollInterval = $pollInterval;
  86.         $this->stopWorkerOnRestartSignalListener = $stopWorkerOnRestartSignalListener;
  87.         $this->stopWorkerOnSigtermSignalListener = $stopWorkerOnSigtermSignalListener;
  88.         $this->dispatchPcntlSignalListener = $dispatchPcntlSignalListener;
  89.         $this->earlyReturnListener = $earlyReturnListener;
  90.         $this->defaultTransportName = $defaultTransportName;
  91.         $this->memoryLimit = $memoryLimit;
  92.     }
  94.     /**
  95.      * @Since("6.0.0.0")
  96.      * @Route("/api/_action/message-queue/consume", name="api.action.message-queue.consume", methods={"POST"})
  97.      */
  98.     public function consumeMessages(Request $request): JsonResponse
  99.     {
  100.         $receiverName = $request->get('receiver');
  102.         if (!$receiverName || !$this->receiverLocator->has($receiverName)) {
  103.             throw new \RuntimeException('No receiver name provided.');
  104.         }
  106.         $receiver = $this->receiverLocator->get($receiverName);
  108.         $workerDispatcher = new EventDispatcher();
  109.         $listener = new CountHandledMessagesListener($this->pollInterval);
  110.         $workerDispatcher->addSubscriber($listener);
  111.         $workerDispatcher->addSubscriber($this->stopWorkerOnRestartSignalListener);
  112.         $workerDispatcher->addSubscriber($this->stopWorkerOnSigtermSignalListener);
  113.         $workerDispatcher->addSubscriber($this->dispatchPcntlSignalListener);
  114.         $workerDispatcher->addSubscriber($this->earlyReturnListener);
  116.         if ($this->memoryLimit !== '-1') {
  117.             $workerDispatcher->addSubscriber(new StopWorkerOnMemoryLimitListener(
  118.                 MemorySizeCalculator::convertToBytes($this->memoryLimit)
  119.             ));
  120.         }
  122.         $worker = new Worker([$this->defaultTransportName => $receiver], $this->bus, $workerDispatcher);
  124.         $worker->run(['sleep' => 50]);
  126.         return new JsonResponse(['handledMessages' => $listener->getHandledMessages()]);
  127.     }
  128. }