diff --git a/application/Espo/Core/Job/JobManager.php b/application/Espo/Core/Job/JobManager.php index 1b51ac6595..9af55a7ce3 100644 --- a/application/Espo/Core/Job/JobManager.php +++ b/application/Espo/Core/Job/JobManager.php @@ -30,6 +30,7 @@ namespace Espo\Core\Job; use Espo\Core\Exceptions\Error; +use Espo\Core\Job\QueueProcessor\Params; use Espo\Core\Utils\Config; use Espo\Core\Utils\File\Manager as FileManager; use Espo\Core\Utils\Log; @@ -44,7 +45,6 @@ use Throwable; class JobManager { private bool $useProcessPool = false; - protected string $lastRunTimeFile = 'data/cache/application/cronLastRunTime.php'; public function __construct( @@ -55,15 +55,13 @@ class JobManager private ScheduleProcessor $scheduleProcessor, private QueueUtil $queueUtil, private AsyncPoolFactory $asyncPoolFactory, - private QueueProcessorFactory $queueProcessorFactory + private QueueProcessor $queueProcessor ) { - if ($this->config->get('jobRunInParallel')) { if ($this->asyncPoolFactory->isSupported()) { $this->useProcessPool = true; - } - else { - $this->log->warning("JobManager: useProcessPool requires pcntl and posix extensions."); + } else { + $this->log->warning("Enabled `jobRunInParallel` parameter requires pcntl and posix extensions."); } } } @@ -93,16 +91,14 @@ class JobManager */ public function processQueue(string $queue, int $limit): void { - $params = QueueProcessorParams + $params = Params ::create() ->withQueue($queue) ->withLimit($limit) ->withUseProcessPool(false) ->withNoLock(true); - $processor = $this->queueProcessorFactory->create($params); - - $processor->process(); + $this->queueProcessor->process($params); } /** @@ -110,30 +106,26 @@ class JobManager */ public function processGroup(string $group, int $limit): void { - $params = QueueProcessorParams + $params = Params ::create() ->withGroup($group) ->withLimit($limit) ->withUseProcessPool(false) ->withNoLock(true); - $processor = $this->queueProcessorFactory->create($params); - - $processor->process(); + $this->queueProcessor->process($params); } private function processMainQueue(): void { $limit = (int) $this->config->get('jobMaxPortion', 0); - $params = QueueProcessorParams + $params = Params ::create() ->withUseProcessPool($this->useProcessPool) ->withLimit($limit); - $processor = $this->queueProcessorFactory->create($params); - - $processor->process(); + $this->queueProcessor->process($params); } /** diff --git a/application/Espo/Core/Job/QueueProcessor.php b/application/Espo/Core/Job/QueueProcessor.php index 946043bedb..4fbbc722c4 100644 --- a/application/Espo/Core/Job/QueueProcessor.php +++ b/application/Espo/Core/Job/QueueProcessor.php @@ -29,47 +29,47 @@ namespace Espo\Core\Job; -use Spatie\Async\Pool as AsyncPool; - +use Espo\Entities\Job as JobEntity; +use Espo\Core\Job\QueueProcessor\Params; use Espo\Core\ORM\EntityManager; use Espo\Core\Utils\DateTime as DateTimeUtil; use Espo\Core\Utils\System; -use Espo\Entities\Job as JobEntity; use Espo\Core\Job\Job\Status; +use Spatie\Async\Pool as AsyncPool; + class QueueProcessor { public function __construct( - private QueueProcessorParams $params, private QueueUtil $queueUtil, private JobRunner $jobRunner, private AsyncPoolFactory $asyncPoolFactory, private EntityManager $entityManager ) {} - public function process(): void + public function process(Params $params): void { - $pool = $this->params->useProcessPool() ? + $pool = $params->useProcessPool() ? $this->asyncPoolFactory->create() : null; $pendingJobList = $this->queueUtil->getPendingJobList( - $this->params->getQueue(), - $this->params->getGroup(), - $this->params->getLimit() + $params->getQueue(), + $params->getGroup(), + $params->getLimit() ); foreach ($pendingJobList as $job) { - $this->processJob($job, $pool); + $this->processJob($params, $job, $pool); } $pool?->wait(); } - private function processJob(JobEntity $job, ?AsyncPool $pool = null): void + private function processJob(Params $params, JobEntity $job, ?AsyncPool $pool = null): void { - $useProcessPool = $this->params->useProcessPool(); - $noLock = $this->params->noLock(); + $useProcessPool = $params->useProcessPool(); + $noLock = $params->noLock(); $lockTable = $job->getScheduledJobId() && !$noLock; diff --git a/application/Espo/Core/Job/QueueProcessorParams.php b/application/Espo/Core/Job/QueueProcessor/Params.php similarity index 98% rename from application/Espo/Core/Job/QueueProcessorParams.php rename to application/Espo/Core/Job/QueueProcessor/Params.php index 6f38936e5e..8b599bc69d 100644 --- a/application/Espo/Core/Job/QueueProcessorParams.php +++ b/application/Espo/Core/Job/QueueProcessor/Params.php @@ -27,9 +27,9 @@ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word. ************************************************************************/ -namespace Espo\Core\Job; +namespace Espo\Core\Job\QueueProcessor; -class QueueProcessorParams +class Params { private bool $useProcessPool = false; private bool $noLock = false; diff --git a/application/Espo/Core/Job/QueueProcessorFactory.php b/application/Espo/Core/Job/QueueProcessorFactory.php deleted file mode 100644 index 467939439c..0000000000 --- a/application/Espo/Core/Job/QueueProcessorFactory.php +++ /dev/null @@ -1,45 +0,0 @@ -injectableFactory->createWith(QueueProcessor::class, [ - 'params' => $params, - ]); - } -} diff --git a/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php b/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php index 420fd4d547..266bc1b1c2 100644 --- a/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php +++ b/tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php @@ -30,7 +30,7 @@ namespace tests\unit\Espo\Core\Job; use Espo\Core\{ - Job\QueueProcessorParams, + Job\QueueProcessor\Params, }; class QueueProcessorParamsTest extends \PHPUnit\Framework\TestCase @@ -41,7 +41,7 @@ class QueueProcessorParamsTest extends \PHPUnit\Framework\TestCase public function testParams1() { - $params = QueueProcessorParams + $params = \Espo\Core\Job\QueueProcessor\Params ::create() ->withLimit(10); @@ -55,7 +55,7 @@ class QueueProcessorParamsTest extends \PHPUnit\Framework\TestCase public function testParams2() { - $params = QueueProcessorParams + $params = \Espo\Core\Job\QueueProcessor\Params ::create() ->withLimit(10) ->withUseProcessPool(true)