Incoviba/oficial
1
0
Fork 0
Code Issues Pull Requests 2 Releases Wiki Activity

Beanstalkd for jobs

Browse Source
This commit is contained in:
Juan Pablo Vial
2025-07-14 14:29:41 -04:00
parent e6e7470bb2
commit 501151a90e
26 changed files with 693 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
CLI.Dockerfile
View File

@ -3,7 +3,8 @@ FROM php:8.4-cli
ENV TZ "${TZ}"
ENV APP_NAME "${APP_NAME}"
RUN apt-get update && apt-get install -y --no-install-recommends cron rsyslog nano && rm -r /var/lib/apt/lists/*
RUN apt-get update && apt-get install -y --no-install-recommends cron rsyslog nano beanstalkd \
&& rm -r /var/lib/apt/lists/*
RUN pecl install xdebug-3.4.2 \
&& docker-php-ext-enable xdebug \

2
Dockerfile
View File

@ -3,7 +3,7 @@ FROM php:8.4-fpm
ENV TZ=America/Santiago
RUN apt-get update && apt-get install -y --no-install-recommends libzip-dev libicu-dev git \
libpng-dev unzip tzdata libxml2-dev \
libpng-dev unzip tzdata libxml2-dev beanstalkd \
&& rm -r /var/lib/apt/lists/* \
&& docker-php-ext-install pdo pdo_mysql zip intl gd bcmath dom \
&& pecl install xdebug-3.4.2 \

3
app/composer.json
View File

@ -20,7 +20,8 @@
"robmorgan/phinx": "^0.16",
"slim/slim": "^4",
"symfony/string": "^7.2",
"tedivm/jshrink": "^1.7"
"tedivm/jshrink": "^1.7",
"xobotyi/beansclient": "^1.0"
},
"require-dev": {
"fakerphp/faker": "^1",

8
app/resources/database/migrations/20250701192525_create_toku_accounts.php
View File

@ -19,14 +19,18 @@ final class CreateTokuAccounts extends AbstractMigration
*/
public function change(): void
{
$this->execute('SET unique_checks=0; SET foreign_key_checks=0;');
$this->table('toku_accounts')
->addColumn('sociedad_rut', 'integer', ['limit' => 8, 'signed' => false, 'null' => false])
->addColumn('toku_id', 'string', ['length' => 255, 'null' => false])
->addColumn('account_key', 'string', ['length' => 255, 'null' => false])
->addColumn('enabled', 'boolean', ['default' => true])
->addIndex(['toku_id'], ['unique' => true])
->addForeignKey('sociedad_rut', 'inmobiliaria', 'rut', ['delete' => 'CASCADE', 'update' => 'CASCADE'])
->addTimestamps()
#->addForeignKey('sociedad_rut', 'inmobiliaria', 'rut', ['delete' => 'CASCADE', 'update' => 'CASCADE'])
->addIndex(['toku_id'], ['unique' => true])
->create();
$this->execute('SET unique_checks=1; SET foreign_key_checks=1;');
}
}

6
app/resources/routes/api/queue.php
View File

@ -2,9 +2,9 @@
use Incoviba\Controller\API\Queues;
$app->group('/queue', function($app) {
$app->get('/jobs[/]', [Queues::class, 'jobs']);
#$app->get('/jobs[/]', [Queues::class, 'jobs']);
$app->group('/run', function($app) {
$app->get('/{job_id:[0-9]+}[/]', [Queues::class, 'run']);
#$app->get('[/]', Queues::class);
#$app->get('/{job_id:[0-9]+}[/]', [Queues::class, 'run']);
$app->get('[/]', Queues::class);
});
});

10
app/setup/setups/services.php
View File

@ -162,6 +162,16 @@ return [
->register('subscription', $container->get(Incoviba\Service\Venta\MediosPago\Toku\Subscription::class))
->register('invoice', $container->get(Incoviba\Service\Venta\MediosPago\Toku\Invoice::class));
},
xobotyi\beansclient\Interfaces\ConnectionInterface::class => function(ContainerInterface $container) {
return new xobotyi\beansclient\Connection(
$container->get('BEANSTALKD_HOST'),
$container->has('BEANSTALKD_PORT') ? $container->get('BEANSTALKD_PORT') : 11300
);
},
Incoviba\Service\MQTT::class => function(ContainerInterface $container) {
return new Incoviba\Service\MQTT()
->register('beanstalkd', $container->get(Incoviba\Service\MQTT\Beanstalkd::class));
},
Incoviba\Service\Queue::class => function(ContainerInterface $container) {
return new Incoviba\Service\Queue(
$container->get(Psr\Log\LoggerInterface::class),

16
app/src/Controller/API/Queues.php
View File

@ -22,20 +22,4 @@ class Queues extends Ideal\Controller
}
return $this->withJson($response, $output);
}
public function jobs(ServerRequestInterface $request, ResponseInterface $response,
Service\Queue $queueService): ResponseInterface
{
$output = [
'jobs' => array_column($queueService->getPendingJobs(), 'id')
];
return $this->withJson($response, $output);
}
public function run(ServerRequestInterface $request, ResponseInterface $response, Service\Queue $queueService,
int $job_id): ResponseInterface
{
if ($queueService->runJob($job_id, $request)) {
return $response->withStatus(200);
}
return $response->withStatus(422);
}
}

