Add auto update count processed item while running job

Author: AUDUL

CHANGELOG.md

@@ -1,3 +1,6 @@
+# Version 5.3.0
+* Added auto update count processed item while running job
+
 # Version 5.2.0
 * Added custom index for job status
 

src/Command/ExecuteDataflowCommand.php

@@ -6,6 +6,7 @@
 
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
 use Symfony\Component\Console\Attribute\AsCommand;
@@ -26,7 +27,7 @@ class ExecuteDataflowCommand extends Command implements LoggerAwareInterface
 {
     use LoggerAwareTrait;
 
-    public function __construct(private DataflowTypeRegistryInterface $registry, private ConnectionFactory $connectionFactory)
+    public function __construct(private DataflowTypeRegistryInterface $registry, private ConnectionFactory $connectionFactory, private JobRepository $jobRepository)
     {
         parent::__construct();
     }
@@ -61,6 +62,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $io = new SymfonyStyle($input, $output);
 
         $dataflowType = $this->registry->getDataflowType($fqcnOrAlias);
+        $dataflowType->setRepository($this->jobRepository);
         if ($dataflowType instanceof LoggerAwareInterface && isset($this->logger)) {
             $dataflowType->setLogger($this->logger);
         }

src/DataflowType/AbstractDataflowType.php

@@ -4,6 +4,7 @@
 
 namespace CodeRhapsodie\DataflowBundle\DataflowType;
 
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
 use Psr\Log\LoggerInterface;
@@ -13,6 +14,10 @@ abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwar
 {
     use LoggerAwareTrait;
 
+    private JobRepository $repository;
+
+    private ?\DateTime $saveDate = null;
+
     /**
      * @codeCoverageIgnore
      */
@@ -21,14 +26,21 @@ public function getAliases(): iterable
         return [];
     }
 
-    public function process(array $options): Result
+    public function process(array $options, ?int $jobId = null): Result
     {
         $optionsResolver = new OptionsResolver();
         $this->configureOptions($optionsResolver);
         $options = $optionsResolver->resolve($options);
 
         $builder = $this->createDataflowBuilder();
         $builder->setName($this->getLabel());
+        $builder->addAfterItemProcessors(function (int|string $index, mixed $item, int $count) use ($jobId) {
+            if ($jobId === null || $this->saveDate->modify('+1 minute') > new \DateTime()) {
+                return;
+            }
+
+            $this->repository->updateCount($jobId, $count);
+        });
         $this->buildDataflow($builder, $options);
         $dataflow = $builder->getDataflow();
         if ($dataflow instanceof LoggerAwareInterface && $this->logger instanceof LoggerInterface) {
@@ -51,4 +63,9 @@ protected function configureOptions(OptionsResolver $optionsResolver): void
     }
 
     abstract protected function buildDataflow(DataflowBuilder $builder, array $options): void;
+
+    public function setRepository(JobRepository $repository): void
+    {
+        $this->repository = $repository;
+    }
 }

src/DataflowType/Dataflow/Dataflow.php

@@ -21,6 +21,13 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
 
     private ?\Closure $customExceptionIndex = null;
 
+    private ?\DateTimeInterface $dateTime = null;
+
+    /**
+     * @var \Closure[]
+     */
+    private array $afterItemProcessors = [];
+
     public function __construct(private iterable $reader, private ?string $name)
     {
     }
@@ -55,6 +62,18 @@ public function setCustomExceptionIndex(callable $callable): self
         return $this;
     }
 
+    /**
+     * @param array<callable> $processors
+     */
+    public function setAfterItemProcessors(array $processors): self
+    {
+        $this->afterItemProcessors = array_map(function (callable $callable) {
+            return \Closure::fromCallable($callable);
+        }, $processors);
+
+        return $this;
+    }
+
     /**
      * {@inheritdoc}
      */
@@ -76,7 +95,7 @@ public function process(): Result
                     $exceptionIndex = $index;
                     try {
                         if (is_callable($this->customExceptionIndex)) {
-                            $exceptionIndex = (string) ($this->customExceptionIndex)($item, $index);
+                            $exceptionIndex = (string)($this->customExceptionIndex)($item, $index);
                         }
                     } catch (\Throwable $e2) {
                         $exceptions[$index] = $e2;
@@ -87,6 +106,10 @@ public function process(): Result
                 }
 
                 ++$count;
+
+                foreach ($this->afterItemProcessors as $afterItemProcessor) {
+                    $afterItemProcessor($index, $item, $count);
+                }
             }
 
             foreach ($this->writers as $writer) {

src/DataflowType/DataflowBuilder.php

@@ -18,6 +18,10 @@ class DataflowBuilder
     private array $writers = [];
 
     private ?\Closure $customExceptionIndex = null;
+    /**
+     * @var \Closure[]
+     */
+    private array $afterItemProcessors = [];
 
     public function setName(string $name): self
     {
@@ -54,6 +58,13 @@ public function setCustomExceptionIndex(callable $callable): self
         return $this;
     }
 
+    public function addAfterItemProcessors(callable $callable): self
+    {
+        $this->afterItemProcessors[] = \Closure::fromCallable($callable);
+
+        return $this;
+    }
+
     public function getDataflow(): DataflowInterface
     {
         $dataflow = new Dataflow($this->reader, $this->name);
@@ -73,6 +84,8 @@ public function getDataflow(): DataflowInterface
             $dataflow->setCustomExceptionIndex($this->customExceptionIndex);
         }
 
+        $dataflow->setAfterItemProcessors($this->afterItemProcessors);
+
         return $dataflow;
     }
 }

src/DataflowType/DataflowTypeInterface.php

@@ -4,11 +4,15 @@
 
 namespace CodeRhapsodie\DataflowBundle\DataflowType;
 
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+
 interface DataflowTypeInterface
 {
     public function getLabel(): string;
 
     public function getAliases(): iterable;
 
-    public function process(array $options): Result;
+    public function process(array $options, ?int $jobId = null): Result;
+
+    public function setRepository(JobRepository $repository): void;
 }

src/Processor/JobProcessor.php

@@ -30,6 +30,7 @@ public function process(Job $job): void
         $this->beforeProcessing($job);
 
         $dataflowType = $this->registry->getDataflowType($job->getDataflowType());
+        $dataflowType->setRepository($this->repository);
         $loggers = [new Logger('dataflow_internal', [$bufferHandler = new BufferHandler()])];
         if (isset($this->logger)) {
             $loggers[] = $this->logger;
@@ -40,7 +41,7 @@ public function process(Job $job): void
             $dataflowType->setLogger($logger);
         }
 
-        $result = $dataflowType->process($job->getOptions());
+        $result = $dataflowType->process($job->getOptions(), $job->getId());
 
         if (!$dataflowType instanceof LoggerAwareInterface) {
             foreach ($result->getExceptions() as $index => $e) {

src/Repository/JobRepository.php

@@ -137,6 +137,11 @@ public function save(Job $job)
         $this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], $this->getFields());
     }
 
+    public function updateCount(int $jobId, int $count): void
+    {
+        $this->connection->update(static::TABLE_NAME, ['count' => $count], ['id' => $jobId]);
+    }
+
     public function createQueryBuilder($alias = null): QueryBuilder
     {
         $qb = $this->connection->createQueryBuilder();