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

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