src/Core/Framework/MessageQueue/ScheduledTask/Scheduler/TaskScheduler.php line 120

  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Framework\MessageQueue\ScheduledTask\Scheduler;
  5. use Shopware\Core\Defaults;
  6. use Shopware\Core\Framework\Context;
  7. use Shopware\Core\Framework\DataAbstractionLayer\EntityRepository;
  8. use Shopware\Core\Framework\DataAbstractionLayer\Search\Aggregation\Metric\MinAggregation;
  9. use Shopware\Core\Framework\DataAbstractionLayer\Search\AggregationResult\AggregationResult;
  10. use Shopware\Core\Framework\DataAbstractionLayer\Search\AggregationResult\Metric\MinResult;
  11. use Shopware\Core\Framework\DataAbstractionLayer\Search\Criteria;
  12. use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\EqualsAnyFilter;
  13. use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\EqualsFilter;
  14. use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\NotFilter;
  15. use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\RangeFilter;
  16. use Shopware\Core\Framework\Log\Package;
  17. use Shopware\Core\Framework\MessageQueue\ScheduledTask\ScheduledTask;
  18. use Shopware\Core\Framework\MessageQueue\ScheduledTask\ScheduledTaskDefinition;
  19. use Shopware\Core\Framework\MessageQueue\ScheduledTask\ScheduledTaskEntity;
  20. use Symfony\Component\DependencyInjection\ParameterBag\ParameterBagInterface;
  21. use Symfony\Component\Messenger\MessageBusInterface;
  23. #[Package('core')]
  24. final class TaskScheduler
  25. {
  26.     /**
  27.      * @internal
  28.      */
  29.     public function __construct(private readonly EntityRepository $scheduledTaskRepository, private readonly MessageBusInterface $bus, private readonly ParameterBagInterface $parameterBag)
  30.     {
  31.     }
  33.     public function queueScheduledTasks(): void
  34.     {
  35.         $criteria = $this->buildCriteriaForAllScheduledTask();
  36.         $context = Context::createDefaultContext();
  37.         $tasks = $this->scheduledTaskRepository->search($criteria, $context)->getEntities();
  39.         if (\count($tasks) === 0) {
  40.             return;
  41.         }
  43.         // Tasks **must not** be queued before their state in the database has been updated. Otherwise,
  44.         // a worker could have already fetched the task and set its state to running before it gets set to
  45.         // queued, thus breaking the task.
  46.         /** @var ScheduledTaskEntity $task */
  47.         foreach ($tasks as $task) {
  48.             $this->queueTask($task, $context);
  49.         }
  50.     }
  52.     public function getNextExecutionTime(): ?\DateTimeInterface
  53.     {
  54.         $criteria = $this->buildCriteriaForNextScheduledTask();
  55.         /** @var AggregationResult $aggregation */
  56.         $aggregation = $this->scheduledTaskRepository
  57.             ->aggregate($criteria, Context::createDefaultContext())
  58.             ->get('nextExecutionTime');
  60.         /** @var MinResult $aggregation */
  61.         if (!$aggregation instanceof MinResult) {
  62.             return null;
  63.         }
  64.         if ($aggregation->getMin() === null) {
  65.             return null;
  66.         }
  68.         return new \DateTime((string) $aggregation->getMin());
  69.     }
  71.     public function getMinRunInterval(): ?int
  72.     {
  73.         $criteria = $this->buildCriteriaForMinRunInterval();
  74.         $aggregation = $this->scheduledTaskRepository
  75.             ->aggregate($criteria, Context::createDefaultContext())
  76.             ->get('runInterval');
  78.         /** @var MinResult $aggregation */
  79.         if (!$aggregation instanceof MinResult) {
  80.             return null;
  81.         }
  82.         if ($aggregation->getMin() === null) {
  83.             return null;
  84.         }
  86.         return (int) $aggregation->getMin();
  87.     }
  89.     private function buildCriteriaForAllScheduledTask(): Criteria
  90.     {
  91.         $criteria = new Criteria();
  92.         $criteria->addFilter(
  93.             new RangeFilter(
  94.                 'nextExecutionTime',
  95.                 [
  96.                     RangeFilter::LT => (new \DateTime())->format(Defaults::STORAGE_DATE_TIME_FORMAT),
  97.                 ]
  98.             ),
  99.             new EqualsAnyFilter('status', [
  100.                 ScheduledTaskDefinition::STATUS_SCHEDULED,
  101.                 ScheduledTaskDefinition::STATUS_SKIPPED,
  102.             ])
  103.         );
  105.         return $criteria;
  106.     }
  108.     private function queueTask(ScheduledTaskEntity $taskEntity, Context $context): void
  109.     {
  110.         $taskClass = $taskEntity->getScheduledTaskClass();
  112.         if (!\is_a($taskClass, ScheduledTask::class, true)) {
  113.             throw new \RuntimeException(sprintf(
  114.                 'Tried to schedule "%s", but class does not extend ScheduledTask',
  115.                 $taskClass
  116.             ));
  117.         }
  119.         if (!$taskClass::shouldRun($this->parameterBag)) {
  120.             $this->scheduledTaskRepository->update([
  121.                 [
  122.                     'id' => $taskEntity->getId(),
  123.                     'nextExecutionTime' => $this->calculateNextExecutionTime($taskEntity),
  124.                     'status' => ScheduledTaskDefinition::STATUS_SKIPPED,
  125.                 ],
  126.             ], $context);
  128.             return;
  129.         }
  131.         $this->scheduledTaskRepository->update([
  132.             [
  133.                 'id' => $taskEntity->getId(),
  134.                 'status' => ScheduledTaskDefinition::STATUS_QUEUED,
  135.             ],
  136.         ], $context);
  138.         $task = new $taskClass();
  139.         $task->setTaskId($taskEntity->getId());
  141.         $this->bus->dispatch($task);
  142.     }
  144.     private function buildCriteriaForNextScheduledTask(): Criteria
  145.     {
  146.         $criteria = new Criteria();
  147.         $criteria->addFilter(
  148.             new EqualsAnyFilter('status', [
  149.                 ScheduledTaskDefinition::STATUS_SCHEDULED,
  150.                 ScheduledTaskDefinition::STATUS_SKIPPED,
  151.             ])
  152.         )
  153.         ->addAggregation(new MinAggregation('nextExecutionTime', 'nextExecutionTime'));
  155.         return $criteria;
  156.     }
  158.     private function buildCriteriaForMinRunInterval(): Criteria
  159.     {
  160.         $criteria = new Criteria();
  161.         $criteria->addFilter(
  162.             new NotFilter(NotFilter::CONNECTION_AND, [
  163.                 new EqualsFilter('status', ScheduledTaskDefinition::STATUS_INACTIVE),
  164.             ])
  165.         )
  166.         ->addAggregation(new MinAggregation('runInterval', 'runInterval'));
  168.         return $criteria;
  169.     }
  171.     private function calculateNextExecutionTime(ScheduledTaskEntity $taskEntity): \DateTimeImmutable
  172.     {
  173.         $now = new \DateTimeImmutable();
  175.         $nextExecutionTimeString = $taskEntity->getNextExecutionTime()->format(Defaults::STORAGE_DATE_TIME_FORMAT);
  176.         $nextExecutionTime = new \DateTimeImmutable($nextExecutionTimeString);
  177.         $newNextExecutionTime = $nextExecutionTime->modify(sprintf('+%d seconds', $taskEntity->getRunInterval()));
  179.         return $newNextExecutionTime < $now ? $now : $newNextExecutionTime;
  180.     }
  181. }