* Add possibility to save exceptions in file

Author: AUDUL

src/DependencyInjection/CodeRhapsodieDataflowExtension.php

@@ -15,7 +15,7 @@
  */
 class CodeRhapsodieDataflowExtension extends Extension
 {
-    public function load(array $configs, ContainerBuilder $container)
+    public function load(array $configs, ContainerBuilder $container): void
     {
         $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('services.yaml');
@@ -29,6 +29,11 @@ public function load(array $configs, ContainerBuilder $container)
 
         $container->setParameter('coderhapsodie.dataflow.dbal_default_connection', $config['dbal_default_connection']);
         $container->setParameter('coderhapsodie.dataflow.default_logger', $config['default_logger']);
+        $container->setParameter('coderhapsodie.dataflow.exception_mode.type', $config['exception_mode']['type']);
+
+        if ($config['exception_mode']['type'] === 'file') {
+            $container->setParameter('coderhapsodie.dataflow.flysystem_service', $config['exception_mode']['flysystem_service']);
+        }
 
         if ($config['messenger_mode']['enabled']) {
             $container->setParameter('coderhapsodie.dataflow.bus', $config['messenger_mode']['bus']);

src/DependencyInjection/Compiler/BusCompilerPass.php

@@ -12,7 +12,7 @@
 
 class BusCompilerPass implements CompilerPassInterface
 {
-    public function process(ContainerBuilder $container)
+    public function process(ContainerBuilder $container): void
     {
         if (!$container->hasParameter('coderhapsodie.dataflow.bus')) {
             return;

src/DependencyInjection/Compiler/DataflowTypeCompilerPass.php

@@ -16,7 +16,7 @@
  */
 class DataflowTypeCompilerPass implements CompilerPassInterface
 {
-    public function process(ContainerBuilder $container)
+    public function process(ContainerBuilder $container): void
     {
         if (!$container->has(DataflowTypeRegistry::class)) {
             return;

src/DependencyInjection/Compiler/DefaultLoggerCompilerPass.php

@@ -12,7 +12,7 @@
 
 class DefaultLoggerCompilerPass implements CompilerPassInterface
 {
-    public function process(ContainerBuilder $container)
+    public function process(ContainerBuilder $container): void
     {
         $defaultLogger = $container->getParameter('coderhapsodie.dataflow.default_logger');
         if (!$container->has($defaultLogger)) {

src/DependencyInjection/Compiler/ExceptionCompilerPass.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler;
+
+use CodeRhapsodie\DataflowBundle\Processor\JobProcessor;
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Exception\InvalidArgumentException;
+use Symfony\Component\DependencyInjection\Reference;
+
+class ExceptionCompilerPass implements CompilerPassInterface
+{
+    public function process(ContainerBuilder $container): void
+    {
+        if (!$container->hasParameter('coderhapsodie.dataflow.flysystem_service')) {
+            return;
+        }
+
+        $flysystem = $container->getParameter('coderhapsodie.dataflow.flysystem_service');
+        if (!$container->has($flysystem)) {
+            throw new InvalidArgumentException(\sprintf('Service "%s" not found', $flysystem));
+        }
+
+        $definition = $container->findDefinition(JobProcessor::class);
+        $definition->setArgument('$filesystem', new Reference($flysystem));
+    }
+}

src/DependencyInjection/Configuration.php

@@ -38,6 +38,20 @@ public function getConfigTreeBuilder(): TreeBuilder
                         ->thenInvalid('You need "symfony/messenger" in order to use Dataflow messenger mode.')
                     ->end()
                 ->end()
+                ->arrayNode('exceptions_mode')
+                    ->addDefaultsIfNotSet()
+                    ->children()
+                        ->scalarNode('type')
+                            ->defaultValue('database')
+                        ->end()
+                        ->scalarNode('flysystem_service')
+                        ->end()
+                        ->validate()
+                            ->ifTrue(static fn($v): bool => 'file' === $v['type'] && !interface_exists('\League\Flysystem\Filesystem'))
+                            ->thenInvalid('You need "league/flysystem" to use Dataflow file exception mode.')
+                        ->end()
+                    ->end()
+                ->end()
             ->end()
         ;
 

src/Processor/JobProcessor.php

@@ -13,6 +13,7 @@
 use CodeRhapsodie\DataflowBundle\Logger\DelegatingLogger;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use League\Flysystem\Filesystem;
 use Monolog\Logger;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
@@ -22,7 +23,12 @@ class JobProcessor implements JobProcessorInterface, LoggerAwareInterface
 {
     use LoggerAwareTrait;
 
-    public function __construct(private JobRepository $repository, private DataflowTypeRegistryInterface $registry, private EventDispatcherInterface $dispatcher)
+    public function __construct(
+        private JobRepository $repository,
+        private DataflowTypeRegistryInterface $registry,
+        private EventDispatcherInterface $dispatcher,
+        private ?Filesystem $filesystem = null
+    )
     {
     }
 
@@ -69,12 +75,19 @@ private function beforeProcessing(Job $job): void
 
     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($bufferLogger->clearBuffer())
+            ->setExceptions($exceptions)
         ;
+
         $this->repository->save($job);
 
         $this->dispatcher->dispatch(new ProcessingEvent($job), Events::AFTER_PROCESSING);