src/Core/Framework/MessageQueue/Api/ConsumeMessagesController.php line 62

  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\MessageQueue\Subscriber\MessageQueueStatsSubscriber;
  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. #[Route(defaults: ['_routeScope' => ['api']])]
  24. #[Package('system-settings')]
  25. class ConsumeMessagesController extends AbstractController
  26. {
  27.     /**
  28.      * @internal
  29.      */
  30.     public function __construct(private readonly ServiceLocator $receiverLocator, private readonly MessageBusInterface $bus, private readonly StopWorkerOnRestartSignalListener $stopWorkerOnRestartSignalListener, private readonly StopWorkerOnSigtermSignalListener $stopWorkerOnSigtermSignalListener, private readonly DispatchPcntlSignalListener $dispatchPcntlSignalListener, private readonly EarlyReturnMessagesListener $earlyReturnListener, private readonly MessageQueueStatsSubscriber $statsSubscriber, private readonly string $defaultTransportName, private readonly string $memoryLimit)
  31.     {
  32.     }
  34.     #[Route(path: '/api/_action/message-queue/consume', name: 'api.action.message-queue.consume', methods: ['POST'])]
  35.     public function consumeMessages(Request $request): JsonResponse
  36.     {
  37.         $receiverName = $request->get('receiver');
  39.         if (!$receiverName || !$this->receiverLocator->has($receiverName)) {
  40.             throw new \RuntimeException('No receiver name provided.');
  41.         }
  43.         $receiver = $this->receiverLocator->get($receiverName);
  45.         $workerDispatcher = new EventDispatcher();
  46.         $listener = new CountHandledMessagesListener();
  47.         $workerDispatcher->addSubscriber($listener);
  48.         $workerDispatcher->addSubscriber($this->statsSubscriber);
  49.         $workerDispatcher->addSubscriber($this->stopWorkerOnRestartSignalListener);
  50.         $workerDispatcher->addSubscriber($this->stopWorkerOnSigtermSignalListener);
  51.         $workerDispatcher->addSubscriber($this->dispatchPcntlSignalListener);
  52.         $workerDispatcher->addSubscriber($this->earlyReturnListener);
  54.         if ($this->memoryLimit !== '-1') {
  55.             $workerDispatcher->addSubscriber(new StopWorkerOnMemoryLimitListener(
  56.                 MemorySizeCalculator::convertToBytes($this->memoryLimit)
  57.             ));
  58.         }
  60.         $worker = new Worker([$this->defaultTransportName => $receiver], $this->bus, $workerDispatcher);
  62.         $worker->run();
  64.         return new JsonResponse(['handledMessages' => $listener->getHandledMessages()]);
  65.     }
  66. }