espocrm-extensions/espocrm-base
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

job refactoring

Browse Source
This commit is contained in:
Yuri Kuznetsov
2021-03-23 13:25:02 +02:00
parent 88e82a511b
commit 9ef3ba9342
9 changed files with 924 additions and 551 deletions
Download Patch File Download Diff File Expand all files Collapse all files
application/Espo/Core/Job/AsyncPoolFactory.php
+58
View File
@@ -0,0 +1,58 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace Espo\Core\Job;
use Espo\Core\Utils\Config;
use Spatie\Async\Pool;
class AsyncPoolFactory
{
private $config;
public function __construct(Config $config)
{
$this->config = $config;
}
public function isSupported() : bool
{
return Pool::isSupported();
}
public function create() : Pool
{
return Pool
::create()
->autoload(getcwd() . '/vendor/autoload.php')
->concurrency($this->config->get('jobPoolConcurrencyNumber'))
->timeout($this->config->get('jobPeriodForActiveProcess'));
}
}
application/Espo/Core/Job/JobManager.php
+43 -421
View File
@@ -30,48 +30,23 @@
namespace Espo\Core\Job;
use Espo\Core\{
Exceptions\Error,
ServiceFactory,
Utils\Config,
Utils\File\Manager as FileManager,
Utils\System,
ORM\EntityManager,
Utils\Log,
Job\JobTargeted,
};
use Espo\{
Entities\Job as JobEntity,
Entities\ScheduledJob as ScheduledJobEntity,
};
use Spatie\Async\Pool as AsyncPool;
use Cron\CronExpression;
use Exception;
use Throwable;
/**
* Handles running jobs.
* Handles processing jobs.
*/
class JobManager
{
private $queueUtil;
private $scheduleUtil;
private $useProcessPool = false;
private $asSoonAsPossibleSchedulingList = [
'*',
'* *',
'* * *',
'* * * *',
'* * * * *',
'* * * * * *',
];
const PENDING = 'Pending';
const READY = 'Ready';
@@ -82,6 +57,8 @@ class JobManager
const FAILED = 'Failed';
private $useProcessPool = false;
protected $lastRunTimeFile = 'data/cache/application/cronLastRunTime.php';
private $config;
@@ -90,33 +67,41 @@ class JobManager
private $entityManager;
private $serviceFactory;
private $jobFactory;
private $jobRunner;
private $log;
private $scheduleProcessor;
private $queueUtil;
private $asyncPoolFactory;
private $queueProcessorFactory;
public function __construct(
Config $config,
FileManager $fileManager,
EntityManager $entityManager,
ServiceFactory $serviceFactory,
JobFactory $jobFactory,
JobRunner $jobRunner,
Log $log,
ScheduleProcessor $scheduleProcessor,
QueueUtil $queueUtil,
ScheduleUtil $scheduleUtil
AsyncPoolFactory $asyncPoolFactory,
QueueProcessorFactory $queueProcessorFactory
) {
$this->config = $config;
$this->fileManager = $fileManager;
$this->entityManager = $entityManager;
$this->serviceFactory = $serviceFactory;
$this->jobFactory = $jobFactory;
$this->jobRunner = $jobRunner;
$this->log = $log;
$this->scheduleProcessor = $scheduleProcessor;
$this->queueUtil = $queueUtil;
$this->scheduleUtil = $scheduleUtil;
$this->asyncPoolFactory = $asyncPoolFactory;
$this->queueProcessorFactory = $queueProcessorFactory;
if ($this->config->get('jobRunInParallel')) {
if (AsyncPool::isSupported()) {
if ($this->asyncPoolFactory->isSupported()) {
$this->useProcessPool = true;
}
else {
@@ -162,16 +147,6 @@ class JobManager
return false;
}
protected function useProcessPool() : bool
{
return $this->useProcessPool;
}
public function setUseProcessPool(bool $useProcessPool) : void
{
$this->useProcessPool = $useProcessPool;
}
/**
* Process jobs. Jobs will be created according scheduling. Then pending jobs will be processed.
* This method supposed to be called on every Cron run or loop iteration of the Daemon.
@@ -189,11 +164,11 @@ class JobManager
$this->queueUtil->markJobsFailed();
$this->queueUtil->updateFailedJobAttempts();
$this->createJobsFromScheduledJobs();
$this->scheduleProcessor->createJobsFromScheduledJobs();
$this->queueUtil->removePendingJobDuplicates();
$this->processPendingJobs();
$this->processMainQueue();
}
/**
@@ -201,123 +176,30 @@ class JobManager
*/
public function processQueue(string $queue, int $limit) : void
{
$this->processPendingJobs($queue, $limit, true, true);
$params = QueueProcessorParams
::fromNothing()
->withQueue($queue)
->withLimit($limit)
->withUseProcessPool(false)
->withNoLock(true);
$processor = $this->queueProcessorFactory->create($params);
$processor->process();
}
protected function processPendingJobs(
?string $queue = null,
?int $limit = null,
bool $poolDisabled = false,
bool $noLock = false
) : void {
if (is_null($limit)) {
$limit = intval($this->config->get('jobMaxPortion', 0));
}
$pendingJobList = $this->queueUtil->getPendingJobList($queue, $limit);
$useProcessPool = $this->useProcessPool();
if ($poolDisabled) {
$useProcessPool = false;
}
$pool = null;
if ($useProcessPool) {
$pool = AsyncPool
::create()
->autoload(getcwd() . '/vendor/autoload.php')
->concurrency($this->config->get('jobPoolConcurrencyNumber'))
->timeout($this->config->get('jobPeriodForActiveProcess'));
}
foreach ($pendingJobList as $job) {
$this->processPendingJob($job, $pool, $noLock);
}
if ($useProcessPool) {
$pool->wait();
}
}
protected function processPendingJob(JobEntity $job, $pool = null, bool $noLock = false) : void
private function processMainQueue() : void
{
$useProcessPool = (bool) $pool;
$limit = (int) $this->config->get('jobMaxPortion', 0);
$lockTable = (bool) $job->getScheduledJobId() && !$noLock;
$params = QueueProcessorParams
::fromNothing()
->withUseProcessPool($this->useProcessPool)
->withLimit($limit);
$skip = false;
$processor = $this->queueProcessorFactory->create($params);
if (!$noLock) {
if ($lockTable) {
// MySQL doesn't allow to lock non-existent rows. We resort to locking an entire table.
$this->entityManager->getLocker()->lockExclusive('Job');
}
else {
$this->entityManager->getTransactionManager()->start();
}
}
if ($noLock || $this->queueUtil->isJobPending($job->id)) {
if (
$job->getScheduledJobId() &&
$this->queueUtil->isScheduledJobRunning(
$job->getScheduledJobId(),
$job->getTargetId(),
$job->getTargetType()
)
) {
$skip = true;
}
} else {
$skip = true;
}
if ($skip && !$noLock) {
if ($lockTable) {
$this->entityManager->getLocker()->rollback();
}
else {
$this->entityManager->getTransactionManager()->rollback();
}
}
if ($skip) {
return;
}
$job->set('startedAt', date('Y-m-d H:i:s'));
if ($useProcessPool) {
$job->set('status', self::READY);
}
else {
$job->set('status', self::RUNNING);
$job->set('pid', System::getPid());
}
$this->entityManager->saveEntity($job);
if (!$noLock) {
if ($lockTable) {
$this->entityManager->getLocker()->commit();
}
else {
$this->entityManager->getTransactionManager()->commit();
}
}
if ($useProcessPool) {
$task = new JobTask($job->id);
$pool->add($task);
return;
}
$this->runJobInternal($job);
$processor->process();
}
/**
@@ -325,30 +207,7 @@ class JobManager
*/
public function runJobById(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() !== self::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', self::RUNNING);
$job->set('pid', System::getPid());
$this->entityManager->saveEntity($job);
$this->runJobInternal($job);
$this->jobRunner->runById($id);
}
/**
@@ -358,243 +217,6 @@ class JobManager
*/
public function runJob(JobEntity $job) : void
{
$this->runJobInternal($job, true);
}
protected function runJobInternal(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;
if ($e->getCode() === -1) {
$job->set('attempts', 0);
$skipLog = true;
}
else {
$this->log->error(
"JobManager: Failed job running, job '{$job->id}'. " .
$e->getMessage() . "; at " . $e->getFile() . ":" . $e->getLine() . "."
);
}
if ($throwException) {
$exception = $e;
}
}
$status = $isSuccess ? self::SUCCESS : self::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 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();
}
$service->$methodName($job->getData(), $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 createJobsFromScheduledJobs() : void
{
$activeScheduledJobList = $this->scheduleUtil->getActiveScheduledJobList();
$runningScheduledJobIdList = $this->queueUtil->getRunningScheduledJobIdList();
foreach ($activeScheduledJobList as $scheduledJob) {
try {
$this->createJobsFromScheduledJob($scheduledJob, $runningScheduledJobIdList);
}
catch (Throwable $e) {
$id = $scheduledJob->getId();
$this->log->error("Scheduled Job '{$id}': " . $e->getMessage());
}
}
}
private function createJobsFromScheduledJob(
ScheduledJobEntity $scheduledJob, array $runningScheduledJobIdList
) : void {
$scheduling = $scheduledJob->getScheduling();
$id = $scheduledJob->getId();
$asSoonAsPossible = in_array($scheduling, $this->asSoonAsPossibleSchedulingList);
if ($asSoonAsPossible) {
$executeTime = date('Y-m-d H:i:s');
}
else {
try {
$cronExpression = CronExpression::factory($scheduling);
}
catch (Exception $e) {
$this->log->error(
"Scheduled Job '{$id}': Scheduling expression error: " .
$e->getMessage() . '.'
);
return;
}
try {
$executeTime = $cronExpression->getNextRunDate()->format('Y-m-d H:i:s');
}
catch (Exception $e) {
$this->log->error(
"Scheduled Job '{$id}': Unsupported scheduling expression '{$scheduling}'."
);
return;
}
$jobAlreadyExists = $this->queueUtil->hasScheduledJobOnMinute($id, $executeTime);
if ($jobAlreadyExists) {
return;
}
}
$jobName = $scheduledJob->getJob();
if ($this->jobFactory->isPreparable($jobName)) {
$jobObj = $this->jobFactory->create($jobName);
$jobObj->prepare($scheduledJob, $executeTime);
return;
}
if (in_array($id, $runningScheduledJobIdList)) {
return;
}
$pendingCount = $this->queueUtil->getPendingCountByScheduledJobId($id);
if ($asSoonAsPossible) {
if ($pendingCount > 0) {
return;
}
}
else {
if ($pendingCount > 1) {
return;
}
}
$this->entityManager->createEntity('Job', [
'name' => $scheduledJob->getName(),
'status' => self::PENDING,
'scheduledJobId' => $id,
'executeTime' => $executeTime,
]);
$this->jobRunner->runThrowingException($job);
}
}
application/Espo/Core/Job/JobRunner.php
+252
View File
@@ -0,0 +1,252 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace Espo\Core\Job;
use Espo\Core\{
Exceptions\Error,
Utils\Log,
ORM\EntityManager,
ServiceFactory,
Utils\System,
};
use Espo\Entities\Job as JobEntity;
use Throwable;
class JobRunner
{
private $jobFactory;
private $scheduleUtil;
private $entityManager;
private $serviceFactory;
private $log;
public function __construct(
JobFactory $jobFactory,
ScheduleUtil $scheduleUtil,
EntityManager $entityManager,
ServiceFactory $serviceFactory,
Log $log
) {
$this->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());
}
}
application/Espo/Core/Job/QueueProcessor.php
+170
View File
@@ -0,0 +1,170 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace Espo\Core\Job;
use Espo\Core\{
Utils\System,
ORM\EntityManager,
};
use Espo\{
Entities\Job as JobEntity,
};
use Spatie\Async\Pool as AsyncPool;
class QueueProcessor
{
private $params;
private $queueUtil;
private $jobRunner;
private $asyncPoolFactory;
private $entityManager;
public function __construct(
QueueProcessorParams $params,
QueueUtil $queueUtil,
JobRunner $jobRunner,
AsyncPoolFactory $asyncPoolFactory,
EntityManager $entityManager
) {
$this->params = $params;
$this->queueUtil = $queueUtil;
$this->jobRunner = $jobRunner;
$this->asyncPoolFactory = $asyncPoolFactory;
$this->entityManager = $entityManager;
}
public function process() : void
{
$pool = null;
if ($this->params->useProcessPool()) {
$pool = $this->asyncPoolFactory->create();
}
$pendingJobList = $this->queueUtil->getPendingJobList(
$this->params->getQueue(),
$this->params->getLimit()
);
foreach ($pendingJobList as $job) {
$this->processJob($job, $pool);
}
if ($pool) {
$pool->wait();
}
}
protected function processJob(JobEntity $job, ?AsyncPool $pool = null) : void
{
$useProcessPool = $this->params->useProcessPool();
$noLock = $this->params->noLock();
$lockTable = (bool) $job->getScheduledJobId() && !$noLock;
$skip = false;
if (!$noLock) {
if ($lockTable) {
// MySQL doesn't allow to lock non-existent rows. We resort to locking an entire table.
$this->entityManager->getLocker()->lockExclusive('Job');
}
else {
$this->entityManager->getTransactionManager()->start();
}
}
if ($noLock || $this->queueUtil->isJobPending($job->id)) {
if (
$job->getScheduledJobId() &&
$this->queueUtil->isScheduledJobRunning(
$job->getScheduledJobId(),
$job->getTargetId(),
$job->getTargetType()
)
) {
$skip = true;
}
} else {
$skip = true;
}
if ($skip && !$noLock) {
if ($lockTable) {
$this->entityManager->getLocker()->rollback();
}
else {
$this->entityManager->getTransactionManager()->rollback();
}
}
if ($skip) {
return;
}
$job->set('startedAt', date('Y-m-d H:i:s'));
if ($useProcessPool) {
$job->set('status', JobManager::READY);
}
else {
$job->set('status', JobManager::RUNNING);
$job->set('pid', System::getPid());
}
$this->entityManager->saveEntity($job);
if (!$noLock) {
if ($lockTable) {
$this->entityManager->getLocker()->commit();
}
else {
$this->entityManager->getTransactionManager()->commit();
}
}
if ($useProcessPool) {
$task = new JobTask($job->id);
$pool->add($task);
return;
}
$this->jobRunner->run($job);
}
}
application/Espo/Core/Job/QueueProcessorFactory.php
+51
View File
@@ -0,0 +1,51 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace Espo\Core\Job;
use Espo\Core\{
InjectableFactory,
};
class QueueProcessorFactory
{
private $injectableFactory;
public function __construct(InjectableFactory $injectableFactory)
{
$this->injectableFactory = $injectableFactory;
}
public function create(QueueProcessorParams $params) : QueueProcessor
{
return $this->injectableFactory->createWith(QueueProcessor::class, [
'params' => $params,
]);
}
}
application/Espo/Core/Job/QueueProcessorParams.php
+102
View File
@@ -0,0 +1,102 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace Espo\Core\Job;
class QueueProcessorParams
{
private $useProcessPool = false;
private $noLock = false;
private $queue = null;
private $limit = 0;
public function withUseProcessPool(bool $useProcessPool) : self
{
$obj = clone $this;
$obj->useProcessPool = $useProcessPool;
return $obj;
}
public function withNoLock(bool $noLock) : self
{
$obj = clone $this;
$obj->noLock = $noLock;
return $obj;
}
public function withQueue(?string $queue) : self
{
$obj = clone $this;
$obj->queue = $queue;
return $obj;
}
public function withLimit(int $limit) : self
{
$obj = clone $this;
$obj->limit = $limit;
return $obj;
}
public function useProcessPool() : bool
{
return $this->useProcessPool;
}
public function noLock() : bool
{
return $this->noLock;
}
public function getQueue() : ?string
{
return $this->queue;
}
public function getLimit() : int
{
return $this->limit;
}
public static function fromNothing() : self
{
return new self();
}
}
application/Espo/Core/Job/ScheduleProcessor.php
+178
View File
@@ -0,0 +1,178 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace Espo\Core\Job;
use Espo\Core\{
ORM\EntityManager,
Utils\Log,
};
use Espo\{
Entities\ScheduledJob as ScheduledJobEntity,
};
use Cron\CronExpression;
use Throwable;
use Exception;
class ScheduleProcessor
{
private $asSoonAsPossibleSchedulingList = [
'*',
'* *',
'* * *',
'* * * *',
'* * * * *',
'* * * * * *',
];
private $log;
private $entityManager;
private $queueUtil;
private $scheduleUtil;
private $jobFactory;
public function __construct(
Log $log,
EntityManager $entityManager,
QueueUtil $queueUtil,
ScheduleUtil $scheduleUtil,
JobFactory $jobFactory
) {
$this->log = $log;
$this->entityManager = $entityManager;
$this->queueUtil = $queueUtil;
$this->scheduleUtil = $scheduleUtil;
$this->jobFactory = $jobFactory;
}
public function createJobsFromScheduledJobs() : void
{
$activeScheduledJobList = $this->scheduleUtil->getActiveScheduledJobList();
$runningScheduledJobIdList = $this->queueUtil->getRunningScheduledJobIdList();
foreach ($activeScheduledJobList as $scheduledJob) {
try {
$this->createJobsFromScheduledJob($scheduledJob, $runningScheduledJobIdList);
}
catch (Throwable $e) {
$id = $scheduledJob->getId();
$this->log->error("Scheduled Job '{$id}': " . $e->getMessage());
}
}
}
private function createJobsFromScheduledJob(
ScheduledJobEntity $scheduledJob,
array $runningScheduledJobIdList
) : void {
$scheduling = $scheduledJob->getScheduling();
$id = $scheduledJob->getId();
$asSoonAsPossible = in_array($scheduling, $this->asSoonAsPossibleSchedulingList);
if ($asSoonAsPossible) {
$executeTime = date('Y-m-d H:i:s');
}
else {
try {
$cronExpression = CronExpression::factory($scheduling);
}
catch (Exception $e) {
$this->log->error(
"Scheduled Job '{$id}': Scheduling expression error: " .
$e->getMessage() . '.'
);
return;
}
try {
$executeTime = $cronExpression->getNextRunDate()->format('Y-m-d H:i:s');
}
catch (Exception $e) {
$this->log->error(
"Scheduled Job '{$id}': Unsupported scheduling expression '{$scheduling}'."
);
return;
}
$jobAlreadyExists = $this->queueUtil->hasScheduledJobOnMinute($id, $executeTime);
if ($jobAlreadyExists) {
return;
}
}
$jobName = $scheduledJob->getJob();
if ($this->jobFactory->isPreparable($jobName)) {
$jobObj = $this->jobFactory->create($jobName);
$jobObj->prepare($scheduledJob, $executeTime);
return;
}
if (in_array($id, $runningScheduledJobIdList)) {
return;
}
$pendingCount = $this->queueUtil->getPendingCountByScheduledJobId($id);
if ($asSoonAsPossible) {
if ($pendingCount > 0) {
return;
}
}
else {
if ($pendingCount > 1) {
return;
}
}
$this->entityManager->createEntity('Job', [
'name' => $scheduledJob->getName(),
'status' => JobManager::PENDING,
'scheduledJobId' => $id,
'executeTime' => $executeTime,
]);
}
}
tests/unit/Espo/Core/Job/JobManagerTest.php
-130
View File
@@ -1,130 +0,0 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace tests\unit\Espo\Core\Job;
use tests\unit\ReflectionHelper;
use Espo\Core\{
Job\JobManager,
Job\JobFactory,
ServiceFactory,
Utils\Config,
Utils\File\Manager as FileManager,
ORM\EntityManager,
Utils\Log,
Job\QueueUtil,
Job\ScheduleUtil,
};
class JobManagerTest extends \PHPUnit\Framework\TestCase
{
protected function setUp() : void
{
$this->serviceFactory = $this->createMock(ServiceFactory::class);
$this->config = $this->createMock(Config::class);
$this->fileManager = $this->createMock(FileManager::class);
$this->jobFactory = $this->createMock(JobFactory::class);
$this->entityManager = $this->createMock(EntityManager::class);
$this->log = $this->createMock(Log::class);
$this->queueUtil = $this->createMock(QueueUtil::class);
$this->scheduleUtil = $this->createMock(ScheduleUtil::class);
$this->manager = new JobManager(
$this->config,
$this->fileManager,
$this->entityManager,
$this->serviceFactory,
$this->jobFactory,
$this->log,
$this->queueUtil,
$this->scheduleUtil
);
$this->reflection = new ReflectionHelper($this->manager);
}
protected function tearDown() : void
{
$this->manager = NULL;
}
public function testCheckLastRunTimeFileDoesnotExist()
{
$this->fileManager
->expects($this->once())
->method('getPhpContents')
->will($this->returnValue(false));
$this->config
->expects($this->any())
->method('get')
->will($this->returnValue(50));
$this->assertTrue($this->reflection->invokeMethod('checkLastRunTime', []));
}
public function testCheckLastRunTime()
{
$this->fileManager
->expects($this->once())
->method('getPhpContents')
->will(
$this->returnValue([
'time' => time() - 60,
])
);
$this->config
->expects($this->any())
->method('get')
->will($this->returnValue(50));
$this->assertTrue( $this->reflection->invokeMethod('checkLastRunTime', []));
}
public function testCheckLastRunTimeTooFrequency()
{
$this->fileManager
->expects($this->once())
->method('getPhpContents')
->will(
$this->returnValue([
'time' => time() - 49,
])
);
$this->config
->expects($this->exactly(1))
->method('get')
->will($this->returnValue(50));
$this->assertFalse($this->reflection->invokeMethod('checkLastRunTime', []));
}
}
tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php
+70
View File
@@ -0,0 +1,70 @@
<?php
/************************************************************************
* This file is part of EspoCRM.
*
* EspoCRM - Open Source CRM application.
* Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
* Website: https://www.espocrm.com
*
* EspoCRM is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* EspoCRM is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with EspoCRM. If not, see http://www.gnu.org/licenses/.
*
* The interactive user interfaces in modified source and object code versions
* of this program must display Appropriate Legal Notices, as required under
* Section 5 of the GNU General Public License version 3.
*
* In accordance with Section 7(b) of the GNU General Public License version 3,
* these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
************************************************************************/
namespace tests\unit\Espo\Core\Console;
use Espo\Core\{
Job\QueueProcessorParams,
};
class QueueProcessorParamsTest extends \PHPUnit\Framework\TestCase
{
protected function setUp() : void
{
}
public function testParams1()
{
$params = QueueProcessorParams
::fromNothing()
->withLimit(10);
$this->assertFalse($params->useProcessPool());
$this->assertFalse($params->noLock());
$this->assertEquals(10, $params->getLimit());
$this->assertNull($params->getQueue());
}
public function testParams2()
{
$params = QueueProcessorParams
::fromNothing()
->withLimit(10)
->withUseProcessPool(true)
->withNoLock(true)
->withQueue('q0');
$this->assertTrue($params->useProcessPool());
$this->assertTrue($params->noLock());
$this->assertEquals('q0', $params->getQueue());
}
}