Add auto update count processed item while running job

Author: AUDUL

src/Command/ExecuteDataflowCommand.php

@@ -4,6 +4,7 @@
 
 namespace CodeRhapsodie\DataflowBundle\Command;
 
+use CodeRhapsodie\DataflowBundle\DataflowType\RepositoryInterface;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
@@ -62,7 +63,10 @@ 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 RepositoryInterface) {
+            $dataflowType->setRepository($this->jobRepository);
+        }
+
         if ($dataflowType instanceof LoggerAwareInterface && isset($this->logger)) {
             $dataflowType->setLogger($this->logger);
         }

src/DataflowType/AbstractDataflowType.php

@@ -10,7 +10,7 @@
 use Psr\Log\LoggerInterface;
 use Symfony\Component\OptionsResolver\OptionsResolver;
 
-abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwareInterface
+abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwareInterface, RepositoryInterface
 {
     use LoggerAwareTrait;
 
@@ -28,18 +28,21 @@ public function getAliases(): iterable
 
     public function process(array $options, ?int $jobId = null): Result
     {
+        $this->saveDate = new \DateTime();
+
         $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()) {
+            if ($jobId === null || $this->saveDate > new \DateTime()) {
                 return;
             }
 
             $this->repository->updateCount($jobId, $count);
+            $this->saveDate = new \DateTime('+1 minute');
         });
         $this->buildDataflow($builder, $options);
         $dataflow = $builder->getDataflow();

src/DataflowType/Dataflow/Dataflow.php

@@ -95,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;

src/DataflowType/DataflowTypeInterface.php

@@ -13,6 +13,4 @@ public function getLabel(): string;
     public function getAliases(): iterable;
 
     public function process(array $options, ?int $jobId = null): Result;
-
-    public function setRepository(JobRepository $repository): void;
 }

src/DataflowType/RepositoryInterface.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\DataflowType;
+
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+
+interface RepositoryInterface
+{
+    public function setRepository(JobRepository $repository): void;
+}

src/Processor/JobProcessor.php

@@ -4,6 +4,7 @@
 
 namespace CodeRhapsodie\DataflowBundle\Processor;
 
+use CodeRhapsodie\DataflowBundle\DataflowType\RepositoryInterface;
 use CodeRhapsodie\DataflowBundle\DataflowType\Result;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Event\Events;
@@ -30,7 +31,10 @@ public function process(Job $job): void
         $this->beforeProcessing($job);
 
         $dataflowType = $this->registry->getDataflowType($job->getDataflowType());
-        $dataflowType->setRepository($this->repository);
+        if ($dataflowType instanceof RepositoryInterface) {
+            $dataflowType->setRepository($this->repository);
+        }
+
         $loggers = [new Logger('dataflow_internal', [$bufferHandler = new BufferHandler()])];
         if (isset($this->logger)) {
             $loggers[] = $this->logger;

src/Resources/config/services.yaml

@@ -23,6 +23,7 @@ services:
         arguments:
             $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
             $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
+            $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
         tags: ['console.command']
 
     CodeRhapsodie\DataflowBundle\Command\JobShowCommand: