* Add possibility to save exceptions in file

Author: AUDUL

src/Command/JobShowCommand.php

@@ -6,6 +6,7 @@
 
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
+use CodeRhapsodie\DataflowBundle\Gateway\JobGateway;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
@@ -26,7 +27,7 @@ class JobShowCommand extends Command
         Job::STATUS_COMPLETED => 'Completed',
     ];
 
-    public function __construct(private JobRepository $jobRepository, private ConnectionFactory $connectionFactory)
+    public function __construct(private JobGateway $jobGateway, private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
     }
@@ -58,9 +59,9 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         }
 
         if ($scheduleId) {
-            $job = $this->jobRepository->findLastForDataflowId($scheduleId);
+            $job = $this->jobGateway->findLastForDataflowId($scheduleId);
         } elseif ($jobId) {
-            $job = $this->jobRepository->find($jobId);
+            $job = $this->jobGateway->find($jobId);
         } else {
             $io->error('You must pass `job-id` or `schedule-id` option.');
 

src/DependencyInjection/CodeRhapsodieDataflowExtension.php

@@ -33,6 +33,7 @@ public function load(array $configs, ContainerBuilder $container): void
 
         if ($config['exceptions_mode']['type'] === 'file') {
             $container->setParameter('coderhapsodie.dataflow.flysystem_service', $config['exceptions_mode']['flysystem_service']);
+            $loader->load('exceptions_services.yaml');
         }
 
         if ($config['messenger_mode']['enabled']) {

src/DependencyInjection/Compiler/ExceptionCompilerPass.php

@@ -4,7 +4,7 @@
 
 namespace CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler;
 
-use CodeRhapsodie\DataflowBundle\Processor\JobProcessor;
+use CodeRhapsodie\DataflowBundle\ExceptionsHandler\FilesystemExceptionHandler;
 use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Exception\InvalidArgumentException;
@@ -23,7 +23,7 @@ public function process(ContainerBuilder $container): void
             throw new InvalidArgumentException(\sprintf('Service "%s" not found', $flysystem));
         }
 
-        $definition = $container->findDefinition(JobProcessor::class);
+        $definition = $container->findDefinition(FilesystemExceptionHandler::class);
         $definition->setArgument('$filesystem', new Reference($flysystem));
     }
 }

src/ExceptionsHandler/ExceptionHandlerInterface.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\ExceptionsHandler;
+
+interface ExceptionHandlerInterface
+{
+    public function save(?int $jobId, ?array $exceptions): void;
+
+    public function find(int $jobId): ?array;
+}

src/ExceptionsHandler/FilesystemExceptionHandler.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\ExceptionsHandler;
+
+use League\Flysystem\Filesystem;
+use League\Flysystem\FilesystemException;
+
+class FilesystemExceptionHandler implements ExceptionHandlerInterface
+{
+    public function __construct(private Filesystem $filesystem)
+    {
+    }
+
+    public function save(?int $jobId, ?array $exceptions): void
+    {
+        if ($jobId === null || empty($exceptions)) {
+            return;
+        }
+
+        $this->filesystem->write(sprintf('dataflow-job-%s.log', $jobId), json_encode($exceptions));
+    }
+
+    public function find(int $jobId): ?array
+    {
+        try {
+            if (!$this->filesystem->has(sprintf('dataflow-job-%s.log', $jobId))) {
+                return [];
+            }
+
+            return json_decode($this->filesystem->read(sprintf('dataflow-job-%s.log', $jobId)), true);
+        } catch (FilesystemException) {
+            return [];
+        }
+    }
+}

src/ExceptionsHandler/NullExceptionHandler.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\ExceptionsHandler;
+
+class NullExceptionHandler implements ExceptionHandlerInterface
+{
+
+    public function save(?int $jobId, ?array $exceptions): void
+    {
+    }
+
+    public function find(int $jobId): ?array
+    {
+        return null;
+    }
+}

src/Gateway/JobGateway.php