18
app/src/Exception/MQTT.php Normal file
View File

@ -0,0 +1,18 @@
<?php
namespace Incoviba\Exception;
use Throwable;
use Exception;
abstract class MQTT extends Exception
{
public function __construct($message = "", $code = 0, ?Throwable $previous = null)
{
$baseCode = 700;
$code = $baseCode + $code;
if ($message == "") {
$message = "MQTT Exception";
}
parent::__construct($message, $code, $previous);
}
}

18
app/src/Exception/MQTT/MissingClient.php Normal file
View File

@ -0,0 +1,18 @@
<?php
namespace Incoviba\Exception\MQTT;
use Throwable;
use Incoviba\Exception\MQTT;
class MissingClient extends MQTT
{
public function __construct(string $host = '', ?Throwable $previous = null)
{
$message = 'Missing MQTT client';
if ($host !== '') {
$message = "{$message} for host {$host}";
}
$code = 1;
parent::__construct($message, $code, $previous);
}
}

15
app/src/Exception/MQTT/MissingJob.php Normal file
View File

@ -0,0 +1,15 @@
<?php
namespace Incoviba\Exception\MQTT;
use Throwable;
use Incoviba\Exception\MQTT;
class MissingJob extends MQTT
{
public function __construct(?Throwable $previous = null)
{
$message = 'Missing MQTT job';
$code = 10;
parent::__construct($message, $code, $previous);
}
}

15
app/src/Exception/MQTT/RemoveJob.php Normal file
View File

@ -0,0 +1,15 @@
<?php
namespace Incoviba\Exception\MQTT;
use Throwable;
use Incoviba\Exception\MQTT;
class RemoveJob extends MQTT
{
public function __construct(int $jobId, ?Throwable $previous = null)
{
$message = "Could not remove job {$jobId}";
$code = 13;
parent::__construct($message, $code, $previous);
}
}

15
app/src/Exception/MQTT/SetJob.php Normal file
View File

@ -0,0 +1,15 @@
<?php
namespace Incoviba\Exception\MQTT;
use Throwable;
use Incoviba\Exception\MQTT;
class SetJob extends MQTT
{
public function __construct(string $payload, ?Throwable $previous = null)
{
$message = "Could not set job with {$payload}";
$code = 11;
parent::__construct($message, $code, $previous);
}
}

138
app/src/Service/Job.php
View File

