121 lines
3.0 KiB
PHP
121 lines
3.0 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace KupShop\SynchronizationBundle\Util;
|
|
|
|
use Symfony\Component\Console\Output\NullOutput;
|
|
use Symfony\Component\Console\Output\OutputInterface;
|
|
use Symfony\Contracts\Service\Attribute\Required;
|
|
|
|
class SynchronizationQueueUtil
|
|
{
|
|
protected SynchronizationJobProvider $jobProvider;
|
|
|
|
public const CACHE_KEY = 'synchronizationQueue';
|
|
public const CACHE_KEY_JOB_STATUSES = 'jobStatuses';
|
|
|
|
public const STATUS_RUNNING = 'running';
|
|
public const STATUS_SUCCESS = 'success';
|
|
public const STATUS_FAILED = 'failed';
|
|
public const CACHE_TTL = 86400;
|
|
|
|
public function addToQueue(string $jobId): void
|
|
{
|
|
$queue = $this->getQueue();
|
|
$queue[$jobId] = $jobId;
|
|
$this->setQueue($queue);
|
|
}
|
|
|
|
public function processOne(?OutputInterface $output = null): bool
|
|
{
|
|
$jobId = $this->getJob();
|
|
if (!$jobId) {
|
|
return false;
|
|
}
|
|
|
|
$job = $this->jobProvider->getJob($jobId);
|
|
if (!$job) {
|
|
return false;
|
|
}
|
|
|
|
try {
|
|
$this->setJobStatus($jobId, self::STATUS_RUNNING);
|
|
($job->fn)($output ?: new NullOutput());
|
|
} catch (\Exception $e) {
|
|
$this->setJobStatus($jobId, self::STATUS_FAILED, $e->getMessage());
|
|
throw $e;
|
|
}
|
|
$this->setJobStatus($jobId, self::STATUS_SUCCESS);
|
|
|
|
return true;
|
|
}
|
|
|
|
public function processAll(): void
|
|
{
|
|
while (!empty($this->getQueue())) {
|
|
$this->processOne();
|
|
}
|
|
}
|
|
|
|
public function isQueued(string $jobId): bool
|
|
{
|
|
return in_array($jobId, $this->getQueue());
|
|
}
|
|
|
|
protected function getQueue(): array
|
|
{
|
|
return getCache(self::CACHE_KEY) ?: [];
|
|
}
|
|
|
|
protected function setQueue(array $queue): void
|
|
{
|
|
setCache(self::CACHE_KEY, $queue);
|
|
}
|
|
|
|
protected function getJob(): ?string
|
|
{
|
|
$queue = $this->getQueue();
|
|
$job = array_shift($queue);
|
|
$this->setQueue($queue);
|
|
|
|
return $job;
|
|
}
|
|
|
|
public function setJobStatus(string $jobId, string $status, string $msg = ''): void
|
|
{
|
|
$statuses = $this->getJobStatuses();
|
|
$currentStatus = $statuses[$jobId] ?? [];
|
|
|
|
$currentStatus['status'] = $status;
|
|
$currentStatus['msg'] = $msg;
|
|
if ($status === self::STATUS_RUNNING) {
|
|
$currentStatus['time'] = time();
|
|
}
|
|
$statuses[$jobId] = $currentStatus;
|
|
|
|
$this->setJobStatuses($statuses);
|
|
}
|
|
|
|
protected function setJobStatuses(array $statuses): void
|
|
{
|
|
setCache(self::CACHE_KEY_JOB_STATUSES, $statuses, self::CACHE_TTL);
|
|
}
|
|
|
|
public function getJobStatuses(): array
|
|
{
|
|
return getCache(self::CACHE_KEY_JOB_STATUSES) ?: [];
|
|
}
|
|
|
|
public function getJobStatus(string $jobId): ?array
|
|
{
|
|
return $this->getJobStatuses()[$jobId] ?? null;
|
|
}
|
|
|
|
#[Required]
|
|
public function setJobProvider(SynchronizationJobProvider $jobProvider): void
|
|
{
|
|
$this->jobProvider = $jobProvider;
|
|
}
|
|
}
|