Merge

app/src/Service/Queue.php

@@ -59,6 +59,8 @@ class Queue extends Ideal\Service
         try {
             if (!$worker->execute($job)) {
                 $this->logger->debug("Could not execute job {$job->id}");
+                $job->retries++;
+                $this->jobService->update($job);
                 return false;
             }
             if (!$this->jobService->execute($job)) {
@@ -67,6 +69,12 @@ class Queue extends Ideal\Service
             }
         } catch (Exception $exception) {
             $this->logger->warning("Could not run job {$job->id}", ['exception' => $exception]);
+            $job->retries++;
+            try {
+                $this->jobService->update($job);
+            } catch (Update $exception) {
+                $this->logger->error($exception->getMessage(), ['job' => $job, 'exception' => $exception]);
+            }
             return false;
         }
         return true;

cli/src/Command/Queue.php

@@ -42,7 +42,14 @@ class Queue extends Command
         }
 
         $io->title("[{$now->format('Y-m-d H:i:s e')}] Running Queue...");
-        return $this->runJob();
+        $results = [];
+        for ($i = 0; $i < $this->batchSize; $i++) {
+            if ($this->jobService->getPending() === 0) {
+                break;
+            }
+            $results []= $this->runJob();
+        }
+        return count(array_filter($results, fn ($result) => $result === self::FAILURE)) === 0 ? self::SUCCESS : self::FAILURE;
     }
 
     protected array $sections;

cli/src/Command/Queue/Pending.php

@@ -1,10 +1,10 @@
 <?php
-namespace Incoviba\Command\Job;
+namespace Incoviba\Command\Queue;
 
-use Symfony\Component\Console;
 use Incoviba\Service;
+use Symfony\Component\Console;
 
-#[Console\Attribute\AsCommand(name: 'jobs:pending', description: 'List pending jobs')]
+#[Console\Attribute\AsCommand(name: 'queue:pending', description: 'List pending jobs in queue')]
 class Pending extends Console\Command\Command
 {
     public function __construct(protected Service\Job $jobService, ?string $name = null)

cli/src/Command/Queue/Push.php

@@ -2,20 +2,22 @@
 namespace Incoviba\Command\Queue;
 
 use Throwable;
+use Psr\Log\LoggerInterface;
 use Symfony\Component\Console;
 use Incoviba\Service;
 
 #[Console\Attribute\AsCommand(name: 'queue:push', description: 'Push a job to the queue')]
 class Push extends Console\Command\Command
 {
-    public function __construct(protected Service\Job $jobService, ?string $name = null)
+    public function __construct(protected LoggerInterface $logger, protected Service\Job $jobService, ?string $name = null)
     {
         parent::__construct($name);
     }
 
     protected function configure(): void
     {
-        $this->addOption('configurations', 'c', Console\Input\InputOption::VALUE_REQUIRED | Console\Input\InputOption::VALUE_IS_ARRAY, 'Job configuration, must be in valid JSON format');
+        $this->addOption('configurations', 'c', Console\Input\InputOption::VALUE_REQUIRED | Console\Input\InputOption::VALUE_IS_ARRAY, 'Job configuration options array, each job configuration must be in valid JSON format');
+        $this->addOption('files', 'f', Console\Input\InputOption::VALUE_REQUIRED | Console\Input\InputOption::VALUE_IS_ARRAY, 'Paths to jobs configurations files with JSON array content');
     }
 
     protected function execute(Console\Input\InputInterface $input, Console\Output\OutputInterface $output): int
@@ -23,8 +25,8 @@ class Push extends Console\Command\Command
         $io = new Console\Style\SymfonyStyle($input, $output);
         $io->title("Pushing job");
 
-        $configurations = $input->getOption('configurations');
+        $configurations = $this->getConfigurations($input);
-        if ($configurations === null) {
+        if (count($configurations) === 0) {
             $io->error('Missing configurations');
             return self::FAILURE;
         }
@@ -46,4 +48,74 @@ class Push extends Console\Command\Command
         }
         return $result;
     }
+
+    protected function getConfigurations(Console\Input\InputInterface $input): array
+    {
+        return [
+            ...$this->getFilesConfigurations($input),
+            ...$this->getOptionConfigurations($input),
+        ];
+    }
+    protected function getFilesConfigurations(Console\Input\InputInterface $input): array
+    {
+        $configurations = [];
+        $files = $input->getOption('files');
+        if ($files === null) {
+            return $configurations;
+        }
+        foreach ($files as $filePath) {
+            if (!file_exists($filePath)) {
+                continue;
+            }
+            $configurations = array_merge($configurations, $this->getFileConfigurations($filePath));
+        }
+        return $configurations;
+    }
+    protected function getFileConfigurations(string $filePath): array
+    {
+        $configurations = [];
+        if (!file_exists($filePath)) {
+            return $configurations;
+        }
+        $json = file_get_contents($filePath);
+        if (!json_validate($json)) {
+            return $configurations;
+        }
+        $tmp = json_decode($json, true);
+        foreach ($tmp as $config) {
+            try {
+                $configurations []= $this->processConfiguration(json_encode($config));
+            } catch (Throwable $exception) {
+                $this->logger->warning($exception->getMessage(), ['exception' => $exception, 'config' => $config]);
+            }
+        }
+        return $configurations;
+    }
+    protected function getOptionConfigurations(Console\Input\InputInterface $input): array
+    {
+        $configurations = [];
+        $configOptions = $input->getOption('configurations');
+        if ($configOptions === null) {
+            return $configurations;
+        }
+        foreach ($configOptions as $config) {
+            try {
+                $configurations []= $this->processConfiguration($config);
+            } catch (Throwable $exception) {
+                $this->logger->warning($exception->getMessage(), ['exception' => $exception, 'config' => $config]);
+            }
+        }
+        return $configurations;
+    }
+    protected function processConfiguration(string $configuration): string
+    {
+        $json = json_decode($configuration, true);
+        if (!array_key_exists('type', $json) and !array_key_exists('configuration', $json)) {
+            throw new Console\Exception\InvalidArgumentException('Missing type or configuration key in JSON');
+        }
+        if (array_key_exists('type', $json)) {
+            return json_encode($json);
+        }
+        return json_encode($json['configuration']);
+    }
 }