@ -5,60 +5,43 @@ use DateInvalidTimeZoneException;
use DateMalformedStringException;
use DateTimeImmutable;
use DateTimeZone;
use InvalidArgumentException;
use OutOfRangeException;
use Psr\Log\LoggerInterface;
use Predis\Connection\ConnectionException;
use Incoviba\Common\Ideal;
use Incoviba\Common\Implement\Exception\EmptyRedis;
use Incoviba\Common\Implement\Exception\EmptyResult;
use Incoviba\Exception\MQTT as MQTTException;
use Incoviba\Exception\ServiceAction\{Create, Delete, Read, Update};
use Incoviba\Repository;
use Incoviba\Model;
use Incoviba\Repository;
class Job extends Ideal\Service
{
public function __construct(LoggerInterface $logger, protected Redis $redisService,
public function __construct(LoggerInterface $logger, protected MQTT $mqttService,
protected Repository\Job $jobRepository)
{
parent::__construct($logger);
}
protected string $redisKey = 'jobs';
public function getPending(null|string|array $orderBy = null): array
public function isPending(): bool
{
try {
$jobs = $this->redisService->get($this->redisKey);
if ($jobs === null) {
return [];
}
$jobs = json_decode($jobs, true);
if ($orderBy !== null) {
uksort($jobs, function($a, $b) use ($orderBy) {
return $a[$orderBy] <=> $b[$orderBy];
});
}
return array_map([$this, 'load'], $jobs);
} catch (ConnectionException | EmptyRedis) {
return [];
return $this->mqttService->exists();
} catch (MQTTException $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
return false;
}
}
/**
* @param int $id
* @return Model\Job
* @throws Read
*/
public function getPendingById(int $id): Model\Job
public function get(): Model\Job
{
$jobs = $this->getJobs();
try {
$idx = $this->findJob($jobs, $id);
} catch (EmptyResult $exception) {
$exception = new OutOfRangeException('Job not found', count($jobs), $exception);
return $this->load(json_decode($this->mqttService->get(), true));
} catch (MQTTException $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
throw new Read(__CLASS__, $exception);
}
return $this->load($jobs[$idx]);
}
/**
@ -71,6 +54,7 @@ class Job extends Ideal\Service
try {
$now = (new DateTimeImmutable('now', new DateTimeZone($_ENV['TZ'] ?? 'America/Santiago')));
} catch (DateMalformedStringException | DateInvalidTimeZoneException $exception) {
$this->logger->warning($exception->getMessage(), ['exception' => $exception]);
$now = new DateTimeImmutable();
}
$data = [
@ -81,17 +65,9 @@ class Job extends Ideal\Service
'updated_at' => null,
'retries' => 0
];
$jobs = [];
try {
$jobs = $this->redisService->get($this->redisKey);
if ($jobs !== null) {
$jobs = json_decode($jobs, true);
}
} catch (EmptyRedis) {}
$jobs []= $data;
try {
$this->redisService->set($this->redisKey, json_encode($jobs), -1);
} catch (ConnectionException $exception) {
$this->mqttService->set(json_encode($data));
} catch (MQTTException $exception) {
throw new Create(__CLASS__, $exception);
}
return $this->load($data);
@ -99,50 +75,35 @@ class Job extends Ideal\Service
/**
* @param Model\Job $job
* @return Model\Job
* @return void
* @throws Update
* @throws Read
*/
public function update(Model\Job $job): Model\Job
public function update(Model\Job $job): void
{
try {
$now = (new DateTimeImmutable('now', new DateTimeZone($_ENV['TZ'] ?? 'America/Santiago')));
} catch (DateMalformedStringException | DateInvalidTimeZoneException) {
$now = new DateTimeImmutable();
}
$jobs = $this->getJobs();
$data = json_decode(json_encode($job), true);
$data['updated_at'] = $now->format('Y-m-d H:i:s');
try {
$idx = $this->findJob($jobs, $job->id);
} catch (EmptyResult $exception) {
throw new Read(__CLASS__, $exception);
}
$jobs[$idx]['updated_at'] = $now->format('Y-m-d H:i:s');
$jobs[$idx]['retries'] = $job->retries;
try {
$this->redisService->set($this->redisKey, json_encode($jobs), -1);
} catch (ConnectionException $exception) {
$this->mqttService->update(json_encode($data));
} catch (MQTTException $exception) {
throw new Update(__CLASS__, $exception);
}
return $this->load($jobs[$idx]);
}
/**
* @param Model\Job $job
* @throws Read
* @throws Delete
*/
public function remove(Model\Job $job): void
{
$jobs = $this->getJobs();
try {
$idx = $this->findJob($jobs, $job->id);
} catch (EmptyResult $exception) {
throw new Read(__CLASS__, $exception);
}
unset($jobs[$idx]);
try {
$this->redisService->set($this->redisKey, json_encode(array_values($jobs)), -1);
} catch (ConnectionException $exception) {
$this->mqttService->remove();
} catch (MQTTException $exception) {
throw new Delete(__CLASS__, $exception);
}
}
@ -150,59 +111,18 @@ class Job extends Ideal\Service
/**
* @param Model\Job $job
* @return bool
* @throws Read | Create
*/
public function execute(Model\Job $job): bool
{
$jobs = $this->getJobs();
try {
$idx = $this->findJob($jobs, $job->id);
} catch (EmptyResult $exception) {
$exception = new OutOfRangeException('Job not found', count($jobs), $exception);
throw new Read(__CLASS__, $exception);
}
unset($jobs[$idx]);
try {
$this->redisService->set($this->redisKey, json_encode(array_values($jobs)), -1);
} catch (ConnectionException $exception) {
throw new Create(__CLASS__, $exception);
}
$this->mqttService->remove();
return true;
} catch (MQTTException $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
return false;
}
}
/**
* @return array
* @throws Read
*/
protected function getJobs(): array
{
try {
$jobs = $this->redisService->get($this->redisKey);
} catch (EmptyRedis $exception) {
throw new Read(__CLASS__, $exception);
}
if ($jobs === null) {
$exception = new InvalidArgumentException("Redis Key {$this->redisKey} not found");
throw new Read(__CLASS__, $exception);
}
return json_decode($jobs, true);
}
/**
* @param array $jobs
* @param int $id
* @return int
* @throws EmptyResult
*/
protected function findJob(array $jobs, int $id): int
{
$idx = array_find_key($jobs, function($job) use ($id) {
return (int) $job['id'] === $id;
});
if ($idx === null) {
throw new EmptyResult("SELECT * FROM jobs WHERE id = ?");
}
return $idx;
}
protected function load(array $data, ?int $id = null): Model\Job
{
$job = new Model\Job();

102
app/src/Service/MQTT.php Normal file
View File

@ -0,0 +1,102 @@
<?php
namespace Incoviba\Service;
use Incoviba\Exception\MQTT as MQTTException;
use Incoviba\Service\MQTT\MQTTInterface;
class MQTT implements MQTTInterface
{
protected array $clients = [];
public function register(string $name, MQTTInterface $client): self
{
$this->clients[$name] = $client;
return $this;
}
public function clientExists(string $name): bool
{
return isset($this->clients[$name]);
}
/**
* @param string|null $name
* @return MQTTInterface
* @throws MQTTException/MissingClient
*/
public function getClient(?string $name = null): MQTTInterface
{
if ($name === null) {
$name = array_keys($this->clients)[0];
}
if (!$this->clientExists($name)) {
throw new MQTTException\MissingClient($name);
}
return $this->clients[$name];
}
/**
* @param string|null $host
* @return bool
* @throws MQTTException/MissingClient
* @throws MQTTException/MissingJob
*/
public function exists(?string $host = null): bool
{
$client = $this->getClient($host);
return $client->exists();
}
/**
* @param string|null $host
* @return string
* @throws MQTTException/MissingClient
* @throws MQTTException/MissingJob
*/
public function get(?string $host = null): string
{
$client = $this->getClient($host);
return $client->get();
}
/**
* @param string $value
* @param int $delay
* @param string|null $host
* @return $this
* @throws MQTTException/MissingClient
* @throws MQTTException/SetJob
*/
public function set(string $value, int $delay = 0, ?string $host = null): self
{
$client = $this->getClient($host);
$client->set($value, $delay);
return $this;
}
/**
* @param int|null $jobId
* @param string|null $host
* @return $this
* @throws MQTTException/MissingJob
* @throws MQTTException/RemoveJob
*/
public function remove(?int $jobId = null, ?string $host = null): self
{
$this->getClient($host)->remove($jobId);
return $this;
}
/**
* @param string $newPayload
* @param int|null $jobId
* @param string|null $host
* @return $this
* @throws MQTTException/MissingJob
* @throws MQTTException/RemoveJob
* @throws MQTTException/SetJob
*/
public function update(string $newPayload, ?int $jobId = null, ?string $host = null): self
{
$this->getClient($host)->update($newPayload, $jobId);
return $this;
}
}

111
app/src/Service/MQTT/Beanstalkd.php Normal file
View File

@ -0,0 +1,111 @@
<?php
namespace Incoviba\Service\MQTT;
use Exception;
use Incoviba\Exception\MQTT\MissingJob;
use Incoviba\Exception\MQTT\RemoveJob;
use Incoviba\Exception\MQTT\SetJob;
use Psr\Log\LoggerInterface;
use xobotyi\beansclient;
use xobotyi\beansclient\Exception\ClientException;
use xobotyi\beansclient\Exception\CommandException;
use xobotyi\beansclient\Exception\JobException;
class Beanstalkd implements MQTTInterface
{
/**
* @throws JobException
* @throws ClientException
* @throws CommandException
*/
public function __construct(protected LoggerInterface $logger, protected beansclient\BeansClient $client,
protected string $tube = 'default', protected int $ttr = beansclient\BeansClient::DEFAULT_TTR,
protected int $priority = 1)
{
$this->client->watchTube($this->tube);
}
public function exists(): bool
{
try {
$stats = $this->client->statsTube($this->tube);
return $stats['current-jobs-ready'] > 0;
} catch (Exception $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
return false;
}
}
protected ?beansclient\Job $currentJob = null;
public function get(): string
{
if (!$this->exists()) {
throw new MissingJob();
}
try {
$job = $this->client->watchTube($this->tube)->reserve();
$this->currentJob = $job;
return $job->payload;
} catch (Exception $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
throw new MissingJob($exception);
}
}
/**
* @param string $value
* @param int $delay
* @return $this
* @throws SetJob
*/
public function set(string $value, int $delay = 0): self
{
try {
$this->client->useTube($this->tube)->put($value, $this->priority, $delay, $this->ttr ?? 0);
} catch (Exception $exception) {
$this->logger->error($exception->getMessage(), ['payload' => $value, 'delay' => $delay, 'exception' => $exception]);
throw new SetJob($value, $exception);
}
return $this;
}
/**
* @param string $newPayload
* @param int|null $jobId
* @return self
* @throws RemoveJob
* @throws SetJob
*/
public function update(string $newPayload, ?int $jobId = null): self
{
if ($jobId === null) {
$jobId = $this->currentJob->id;
}
return $this->remove($jobId)
->set($newPayload);
}
/**
* @param int|null $jobId
* @return $this
* @throws RemoveJob
*/
public function remove(?int $jobId = null): self
{
if ($jobId === null) {
$jobId = $this->currentJob->id;
}
try {
if (!$this->client->useTube($this->tube)->delete($jobId)) {
throw new JobException("Failed to delete job {$jobId}");
}
if ($this->currentJob !== null && $this->currentJob->id === $jobId) {
$this->currentJob = null;
}
} catch (Exception $exception) {
$this->logger->error($exception->getMessage(), ['jobId' => $jobId, 'exception' => $exception]);
throw new RemoveJob($jobId, $exception);
}
return $this;
}
}

25
app/src/Service/MQTT/MQTTInterface.php Normal file
View File

@ -0,0 +1,25 @@
<?php
namespace Incoviba\Service\MQTT;
use Incoviba\Exception\MQTT\MissingJob;
interface MQTTInterface {
/**
* @return bool
*/
public function exists(): bool;
/**
* @return string
* @throws MissingJob
*/
public function get(): string;
/**
* @param string $value
* @param int $delay
* @return self
*/
public function set(string $value, int $delay = 0): self;
public function remove(?int $jobId = null): self;
}

56
app/src/Service/Queue.php
View File

@ -7,6 +7,7 @@ use Psr\Log\LoggerInterface;
use Incoviba\Common\Ideal;
use Incoviba\Exception\ServiceAction\{Create, Delete, Read, Update};
use Incoviba\Service;
use Incoviba\Model;
class Queue extends Ideal\Service
{
@ -29,7 +30,7 @@ class Queue extends Ideal\Service
try {
$this->jobService->add($configuration);
return true;
} catch (Read $exception) {
} catch (Create $exception) {
$final = new Exception("Could not enqueue job", 0, $exception);
$this->logger->warning($final);
return false;
@ -40,22 +41,8 @@ class Queue extends Ideal\Service
return $this->enqueue($configuration);
}
/**
* @return array
*/
public function getPendingJobs(): array
public function runJob(Model\Job $job, ?RequestInterface $request = null): bool
{
return $this->jobService->getPending();
}
public function runJob(int $job_id, ?RequestInterface $request = null): bool
{
try {
$job = $this->jobService->getPendingById($job_id);
} catch (Read $exception) {
$this->logger->debug($exception);
return false;
}
$type = 'default';
if (isset($job->configuration['type'])) {
$type = strtolower($job->configuration['type']);
@ -71,50 +58,49 @@ class Queue extends Ideal\Service
try {
if (!$worker->execute($job)) {
$this->logger->debug("Could not execute job {$job_id}");
$this->logger->debug("Could not execute job {$job->id}");
return false;
}
if (!$this->jobService->execute($job)) {
$this->logger->debug("Could not remove job {$job_id}");
$this->logger->debug("Could not remove job {$job->id}");
return false;
}
} catch (Exception $exception) {
$final = new Exception("Could not run job", 0, $exception);
$this->logger->warning($final);
$this->logger->warning("Could not run job {$job->id}", ['exception' => $exception]);
return false;
}
return true;
}
public function run(?RequestInterface $request = null): bool
{
$jobs = $this->jobService->getPending();
if (count($jobs) === 0) {
$this->logger->debug("No pending jobs");
if (!$this->jobService->isPending()) {
return true;
}
$errors = [];
foreach ($jobs as $job) {
try {
$job = $this->jobService->get();
} catch (Read $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
return false;
}
if ($job->retries >= $this->maxRetries) {
try {
$this->jobService->remove($job);
} catch (Read | Delete $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
} catch (Delete $exception) {
$this->logger->error($exception->getMessage(), ['job' => $job, 'exception' => $exception]);
}
continue;
return true;
}
try {
$this->runJob($job->id, $request);
$this->runJob($job, $request);
} catch (Exception) {
$job->retries ++;
try {
$this->jobService->update($job);
} catch (Read | Update $exception) {
$this->logger->error($exception->getMessage(), ['exception' => $exception]);
} catch (Update $exception) {
$this->logger->error($exception->getMessage(), ['job' => $job, 'exception' => $exception]);
}
$errors []= $job->id;
return false;
}
}
return count($errors) === 0;
return true;
}
}

101
app/tests/unit/src/Service/MQTT/BeanstalkdTest.php Normal file
View File

@ -0,0 +1,101 @@
<?php
namespace Incoviba\Test\Service\MQTT;
use Psr\Log\LoggerInterface;
use PHPUnit\Framework\TestCase;
use xobotyi\beansclient\BeansClient;
use xobotyi\beansclient\Connection;
use xobotyi\beansclient\Exception\JobException;
use xobotyi\beansclient\Job;
use Incoviba\Exception\MQTT\MissingJob;
use Incoviba\Service\MQTT\Beanstalkd;
class BeanstalkdTest extends TestCase
{
protected LoggerInterface $logger;
protected BeansClient $client;
protected function setUp(): void
{
$this->logger = $this->getMockBuilder(LoggerInterface::class)->getMock();
$this->client = $this->getMockBuilder(BeansClient::class)->disableOriginalConstructor()->getMock();
}
public function testExists(): void
{
$stats = ['current-jobs-ready' => 1];
$this->client->method('watchTube')->willReturn($this->client);
$this->client->method('statsTube')->willReturn($stats);
$service = new Beanstalkd($this->logger, $this->client);
$this->assertTrue($service->exists());
}
public function testNotExists(): void
{
$stats = ['current-jobs-ready' => 0];
$this->client->method('watchTube')->willReturn($this->client);
$this->client->method('statsTube')->willReturn($stats);
$service = new Beanstalkd($this->logger, $this->client);
$this->assertFalse($service->exists());
}
public function testGet(): void
{
$jobData = [
'id' => 1,
'configuration' => [
'type' => 'service',
],
'created_at' => '2020-01-01 00:00:00',
];
$connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
$connection->method('isActive')->willReturn(true);
$this->client->method('getConnection')->willReturn($connection);
$this->client->method('watchTube')->willReturn($this->client);
$this->client->method('statsTube')->willReturn(['current-jobs-ready' => 1]);
$job = new Job($this->client, 1, 'ready', json_encode($jobData));
$this->client->method('reserve')->willReturn($job);
$service = new Beanstalkd($this->logger, $this->client);
$this->assertEquals(json_encode($jobData), $service->get());
}
public function testGetException(): void
{
$this->client->method('watchTube')->willReturn($this->client);
$this->client->method('statsTube')->willReturn(['current-jobs-ready' => 0]);
$service = new Beanstalkd($this->logger, $this->client);
$this->expectException(MissingJob::class);
$service->get();
$this->client->method('statsTube')->willReturn(['current-jobs-ready' => 1]);
$exception = new JobException();
$this->client->method('reserve')->willThrowException($exception);
$this->expectException(MissingJob::class);
$service->get();
}
public function testSet(): void
{
$this->client->method('useTube')->willReturn($this->client);
$this->client->method('put');
$service = new Beanstalkd($this->logger, $this->client);
$service->set('test');
$this->assertTrue(true);
}
public function testSetException(): void
{
$this->client->method('useTube')->willReturn($this->client);
$exception = new JobException();
$this->client->method('put')->willThrowException($exception);
$service = new Beanstalkd($this->logger, $this->client);
$service->set('test');
$this->assertTrue(true);
}
public function testRemove(): void
{
$this->client->method('useTube')->willReturn($this->client);
$this->client->method('delete')->willReturn(true);
$service = new Beanstalkd($this->logger, $this->client);
$service->remove(1);
$this->assertTrue(true);
$this->client->method('delete')->willReturn(false);
$service->remove(1);
$this->assertTrue(true);
}
}

69
app/tests/unit/src/Service/MQTTTest.php Normal file
View File

@ -0,0 +1,69 @@
<?php
namespace Incoviba\Test\Service;
use Psr\Log\LoggerInterface;
use PHPUnit\Framework\TestCase;
use xobotyi\beansclient\BeansClient;
use Incoviba\Exception\MQTT\MissingClient;
use Incoviba\Service\MQTT;
use Incoviba\Service\MQTT\Beanstalkd;
class MQTTTest extends TestCase
{
public function testRegisterAndClientExistsAndGet(): void
{
$mqtt = new MQTT();
$beanstalkd = $this->getMockBuilder(Beanstalkd::class)->disableOriginalConstructor()->getMock();
$mqtt->register('beanstalkd', $beanstalkd);
$this->assertTrue($mqtt->clientExists('beanstalkd'));
}
public function testGetClient(): void
{
$mqtt = new MQTT();
$logger = $this->getMockBuilder(LoggerInterface::class)->disableOriginalConstructor()->getMock();
$client = $this->getMockBuilder(BeansClient::class)->disableOriginalConstructor()->getMock();
$beanstalkd = new Beanstalkd($logger, $client);
$mqtt->register('beanstalkd', $beanstalkd);
$this->assertEquals($beanstalkd, $mqtt->getClient('beanstalkd'));
}
public function testGetClientException(): void
{
$mqtt = new MQTT();
$this->expectException(MissingClient::class);
$mqtt->getClient('test');
}
public function testExists(): void
{
$mqtt = new MQTT();
$beanstalkd = $this->getMockBuilder(Beanstalkd::class)->disableOriginalConstructor()->getMock();
$beanstalkd->method('exists')->willReturn(true);
$mqtt->register('beanstalkd', $beanstalkd);
$this->assertTrue($mqtt->exists('beanstalkd'));
}
public function testGet(): void
{
$mqtt = new MQTT();
$beanstalkd = $this->getMockBuilder(Beanstalkd::class)->disableOriginalConstructor()->getMock();
$beanstalkd->method('get')->willReturn('test');
$mqtt->register('beanstalkd', $beanstalkd);
$this->assertEquals('test', $mqtt->get('beanstalkd'));
}
public function testSet(): void
{
$mqtt = new MQTT();
$beanstalkd = $this->getMockBuilder(Beanstalkd::class)->disableOriginalConstructor()->getMock();
$beanstalkd->method('set');
$mqtt->register('beanstalkd', $beanstalkd);
$mqtt->set('test', 0, 'beanstalkd');
$this->assertTrue(true);
}
public function testRemove(): void
{
$mqtt = new MQTT();
$beanstalkd = $this->getMockBuilder(Beanstalkd::class)->disableOriginalConstructor()->getMock();
$beanstalkd->method('remove');
$mqtt->register('beanstalkd', $beanstalkd);
$mqtt->remove(0, 'beanstalkd');
$this->assertTrue(true);
}
}

66
app/tests/unit/src/Service/QueueTest.php Normal file
View File

@ -0,0 +1,66 @@
<?php
namespace Incoviba\Test\Service;
use Psr\Log\LoggerInterface;
use PHPUnit\Framework\TestCase;
use Incoviba\Service\Job;
use Incoviba\Service\Queue;
use Incoviba\Service\Worker;
use Incoviba\Model;
class QueueTest extends TestCase
{
protected LoggerInterface $logger;
protected Job $jobService;
protected Worker $defaultWorker;
protected function setUp(): void
{
$this->logger = $this->getMockBuilder(LoggerInterface::class)
->disableOriginalConstructor()
->getMock();
$this->jobService = $this->getMockBuilder(Job::class)
->disableOriginalConstructor()
->getMock();
$this->defaultWorker = $this->getMockBuilder(Worker::class)
->disableOriginalConstructor()
->getMock();
}
public function testRegister(): void
{
$queue = new Queue($this->logger, $this->jobService, $this->defaultWorker);
$worker = $this->getMockBuilder(Worker::class)
->disableOriginalConstructor()
->getMock();
$queue->register('test', $worker);
$this->assertTrue(true);
}
public function testEnqueue(): void
{
$queue = new Queue($this->logger, $this->jobService, $this->defaultWorker);
$jobData = ['test' => 'test'];
$result = $queue->enqueue($jobData);
$this->assertTrue($result);
$result = $queue->push($jobData);
$this->assertTrue($result);
}
public function testRun(): void
{
$queue = new Queue($this->logger, $this->jobService, $this->defaultWorker);
$result = $queue->run();
$this->assertTrue($result);
$jobData = [
'type' => 'test',
];
$job = new Model\Job();
$job->id = 1;
$job->configuration = $jobData;
$this->jobService->method('isPending')->willReturn(true);
$this->jobService->method('get')->willReturn($job);
$result = $queue->run();
$this->assertTrue($result);
}
}

7
cli/entrypoint
View File

@ -6,8 +6,13 @@ then
then
CMD=$1
shift
if [[ $# -gt 0 ]]
then
$CMD -c "$@"
exit
exit 0
fi
$CMD
exit 0
fi
fi

143
cli/src/Command/Job/Run.php
View File

@ -9,7 +9,7 @@ use Incoviba\Service;
use Psr\Log\LoggerInterface;
use Symfony\Component\Console;
#[Console\Attribute\AsCommand(name: 'jobs:run', description: 'Run jobs')]
#[Console\Attribute\AsCommand(name: 'jobs:run', description: 'Run job')]
class Run extends Console\Command\Command
{
public function __construct(protected Service\FastCGI $fastcgi, protected LoggerInterface $logger,
@ -18,13 +18,6 @@ class Run extends Console\Command\Command
parent::__construct($name);
}
protected function configure(): void
{
$this->addArgument('job_ids',
Console\Input\InputArgument::IS_ARRAY | Console\Input\InputArgument::REQUIRED, 'Job IDs');
}
protected array $output = [];
public function execute(Console\Input\InputInterface $input, Console\Output\OutputInterface $output): int
{
try {
@ -33,44 +26,13 @@ class Run extends Console\Command\Command
$now = new DateTimeImmutable();
}
$jobIds = $input->getArgument('job_ids');
$jobCount = count($jobIds);
$this->pushOutput('top', ['message' => "[{$now->format('Y-m-d H:i:s e')}] Running {$jobCount} jobs..."]);
$this->pushOutput('bottom', ['table' => [
['Job IDs'],
array_map(function($row) {return [$row];},$jobIds)
]]);
$this->pushOutput('top', ['progress' => $jobCount]);
$result = $this->runJobs($jobIds);
$this->pushOutput('top', ['progress' => 'finish']);
$this->writeOutput($input, $output);
return $result;
$output->writeln("[{$now->format('Y-m-d H:i:s e')}] Running Ready Job...");
$this->runJob();
return $this->getResponses();
}
protected function runJobs(array $jobIds): int
protected function runJob(): bool
{
$pendingJobs = [];
foreach ($jobIds as $jobId) {
if (!$this->runJob($jobId)) {
$pendingJobs []= $jobId;
}
}
$result = $this->getResponses();
if (count($pendingJobs) > 0) {
if ($this->runJobs($pendingJobs) === self::FAILURE) {
$result = self::FAILURE;
}
}
return $result;
}
protected function runJob(int $jobId): bool
{
$uri = "/api/queue/run/{$jobId}";
$this->pushOutput('bottom', ['message' => "GET {$uri}"]);
$uri = "/api/queue/run";
try {
$this->fastcgi->get($uri);
@ -85,7 +47,6 @@ class Run extends Console\Command\Command
$result = self::SUCCESS;
$responses = $this->fastcgi->awaitResponses();
foreach ($responses as $response) {
$this->pushOutput('top', ['progress' => 'advance']);
if ($response->getError() !== '') {
$this->logger->error("Error running job", [
'error' => $response->getError(),
@ -93,100 +54,8 @@ class Run extends Console\Command\Command
'headers' => $response->getHeaders(),
]);
$result = self::FAILURE;
continue;
}
$this->pushOutput('bottom', ['message' => $response->getBody()]);
}
return $result;
}
protected function pushOutput(string $section, array $configuration): void
{
if (!isset($this->output[$section])) {
$this->output[$section] = [];
}
foreach ($configuration as $key => $value) {
if (!isset($this->output[$section][$key])) {
$this->output[$section][$key] = [];
}
$this->output[$section][$key] []= $value;
}
if (isset($this->output[$section]['progress'])) {
usort($this->output[$section]['progress'], function($a, $b) {
if ($a === $b) {
return 0;
}
if (is_int($a)) {
return -1;
}
if (is_int($b)) {
return 1;
}
if ($a === 'finish') {
return 1;
}
if ($b === 'finish') {
return -1;
}
return 0;
});
}
}
protected function writeOutput(Console\Input\InputInterface $input, Console\Output\OutputInterface $output): void
{
$sectionNames = array_keys($this->output);
$ios = [];
foreach ($sectionNames as $sectionName) {
$section = $output->section();
$ios[$sectionName] = new Console\Style\SymfonyStyle($input, $section);
}
foreach ($this->output as $sectionName => $configurations) {
$io = $ios[$sectionName];
$this->writeSection($io, $configurations);
}
}
protected function writeSection(Console\Style\SymfonyStyle $io, array $configurations): void
{
if (array_key_exists('table', $configurations)) {
$this->writeTables($io, $configurations['table']);
}
if (array_key_exists('progress', $configurations)) {
$this->writeProgress($io, $configurations['progress']);
}
if (array_key_exists('message', $configurations)) {
$this->writeMessages($io, $configurations['message']);
}
}
protected function writeTables(Console\Style\SymfonyStyle $io, array $tableConfigurations): void
{
foreach ($tableConfigurations as $tableData) {
$io->table(...$tableData);
}
}
protected function writeMessages(Console\Style\SymfonyStyle $io, array $messages): void
{
foreach ($messages as $message) {
$io->writeln($message);
}
}
protected function writeProgress(Console\Style\SymfonyStyle $io, array $progresses): void
{
$progressBar = null;
foreach ($progresses as $progress) {
if ($progress === 'advance' and $progressBar !== null) {
$progressBar->advance();
continue;
}
if ($progress === 'finish' and $progressBar !== null) {
$progressBar->finish();
continue;
}
if (in_array($progress, ['finish', 'advance'])) {
continue;
}
$progressBar = $io->createProgressBar($progress);
}
$io->newLine();
}
}

74
cli/src/Command/Queue.php
View File

@ -37,83 +37,15 @@ class Queue extends Command
$now = new DateTimeImmutable('now', $this->timezone);
$io->title("[{$now->format('Y-m-d H:i:s e')}] Running Queue...");
$jobs = $this->getJobs();
$jobCount = count($jobs);
if ($jobCount === 0) {
return Console\Command\Command::SUCCESS;
}
$io->writeln("Found {$jobCount} jobs to run");
$result = $this->runJob($jobs[0]);
/*$result = $this->runJobs($io, $jobs);
foreach ($this->outputs as $output) {
$this->sections['bottom']->writeln($output);
}*/
return $result;
return $this->runJob();
}
protected array $sections;
protected function getJobs(): array
{
$this->logger->debug("Getting jobs");
$jobs = $this->jobService->getPending();
$jobCount = count($jobs);
if ($jobCount === 0) {
$this->logger->debug("No jobs to run");
return [];
}
$this->logger->debug("Found {$jobCount} jobs");
return array_column($jobs, 'id');
}
protected function runJobs(Console\Style\SymfonyStyle $io, array $jobs): int
{
$chunks = array_chunk($jobs, $this->batchSize);
$chunkCount = count($chunks);
$result = self::SUCCESS;
$progress1 = $io->createProgressBar($chunkCount);
$progress1->start();
foreach ($chunks as $chunk) {
if ($this->runJobBatch($chunk) === self::FAILURE) {
$result = self::FAILURE;
}
$progress1->advance();
}
$progress1->finish();
return $result;
}
protected array $outputs = [];
protected function runJobBatch(array $jobIds): int
protected function runJob(): int
{
$baseCommand = "{$this->baseCommand} jobs:run";
$jobsLine = implode(' ', $jobIds);
$command = "{$baseCommand} {$jobsLine}";
try {
exec($command, $output, $resultCode);
$this->outputs []= $output;
} catch (Throwable $exception) {
$this->logger->error("Failed to run command", [
'command' => $command,
'exception' => $exception
]);
return self::FAILURE;
}
if ($resultCode !== 0) {
$this->logger->error("Failed to run command", [
'command' => $command,
'result_code' => $resultCode
]);
return self::FAILURE;
}
return self::SUCCESS;
}
protected function runJob(int $jobId): int
{
$baseCommand = "{$this->baseCommand} jobs:run";
$command = "{$baseCommand} {$jobId}";
$command = "{$baseCommand}";
try {
exec($command, $output, $resultCode);
$this->outputs []= $output;

2
cli/src/Service/Job.php
View File

@ -11,7 +11,7 @@ use Predis\Connection\ConnectionException;
class Job
{
public function __construct(protected LoggerInterface $logger, protected Redis $redisService)
public function __construct(protected LoggerInterface $logger)
{
$this->redisKey = 'jobs';
}

21
mqtt.compose.yml Normal file
View File

@ -0,0 +1,21 @@
services:
mqtt:
profiles:
- mqtt
container_name: incoviba_mqtt
image: maateen/docker-beanstalkd
restart: unless-stopped
volumes:
- incoviba_mqtt:/var/lib/beanstalkd
mqtt-admin:
profiles:
- mqtt
container_name: incoviba_mqtt_admin
image: mitulislam/beanstalkd-aurora:latest
restart: unless-stopped
ports:
- "8093:3000"
volumes:
incoviba_mqtt: {}

15
testing.compose.yml
View File

@ -18,6 +18,8 @@ services:
condition: service_healthy
test-redis:
condition: service_healthy
test-mqtt:
condition: service_started
test-db:
profiles:
@ -48,6 +50,19 @@ services:
networks:
- testing
test-mqtt:
profiles:
- testing
image: maateen/docker-beanstalkd
container_name: incoviba_test_mqtt
healthcheck:
test: [ "CMD", "nc", "-z", "localhost", "11300" ]
interval: 5s
timeout: 5s
retries: 5
networks:
- testing
volumes:
test-db: {}