ref

2022-11-03 11:38:23 +02:00
parent b2b958c946
commit addda0d9a8
3 changed files with 86 additions and 81 deletions
@@ -29,33 +29,23 @@

 namespace Espo\Core\Job;

-use Espo\Core\{
-    Utils\Config,
-    ORM\EntityManager,
-    Utils\System,
-    Utils\DateTime as DateTimeUtil,
-};
-
+use Espo\Core\ORM\EntityManager;
+use Espo\Core\Utils\Config;
+use Espo\Core\Utils\DateTime as DateTimeUtil;
+use Espo\Core\Utils\System;
 use Espo\Core\Job\Job\Status;
-
 use Espo\Entities\Job as JobEntity;
-
 use Espo\ORM\Collection;
-
 use DateTime;

 class QueueUtil
 {
    private Config $config;
-
    private EntityManager $entityManager;
-
    private ScheduleUtil $scheduleUtil;
-
    private MetadataProvider $metadataProvider;

    private const NOT_EXISTING_PROCESS_PERIOD = 300;
-
    private const READY_NOT_STARTED_PERIOD = 60;

    public function __construct(
@@ -85,17 +75,16 @@ class QueueUtil
            return false;
        }

-        return $job->get('status') === Status::PENDING;
+        return $job->getStatus() === Status::PENDING;
    }

    /**
-     * @return JobEntity[]
-     * @phpstan-return Collection<JobEntity>
+     * @return Collection<JobEntity>
     */
    public function getPendingJobList(?string $queue = null, ?string $group = null, int $limit = 0): Collection
    {
        $builder = $this->entityManager
-            ->getRDBRepository(JobEntity::ENTITY_TYPE)
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select([
                'id',
                'scheduledJobId',
@@ -122,10 +111,7 @@ class QueueUtil
            $builder->limit(0, $limit);
        }

-        /** @var Collection<JobEntity> $collection */
-        $collection = $builder->find();
-
-        return $collection;
+        return $builder->find();
    }

    public function isScheduledJobRunning(
@@ -150,7 +136,7 @@ class QueueUtil
        }

        return (bool) $this->entityManager
-            ->getRDBRepository(JobEntity::ENTITY_TYPE)
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select(['id'])
            ->where($where)
            ->findOne();
@@ -164,7 +150,7 @@ class QueueUtil
        $list = [];

        $jobList = $this->entityManager
-            ->getRDBRepository(JobEntity::ENTITY_TYPE)
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select(['scheduledJobId'])
            ->leftJoin('scheduledJob')
            ->where([
@@ -179,7 +165,13 @@ class QueueUtil
            ->find();

        foreach ($jobList as $job) {
-            $list[] = $job->get('scheduledJobId');
+            $scheduledJobId =  $job->getScheduledJobId();
+
+            if (!$scheduledJobId) {
+                continue;
+            }
+
+            $list[] = $scheduledJobId;
        }

        return $list;
@@ -193,7 +185,7 @@ class QueueUtil
        $toString = $dateObj->format('Y-m-d H:i:59');

        $job = $this->entityManager
-            ->getRDBRepository('Job')
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select(['id'])
            ->where([
                'scheduledJobId' => $scheduledJobId,
@@ -213,15 +205,13 @@ class QueueUtil

    public function getPendingCountByScheduledJobId(string $scheduledJobId): int
    {
-        $countPending = $this->entityManager
-            ->getRDBRepository(JobEntity::ENTITY_TYPE)
+        return $this->entityManager
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->where([
                'scheduledJobId' => $scheduledJobId,
                'status' => Status::PENDING,
            ])
            ->count();
-
-        return $countPending;
    }

    public function markJobsFailed(): void
@@ -232,7 +222,7 @@ class QueueUtil
        $this->markJobsFailedByPeriod();
    }

-    protected function markJobsFailedByNotExistingProcesses(): void
+    private function markJobsFailedByNotExistingProcesses(): void
    {
        $timeThreshold = time() - $this->config->get(
            'jobPeriodForNotExistingProcess',
@@ -242,7 +232,7 @@ class QueueUtil
        $dateTimeThreshold = date(DateTimeUtil::SYSTEM_DATE_TIME_FORMAT, $timeThreshold);

        $runningJobList = $this->entityManager
-            ->getRDBRepository('Job')
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select([
                'id',
                'scheduledJobId',
@@ -261,7 +251,9 @@ class QueueUtil
        $failedJobList = [];

        foreach ($runningJobList as $job) {
-            if ($job->get('pid') && !System::isProcessActive($job->get('pid'))) {
+            $pid = $job->getPid();
+
+            if ($pid && !System::isProcessActive($pid)) {
                $failedJobList[] = $job;
            }
        }
@@ -269,15 +261,15 @@ class QueueUtil
        $this->markJobListFailed($failedJobList);
    }

-    protected function markJobsFailedReadyNotStarted(): void
+    private function markJobsFailedReadyNotStarted(): void
    {
        $timeThreshold = time() -
-            $this->config->get('jobPeriodForReadyNotStarted', SELF::READY_NOT_STARTED_PERIOD);
+            $this->config->get('jobPeriodForReadyNotStarted', self::READY_NOT_STARTED_PERIOD);

        $dateTimeThreshold = date(DateTimeUtil::SYSTEM_DATE_TIME_FORMAT, $timeThreshold);

        $failedJobList = $this->entityManager
-            ->getRDBRepository('Job')
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select([
                'id',
                'scheduledJobId',
@@ -309,7 +301,7 @@ class QueueUtil
        $dateTimeThreshold = date(DateTimeUtil::SYSTEM_DATE_TIME_FORMAT, $timeThreshold);

        $runningJobList = $this->entityManager
-            ->getRDBRepository(JobEntity::ENTITY_TYPE)
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select([
                'id',
                'scheduledJobId',
@@ -334,7 +326,9 @@ class QueueUtil
                continue;
            }

-            if (!$job->get('pid') || !System::isProcessActive($job->get('pid'))) {
+            $pid = $job->getPid();
+
+            if (!$pid || !System::isProcessActive($pid)) {
                $failedJobList[] = $job;
            }
        }
@@ -357,7 +351,8 @@ class QueueUtil
            $jobIdList[] = $job->getId();
        }

-        $updateQuery = $this->entityManager->getQueryBuilder()
+        $updateQuery = $this->entityManager
+            ->getQueryBuilder()
            ->update()
            ->in(JobEntity::ENTITY_TYPE)
            ->set([
@@ -372,16 +367,18 @@ class QueueUtil
        $this->entityManager->getQueryExecutor()->execute($updateQuery);

        foreach ($jobList as $job) {
-            if (!$job->get('scheduledJobId')) {
+            $scheduledJobId = $job->getScheduledJobId();
+
+            if (!$scheduledJobId) {
                continue;
            }

            $this->scheduleUtil->addLogRecord(
-                $job->get('scheduledJobId'),
+                $scheduledJobId,
                Status::FAILED,
-                $job->get('startedAt'),
-                $job->get('targetId'),
-                $job->get('targetType')
+                $job->getStartedAt(),
+                $job->getTargetId(),
+                $job->getTargetType()
            );
        }
    }
@@ -412,16 +409,18 @@ class QueueUtil
        $scheduledJobIdList = [];

        foreach ($duplicateJobList as $duplicateJob) {
-            if (!$duplicateJob->get('scheduledJobId')) {
+            $scheduledJobId = $duplicateJob->getScheduledJobId();
+
+            if (!$scheduledJobId) {
                continue;
            }

-            $scheduledJobIdList[] = $duplicateJob->get('scheduledJobId');
+            $scheduledJobIdList[] = $scheduledJobId;
        }

        foreach ($scheduledJobIdList as $scheduledJobId) {
            $toRemoveJobList = $this->entityManager
-                ->getRDBRepository(JobEntity::ENTITY_TYPE)
+                ->getRDBRepositoryByClass(JobEntity::class)
                ->select(['id'])
                ->where([
                    'scheduledJobId' => $scheduledJobId,
@@ -460,7 +459,7 @@ class QueueUtil
    public function updateFailedJobAttempts(): void
    {
        $jobCollection = $this->entityManager
-            ->getRDBRepository(JobEntity::ENTITY_TYPE)
+            ->getRDBRepositoryByClass(JobEntity::class)
            ->select([
                'id',
                'attempts',
@@ -474,8 +473,8 @@ class QueueUtil
            ->find();

         foreach ($jobCollection as $job) {
-            $failedAttempts = $job->get('failedAttempts') ?? 0;
-            $attempts = $job->get('attempts');
+            $failedAttempts = $job->getFailedAttempts();
+            $attempts = $job->getAttempts();

            $job->set([
                'status' => Status::PENDING,
@@ -30,19 +30,12 @@
 namespace Espo\Core\Job;

 use Espo\Core\Job\Preparator\Data as PreparatorData;
-
-use Espo\Core\{
-    ORM\EntityManager,
-    Utils\Log,
-    Utils\DateTime as DateTimeUtil,
-};
-
+use Espo\Core\ORM\EntityManager;
+use Espo\Core\Utils\DateTime as DateTimeUtil;
+use Espo\Core\Utils\Log;
 use Espo\Core\Job\Job\Status;
-
-use Espo\Entities\{
-    ScheduledJob as ScheduledJobEntity,
-    Job as JobEntity,
-};
+use Espo\Entities\Job as JobEntity;
+use Espo\Entities\ScheduledJob as ScheduledJobEntity;

 use Cron\CronExpression;

@@ -55,9 +48,7 @@ use DateTimeImmutable;
 */
 class ScheduleProcessor
 {
-    /**
-     * @var string[]
-     */
+    /** @var string[] */
    private $asSoonAsPossibleSchedulingList = [
        '*',
        '* *',
@@ -67,17 +58,12 @@ class ScheduleProcessor
        '* * * * * *',
    ];

-    private $log;
-
-    private $entityManager;
-
-    private $queueUtil;
-
-    private $scheduleUtil;
-
-    private $metadataProvider;
-
-    private $preparatorFactory;
+    private Log $log;
+    private EntityManager $entityManager;
+    private QueueUtil $queueUtil;
+    private ScheduleUtil $scheduleUtil;
+    private PreparatorFactory $preparatorFactory;
+    private MetadataProvider $metadataProvider;

    public function __construct(
        Log $log,
@@ -98,7 +84,6 @@ class ScheduleProcessor
    public function process(): void
    {
        $activeScheduledJobList = $this->scheduleUtil->getActiveScheduledJobList();
-
        $runningScheduledJobIdList = $this->queueUtil->getRunningScheduledJobIdList();

        foreach ($activeScheduledJobList as $scheduledJob) {
@@ -206,9 +191,7 @@ class ScheduleProcessor
            return $cronExpression->getNextRunDate()->format(DateTimeUtil::SYSTEM_DATE_TIME_FORMAT);
        }
        catch (Exception $e) {
-            $this->log->error(
-                "Scheduled Job '{$id}': Unsupported scheduling expression '{$scheduling}'."
-            );
+            $this->log->error("Scheduled Job '{$id}': Unsupported scheduling expression '{$scheduling}'.");

            return null;
        }
@@ -30,7 +30,6 @@
 namespace Espo\Entities;

 use Espo\Core\ORM\Entity;
-
 use stdClass;

 class Job extends Entity
@@ -148,4 +147,28 @@ class Job extends Entity
    {
        return $this->get('startedAt');
    }
+
+    /**
+     * Get a PID.
+     */
+    public function getPid(): ?int
+    {
+        return $this->get('pid');
+    }
+
+    /**
+     * Get a number of attempts left.
+     */
+    public function getAttempts(): int
+    {
+        return $this->get('attempts') ?? 0;
+    }
+
+    /**
+     * Get a number of failed attempts.
+     */
+    public function getFailedAttempts(): int
+    {
+        return $this->get('failedAttempts') ?? 0;
+    }
 }