* Add streamed exceptions

Author: AUDUL

composer.json

@@ -64,7 +64,8 @@
         "phpunit/phpunit": "^11",
         "portphp/portphp": "^1.9",
         "rector/rector": "^2.0",
-        "symfony/messenger": "^7.0"
+        "symfony/messenger": "^7.0",
+        "league/flysystem": "3.*"
     },
     "suggest": {
         "amphp/amp": "Provide asynchronous steps for your dataflows",

src/DataflowType/AbstractDataflowType.php

@@ -46,7 +46,7 @@ public function process(array $options, ?int $jobId = null): Result
         });
         $this->buildDataflow($builder, $options);
         $dataflow = $builder->getDataflow();
-        if ($dataflow instanceof LoggerAwareInterface && $this->logger instanceof LoggerInterface) {
+        if ($this->logger instanceof LoggerInterface) {
             $dataflow->setLogger($this->logger);
         }
 

src/DataflowType/Dataflow/Dataflow.php

@@ -9,7 +9,7 @@
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
 
-class Dataflow implements DataflowInterface, LoggerAwareInterface
+class Dataflow implements DataflowInterface
 {
     use LoggerAwareTrait;
 
@@ -75,7 +75,7 @@ public function setAfterItemProcessors(array $processors): self
     public function process(): Result
     {
         $count = 0;
-        $exceptions = [];
+        $countExceptions = 0;
         $startTime = new \DateTime();
 
         try {
@@ -93,10 +93,10 @@ public function process(): Result
                             $exceptionIndex = (string) ($this->customExceptionIndex)($item, $index);
                         }
                     } catch (\Throwable $e2) {
-                        $exceptions[$index] = $e2;
+                        ++$countExceptions;
                         $this->logException($e2, $index);
                     }
-                    $exceptions[$exceptionIndex] = $e;
+                    ++$countExceptions;
                     $this->logException($e, $exceptionIndex);
                 }
 
@@ -111,11 +111,11 @@ public function process(): Result
                 $writer->finish();
             }
         } catch (\Throwable $e) {
-            $exceptions[] = $e;
+            ++$countExceptions;
             $this->logException($e);
         }
 
-        return new Result($this->name, $startTime, new \DateTime(), $count, $exceptions);
+        return new Result($this->name, $startTime, new \DateTime(), $count, $countExceptions);
     }
 
     private function processItem(mixed $item): void

src/DataflowType/Dataflow/DataflowInterface.php

@@ -5,11 +5,12 @@
 namespace CodeRhapsodie\DataflowBundle\DataflowType\Dataflow;
 
 use CodeRhapsodie\DataflowBundle\DataflowType\Result;
+use Psr\Log\LoggerAwareInterface;
 
 /**
  * Combines a reader, steps and writers as a data processing workflow.
  */
-interface DataflowInterface
+interface DataflowInterface extends LoggerAwareInterface
 {
     /**
      * Processes the data.

src/DataflowType/DataflowTypeInterface.php

@@ -4,7 +4,9 @@
 
 namespace CodeRhapsodie\DataflowBundle\DataflowType;
 
-interface DataflowTypeInterface
+use Psr\Log\LoggerAwareInterface;
+
+interface DataflowTypeInterface extends LoggerAwareInterface
 {
     public function getLabel(): string;
 

src/DataflowType/Result.php

@@ -11,18 +11,12 @@ class Result
 {
     private \DateInterval $elapsed;
 
-    private int $errorCount = 0;
-
     private int $successCount = 0;
 
-    private array $exceptions;
-
-    public function __construct(private string $name, private \DateTimeInterface $startTime, private \DateTimeInterface $endTime, private int $totalProcessedCount, array $exceptions)
+    public function __construct(private string $name, private \DateTimeInterface $startTime, private \DateTimeInterface $endTime, private int $totalProcessedCount, private int $errorCount)
     {
         $this->elapsed = $startTime->diff($endTime);
-        $this->errorCount = \count($exceptions);
         $this->successCount = $totalProcessedCount - $this->errorCount;
-        $this->exceptions = $exceptions;
     }
 
     public function getName(): string

src/Entity/Job.php

@@ -61,6 +61,11 @@ class Job
 
     private ?\DateTimeInterface $endTime = null;
 
+    /**
+     * @var resource|null
+     */
+    private $streamExceptions = null;
+
     public static function createFromScheduledDataflow(ScheduledDataflow $scheduled): self
     {
         return (new static())
@@ -243,4 +248,22 @@ public function setEndTime(?\DateTimeInterface $endTime): self
 
         return $this;
     }
+
+    /**
+     * @return resource|null
+     */
+    public function getStreamExceptions()
+    {
+        return $this->streamExceptions;
+    }
+
+    /**
+     * @param resource $streamExceptions
+     */
+    public function setStreamExceptions($streamExceptions): self
+    {
+        $this->streamExceptions = $streamExceptions;
+
+        return $this;
+    }
 }

