vendor/pimcore/portal-engine/src/EventSubscriber/BatchTaskSubscriber.php line 78

Open in your IDE?
  1. <?php
  3. /**
  4.  * Pimcore
  5.  *
  6.  * This source file is available under following license:
  7.  * - Pimcore Commercial License (PCL)
  8.  *
  9.  *  @copyright  Copyright (c) Pimcore GmbH (http://www.pimcore.org)
  10.  *  @license    http://www.pimcore.org/license     PCL
  11.  */
  13. namespace Pimcore\Bundle\PortalEngineBundle\EventSubscriber;
  15. use Pimcore\Bundle\PortalEngineBundle\Entity\BatchTask;
  16. use Pimcore\Bundle\PortalEngineBundle\Message\BatchTask\Interfaces\BatchTaskMessageInterface;
  17. use Pimcore\Bundle\PortalEngineBundle\Message\BatchTask\Interfaces\SequentialBatchTaskMessageInterface;
  18. use Pimcore\Bundle\PortalEngineBundle\Message\BatchTask\Interfaces\SplittedBatchTaskMessageInterface;
  19. use Pimcore\Bundle\PortalEngineBundle\Message\BatchTask\Interfaces\TriggerFinishedMessageBatchTaskMessageInterface;
  20. use Pimcore\Bundle\PortalEngineBundle\Service\BatchTask\BatchTaskService;
  21. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  22. use Symfony\Component\HttpKernel\Event\TerminateEvent;
  23. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  24. use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
  25. use Symfony\Component\Messenger\MessageBusInterface;
  27. /**
  28.  * Class IndexUpdateListener
  29.  *
  30.  * @package Pimcore\Bundle\PortalEngineBundle\EventListener
  31.  */
  32. class BatchTaskSubscriber implements EventSubscriberInterface
  33. {
  34.     /**
  35.      * @var BatchTaskService
  36.      */
  37.     protected $batchTaskService;
  39.     /**
  40.      * @var MessageBusInterface
  41.      */
  42.     protected $messageBus;
  44.     /**
  45.      * @var BatchTask
  46.      */
  47.     protected $terminateBatchTask;
  49.     /**
  50.      * @param BatchTaskService $batchTaskService
  51.      * @param MessageBusInterface $messengerBusPortalEngine
  52.      */
  53.     public function __construct(BatchTaskService $batchTaskService, MessageBusInterface $messengerBusPortalEngine)
  54.     {
  55.         $this->batchTaskService = $batchTaskService;
  56.         $this->messageBus = $messengerBusPortalEngine;
  57.     }
  59.     /**
  60.      * @return array
  61.      */
  62.     public static function getSubscribedEvents()
  63.     {
  64.         return [
  65.             WorkerMessageFailedEvent::class => 'onBatchTaskMessageFailed',
  66.             WorkerMessageHandledEvent::class => 'onWorkerMessageHandled',
  67.             TerminateEvent::class => 'onTerminate',
  68.         ];
  69.     }
  71.     /**
  72.      * Mark batch tasks with failed items as finished as otherwise they will run forever.
  73.      *
  74.      * @param WorkerMessageFailedEvent $event
  75.      *
  76.      * @throws \Doctrine\DBAL\DBALException
  77.      */
  78.     public function onBatchTaskMessageFailed(WorkerMessageFailedEvent $event)
  79.     {
  80.         $message = $event->getEnvelope()->getMessage();
  81.         if (!$message instanceof BatchTaskMessageInterface) {
  82.             return;
  83.         }
  85.         if ($event->willRetry()) {
  86.             return;
  87.         }
  89.         if (!$batchTask = $this->batchTaskService->getTaskById($message->getTaskId())) {
  90.             return;
  91.         }
  93.         foreach (array_keys($message->getItems()) as $itemIndex) {
  94.             if (!$this->batchTaskService->isItemIndexProcessed($batchTask, $itemIndex)) {
  95.                 $this->batchTaskService->markItemIndexAsProcessed($batchTask, $itemIndex);
  96.             }
  97.         }
  99.         $this->checkBatchTaskFinished($batchTask, $message);
  100.     }
  102.     public function onWorkerMessageHandled(WorkerMessageHandledEvent $event)
  103.     {
  104.         $message = $event->getEnvelope()->getMessage();
  105.         if (!$message instanceof BatchTaskMessageInterface) {
  106.             return;
  107.         }
  109.         if ($message instanceof SplittedBatchTaskMessageInterface) {
  110.             return;
  111.         }
  113.         if (!$batchTask = $this->batchTaskService->getTaskById($message->getTaskId())) {
  114.             return;
  115.         }
  117.         if ($message instanceof SequentialBatchTaskMessageInterface && $message->hasRemainingItems()) {
  118.             $remainingMessage = $message->createRemainingMessage($this->batchTaskService);
  119.             $this->messageBus->dispatch($remainingMessage);
  121.             return;
  122.         }
  124.         $this->checkBatchTaskFinished($batchTask, $message);
  125.     }
  127.     public function onTerminate(TerminateEvent $event)
  128.     {
  129.         $this->batchTaskService->terminateBatchTask();
  130.     }
  132.     public function setTerminateBatchTask(BatchTask $batchTask)
  133.     {
  134.         $this->terminateBatchTask = $batchTask;
  135.     }
  137.     protected function checkBatchTaskFinished(BatchTask $batchTask, BatchTaskMessageInterface $message)
  138.     {
  139.         $this->batchTaskService->checkBatchTaskFinished($batchTask);
  141.         if ($message instanceof TriggerFinishedMessageBatchTaskMessageInterface) {
  142.             $finishedMessage = $message->createFinishedMessage();
  143.             $this->messageBus->dispatch($finishedMessage);
  144.         }
  145.     }
  146. }