jobManager = $jobManager; $this->entityManager = $entityManager; $this->config = $config; } public function run(JobData $data): void { $limit = $this->config->get('jobGroupMaxPortion') ?? self::PORTION_NUMBER; $group = $data->get('group'); $this->jobManager->processGroup($group, $limit); } public function prepare(ScheduledJobData $data, DateTimeImmutable $executeTime): void { $groupList = []; $query = $this->entityManager ->getQueryBuilder() ->select('group') ->from(JobEntity::ENTITY_TYPE) ->where([ 'status' => JobStatus::PENDING, 'queue' => null, 'group!=' => null, 'executeTime<=' => $executeTime->format(DateTime::SYSTEM_DATE_TIME_FORMAT), ]) ->group('group') ->build(); $sth = $this->entityManager->getQueryExecutor()->execute($query); while ($row = $sth->fetch()) { $group = $row['group']; if ($group === null) { continue; } $groupList[] = $group; } if (!count($groupList)) { return; } foreach ($groupList as $group) { $existingJob = $this->entityManager ->getRDBRepository(JobEntity::ENTITY_TYPE) ->select('id') ->where([ 'scheduledJobId' => $data->getId(), 'targetGroup' => $group, 'status' => [ JobStatus::RUNNING, JobStatus::READY, JobStatus::PENDING, ], ]) ->findOne(); if ($existingJob) { continue; } $name = $data->getName() . ' :: ' . $group; $this->entityManager->createEntity(JobEntity::ENTITY_TYPE, [ 'scheduledJobId' => $data->getId(), 'executeTime' => $executeTime->format(DateTime::SYSTEM_DATE_TIME_FORMAT), 'name' => $name, 'data' => [ 'group' => $group, ], 'targetGroup' => $group, ]); } } }