src/ExceptionsHandler/ExceptionHandlerInterface.php

@@ -6,7 +6,9 @@
 
 interface ExceptionHandlerInterface
 {
-    public function save(?int $jobId, ?array $exceptions): void;
+    /** @param resource $exceptions */
+    public function save(?int $jobId, $exceptions): void;
 
-    public function find(int $jobId): ?array;
+    /** @return null|resource */
+    public function find(int $jobId);
 }

src/ExceptionsHandler/FilesystemExceptionHandler.php

@@ -13,25 +13,51 @@ public function __construct(private Filesystem $filesystem)
     {
     }
 
-    public function save(?int $jobId, ?array $exceptions): void
+    /** @inheritDoc */
+    public function save(?int $jobId, $exceptions): void
     {
-        if ($jobId === null || empty($exceptions)) {
+        if ($jobId === null || stream_get_contents($exceptions, 1) === false) {
             return;
         }
 
-        $this->filesystem->write(\sprintf('dataflow-job-%s.log', $jobId), json_encode($exceptions));
+        $path = \sprintf('dataflow-job-%s.log', $jobId);
+        rewind($exceptions);
+
+        if ($this->filesystem->fileExists($path)) {
+            $existingStream = $this->filesystem->readStream($path);
+
+            $combined = fopen('php://temp', 'r+');
+
+            stream_copy_to_stream($existingStream, $combined);
+            stream_copy_to_stream($exceptions, $combined);
+
+            rewind($combined);
+
+            $this->filesystem->delete($path);
+            $this->filesystem->writeStream($path, $combined);
+
+            fclose($existingStream);
+            fclose($combined);
+
+            return;
+        }
+
+        $this->filesystem->writeStream($path, $exceptions);
+
+        fclose($exceptions);
     }
 
-    public function find(int $jobId): ?array
+    /** @inheritDoc */
+    public function find(int $jobId)
     {
         try {
             if (!$this->filesystem->fileExists(\sprintf('dataflow-job-%s.log', $jobId))) {
-                return [];
+                return null;
             }
 
-            return json_decode($this->filesystem->read(\sprintf('dataflow-job-%s.log', $jobId)), true);
+            return $this->filesystem->readStream(\sprintf('dataflow-job-%s.log', $jobId));
         } catch (FilesystemException) {
-            return [];
+            return null;
         }
     }
 }

src/ExceptionsHandler/NullExceptionHandler.php

@@ -6,11 +6,13 @@
 
 class NullExceptionHandler implements ExceptionHandlerInterface
 {
-    public function save(?int $jobId, ?array $exceptions): void
+    /** @inheritDoc */
+    public function save(?int $jobId, $exceptions): void
     {
     }
 
-    public function find(int $jobId): ?array
+    /** @inheritDoc */
+    public function find(int $jobId)
     {
         return null;
     }