@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\Gateway;
+
+use CodeRhapsodie\DataflowBundle\Entity\Job;
+use CodeRhapsodie\DataflowBundle\ExceptionsHandler\ExceptionHandlerInterface;
+use CodeRhapsodie\DataflowBundle\ExceptionsHandler\NullExceptionHandler;
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+
+class JobGateway
+{
+    public function __construct(private JobRepository $repository, private ExceptionHandlerInterface $exceptionHandler)
+    {
+    }
+
+    public function find(int $jobId): ?Job
+    {
+        $job = $this->repository->find($jobId);
+
+        return $this->loadExceptions($job);
+    }
+
+    public function save(Job $job): void
+    {
+        if (!$this->exceptionHandler instanceof NullExceptionHandler) {
+            $this->exceptionHandler->save($job->getId(), $job->getExceptions());
+            $job->setExceptions([]);
+        }
+
+        $this->repository->save($job);
+    }
+
+    public function findLastForDataflowId(int $scheduleId): ?Job
+    {
+        $job = $this->repository->findLastForDataflowId($scheduleId);
+
+        return $this->loadExceptions($job);
+    }
+
+    private function loadExceptions(?Job $job): ?Job
+    {
+        if ($job === null || $this->exceptionHandler instanceof NullExceptionHandler) {
+            return $job;
+        }
+
+        $this->exceptionHandler->save($job->getId(), $job->getExceptions());
+
+        return $job->setExceptions($this->exceptionHandler->find($job->getId()));
+    }
+}

src/Processor/JobProcessor.php

@@ -9,6 +9,7 @@
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Event\Events;
 use CodeRhapsodie\DataflowBundle\Event\ProcessingEvent;
+use CodeRhapsodie\DataflowBundle\Gateway\JobGateway;
 use CodeRhapsodie\DataflowBundle\Logger\BufferHandler;
 use CodeRhapsodie\DataflowBundle\Logger\DelegatingLogger;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
@@ -27,7 +28,7 @@ public function __construct(
         private JobRepository $repository,
         private DataflowTypeRegistryInterface $registry,
         private EventDispatcherInterface $dispatcher,
-        private ?Filesystem $filesystem = null
+        private JobGateway $jobGateway,
     )
     {
     }
@@ -70,25 +71,19 @@ private function beforeProcessing(Job $job): void
             ->setStatus(Job::STATUS_RUNNING)
             ->setStartTime(new \DateTime())
         ;
-        $this->repository->save($job);
+        $this->jobGateway->save($job);
     }
 
     private function afterProcessing(Job $job, Result $result, BufferHandler $bufferLogger): void
     {
-        $exceptions = $bufferLogger->clearBuffer();
-        if ($this->filesystem) {
-            $this->filesystem->write(sprintf('dataflow-job-%s.log',$job->getId()), json_encode($exceptions));
-            $exceptions = [];
-        }
-
         $job
             ->setEndTime($result->getEndTime())
             ->setStatus(Job::STATUS_COMPLETED)
             ->setCount($result->getSuccessCount())
-            ->setExceptions($exceptions)
+            ->setExceptions($bufferLogger->clearBuffer())
         ;
 
-        $this->repository->save($job);
+        $this->jobGateway->save($job);
 
         $this->dispatcher->dispatch(new ProcessingEvent($job), Events::AFTER_PROCESSING);
     }

src/Resources/config/exceptions_services.yaml

@@ -0,0 +1,5 @@
+services:
+    CodeRhapsodie\DataflowBundle\ExceptionsHandler\ExceptionHandlerInterface: '@CodeRhapsodie\DataflowBundle\ExceptionsHandler\FilesystemExceptionHandler'
+    CodeRhapsodie\DataflowBundle\ExceptionsHandler\FilesystemExceptionHandler:
+        arguments:
+            $filesystem: ~ # Filled in compiler pass

src/Resources/config/services.yaml

@@ -28,7 +28,7 @@ services:
 
     CodeRhapsodie\DataflowBundle\Command\JobShowCommand:
         arguments:
-            $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
+            $jobGateway: '@CodeRhapsodie\DataflowBundle\Gateway\JobGateway'
             $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
         tags: ['console.command']
 
@@ -93,3 +93,10 @@ services:
             $repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
             $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
             $dispatcher: '@event_dispatcher'
+            $jobGateway: '@CodeRhapsodie\DataflowBundle\Gateway\JobGateway'
+    CodeRhapsodie\DataflowBundle\ExceptionsHandler\ExceptionHandlerInterface: '@CodeRhapsodie\DataflowBundle\ExceptionsHandler\NullExceptionHandler'
+    CodeRhapsodie\DataflowBundle\ExceptionsHandler\NullExceptionHandler:
+    CodeRhapsodie\DataflowBundle\Gateway\JobGateway:
+        arguments:
+            $repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
+            $exceptionHandler: '@CodeRhapsodie\DataflowBundle\ExceptionsHandler\ExceptionHandlerInterface'