jobFactory = $jobFactory; $this->scheduleUtil = $scheduleUtil; $this->entityManager = $entityManager; $this->serviceFactory = $serviceFactory; $this->log = $log; } /** * Run a job entity. Does not throw exceptions. */ public function run(JobEntity $job): void { $this->runInternal($job, false); } /** * Run a job entity. Throws exceptions. * * @throws Throwable */ public function runThrowingException(JobEntity $job): void { $this->runInternal($job, true); } /** * Run a job by ID. A job must have status 'Ready'. * Used when running jobs in parallel processes. */ public function runById(string $id): void { if ($id === '') { throw new Error(); } $job = $this->entityManager->getEntity('Job', $id); if (!$job) { throw new Error("Job {$id} not found."); } if ($job->getStatus() !== JobManager::READY) { throw new Error("Can't run job {$id} with no status Ready."); } if (!$job->getStartedAt()) { $job->set('startedAt', date('Y-m-d H:i:s')); } $job->set('status', JobManager::RUNNING); $job->set('pid', System::getPid()); $this->entityManager->saveEntity($job); $this->run($job); } private function runInternal(JobEntity $job, bool $throwException = false): void { $isSuccess = true; $skipLog = false; $exception = null; try { if ($job->getScheduledJobId()) { $this->runScheduledJob($job); } else if ($job->getJob()) { $this->runJobByName($job); } else if ($job->getServiceName()) { $this->runService($job); } else { $id = $job->getId(); throw new Error("Not runnable job '{$id}'."); } } catch (Throwable $e) { $isSuccess = false; $this->log->error( "JobManager: Failed job running, job '{$job->id}'. " . $e->getMessage() . "; at " . $e->getFile() . ":" . $e->getLine() . "." ); if ($throwException) { $exception = $e; } } $status = $isSuccess ? JobManager::SUCCESS : JobManager::FAILED; $job->set('status', $status); if ($isSuccess) { $job->set('executedAt', date('Y-m-d H:i:s')); } $this->entityManager->saveEntity($job); if ($throwException && $exception) { throw new $exception; } if ($job->getScheduledJobId() && !$skipLog) { $this->scheduleUtil->addLogRecord( $job->getScheduledJobId(), $status, null, $job->getTargetId(), $job->getTargetType() ); } } protected function runJobByName(JobEntity $job): void { $jobName = $job->getJob(); $obj = $this->jobFactory->create($jobName); if ($obj instanceof JobTargeted) { $obj->run($job->getTargetType(), $job->getTargetId(), $job->getData()); return; } if (!method_exists($obj, 'run')) { throw new Error("No 'run' method in job '{$jobName}'."); } $obj->run(); } protected function runScheduledJob(JobEntity $job): void { $jobName = $job->getScheduledJobJob(); if (!$jobName) { throw new Error( "Can't run job '" . $job->getId() . "'. Not a scheduled job." ); } $obj = $this->jobFactory->create($jobName); if ($obj instanceof JobTargeted) { $obj->run($job->getTargetType(), $job->getTargetId(), $job->getData()); return; } if (!method_exists($obj, 'run')) { throw new Error("No 'run' method in job '{$jobName}'."); } $obj->run(); } protected function runService(JobEntity $job): void { $serviceName = $job->getServiceName(); if (!$serviceName) { throw new Error("Job with empty serviceName."); } if (!$this->serviceFactory->checkExists($serviceName)) { throw new Error(); } $service = $this->serviceFactory->create($serviceName); $methodName = $job->getMethodName(); if (!$methodName) { throw new Error('Job with empty methodName.'); } if (!method_exists($service, $methodName)) { throw new Error("No method '{$methodName}' in service '{$serviceName}'."); } $service->$methodName($job->getData(), $job->getTargetId(), $job->getTargetType()); } }