Queue command with direct redis access so it's faster

This commit is contained in:
Juan Pablo Vial
2025-05-16 13:56:32 -04:00
parent f47f86dd2b
commit 8ba54fd3ad
6 changed files with 103 additions and 18 deletions

View File

@ -6,6 +6,7 @@
"guzzlehttp/guzzle": "^7.8", "guzzlehttp/guzzle": "^7.8",
"monolog/monolog": "^3.5", "monolog/monolog": "^3.5",
"php-di/php-di": "^7.0", "php-di/php-di": "^7.0",
"predis/predis": "^3.0",
"symfony/console": "^6.3" "symfony/console": "^6.3"
}, },
"require-dev": { "require-dev": {

View File

@ -4,7 +4,7 @@ use Psr\Container\ContainerInterface;
return [ return [
Psr\Log\LoggerInterface::class => function(ContainerInterface $container) { Psr\Log\LoggerInterface::class => function(ContainerInterface $container) {
$minLogLevel = Monolog\Level::Debug; $minLogLevel = Monolog\Level::Debug;
if ($container->has('DEBUG') and !$container->get('DEBUG')) { if ($container->has('DEBUG') and $container->get('DEBUG') === 'false') {
$minLogLevel = Monolog\Level::Warning; $minLogLevel = Monolog\Level::Warning;
} }
$handlers = []; $handlers = [];

View File

@ -0,0 +1,19 @@
<?php
use Psr\Container\ContainerInterface;
return [
Predis\ClientInterface::class => function(ContainerInterface $container) {
$options = [
'scheme' => 'tcp',
'host' => $container->get('REDIS_HOST'),
'port' => $container->get('REDIS_PORT')
];
if ($container->has('REDIS_USER')) {
$options['username'] = $container->get('REDIS_USER');
}
if ($container->has('REDIS_PASSWORD')) {
$options['password'] = $container->get('REDIS_PASSWORD');
}
return new Predis\Client($options);
},
];

View File

@ -2,7 +2,10 @@
namespace Incoviba\Command; namespace Incoviba\Command;
use DateTimeImmutable; use DateTimeImmutable;
use Psr\Http\Client\ClientInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\Console; use Symfony\Component\Console;
use Incoviba\Service\Job;
use Incoviba\Common\Alias\Command; use Incoviba\Common\Alias\Command;
#[Console\Attribute\AsCommand( #[Console\Attribute\AsCommand(
@ -11,6 +14,12 @@ use Incoviba\Common\Alias\Command;
)] )]
class Queue extends Command class Queue extends Command
{ {
public function __construct(ClientInterface $client, LoggerInterface $logger,
protected Job $jobService, ?string $name = null)
{
parent::__construct($client, $logger, $name);
}
public function execute(Console\Input\InputInterface $input, Console\Output\OutputInterface $output): int public function execute(Console\Input\InputInterface $input, Console\Output\OutputInterface $output): int
{ {
$this->logger->debug("Running {$this->getName()}"); $this->logger->debug("Running {$this->getName()}");
@ -19,29 +28,18 @@ class Queue extends Command
$io->title("[{$now->format('Y-m-d H:i:s e')}] Running Queue..."); $io->title("[{$now->format('Y-m-d H:i:s e')}] Running Queue...");
$jobs = $this->getJobs($output); $jobs = $this->getJobs($output);
if (is_int($jobs)) { if (count($jobs) === 0) {
return $jobs; return Console\Command\Command::SUCCESS;
} }
return $this->runJobs($output, $jobs); return $this->runJobs($output, $jobs);
} }
protected function getJobs(Console\Output\OutputInterface $output): int|array protected function getJobs(Console\Output\OutputInterface $output): array
{ {
$uri = '/api/queue/jobs'; $this->logger->debug("Getting jobs");
$output->writeln("GET {$uri}"); $jobs = $this->jobService->getPending();
$response = $this->client->get($uri); return array_column($jobs, 'id');
$output->writeln("Response Code: {$response->getStatusCode()}");
if ($response->getStatusCode() !== 200) {
return Console\Command\Command::FAILURE;
}
$contents = $response->getBody()->getContents();
if (empty($contents)) {
return [];
}
return json_decode($contents, true)['jobs'];
} }
protected function runJobs(Console\Output\OutputInterface $output, array $jobs): int protected function runJobs(Console\Output\OutputInterface $output, array $jobs): int
{ {

26
cli/src/Service/Job.php Normal file
View File

@ -0,0 +1,26 @@
<?php
namespace Incoviba\Service;
use Exception;
use Predis\Connection\ConnectionException;
use Psr\Log\LoggerInterface;
class Job
{
public function __construct(protected LoggerInterface $logger, protected Redis $redisService)
{
$this->redisKey = 'jobs';
}
protected string $redisKey;
public function getPending(): array
{
try {
$jobs = $this->redisService->get($this->redisKey);
return json_decode($jobs, true);
} catch (ConnectionException|Exception $exception) {
$this->logger->warning($exception);
return [];
}
}
}

41
cli/src/Service/Redis.php Normal file
View File

@ -0,0 +1,41 @@
<?php
namespace Incoviba\Service;
use Exception;
use Predis\ClientInterface;
use Predis\Connection\ConnectionException;
class Redis
{
public function __construct(protected ClientInterface $client) {}
/**
* @param string $name
* @return string|null
* @throws Exception|ConnectionException
*/
public function get(string $name): ?string
{
if (!$this->client->exists($name)) {
throw new Exception($name);
}
return $this->client->get($name);
}
/**
* @param string $name
* @param mixed $value
* @param int $expirationTTL
* @return void
* @throws ConnectionException
*/
public function set(string $name, mixed $value, int $expirationTTL = 60 * 60 * 24): void
{
$resolution = 'EX';
if ($expirationTTL === -1) {
$resolution = null;
$expirationTTL = null;
}
$this->client->set($name, $value, $resolution, $expirationTTL);
}
}