feat: Etl progress information is now saved on the execution entity(updated every 5 seconds)

Author: oliverde8

CHANGELOG.md

@@ -1,8 +1,10 @@
 # 1.1.0
 
-- :star2: Support for php-etl 1.1 new operations has been added.
+- :star2: Support for php-etl 1.2 new operations has been added. (Skipped 1.1)
 - :star2: Support for symfony 6.0 has been added.
 - :star2: Support for using input parameters in operation options has been added.
+- :star2: Added support to the command line to displayt ETL progress, (feature from php-etl 1.2)
+- :star2: Etl progress information is now saved on the execution entity(updated every 5 seconds)
 - :collision: All operations are no longer built during DI compilation. This should improve peformance of cache warmup.
 - :collision: Support for Symfony 4.4 has been dropped.
 - :collision: Support for php 7.4 has been dropped.

Entity/EtlExecution.php

@@ -7,6 +7,7 @@
 
 /**
  * @ORM\Entity(repositoryClass=EtlExecutionRepository::class)
+ *
  */
 class EtlExecution
 {
@@ -255,7 +256,7 @@ public function setRunTime(int $runTime): void
     {
         $this->runTime = $runTime;
     }
-    
+
     /**
      * @return mixed
      */
@@ -286,6 +287,10 @@ public function setDefinition(string $definition): self
 
     public function getStepStats(): ?string
     {
+        if ($this->stepStats == "[]") {
+            // Legacy json stuff.
+            return null;
+        }
         return $this->stepStats;
     }
 

Services/ChainProcessorsManager.php

@@ -7,6 +7,7 @@
 use Oliverde8\PhpEtlBundle\Exception\UnknownChainException;
 use Oliverde8\PhpEtlBundle\Factory\ChainFactory;
 use Oliverde8\PhpEtlBundle\Repository\EtlExecutionRepository;
+use function Clue\StreamFilter\fun;
 
 class ChainProcessorsManager
 {
@@ -124,7 +125,18 @@ public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator
             ];
 
             // Start the process.
-            $processor->process($iterator, $params, $observerCallback);
+            $observerProcessTime = 0;
+            $processor->process($iterator, $params, function (array $operationStates, int $processedItems, int $returnedItems, bool $hasFinished = false) use ($observerCallback, &$observerProcessTime, $execution) {
+                $observerCallback($operationStates, $processedItems, $returnedItems, $hasFinished);
+
+                if ((time() - $observerProcessTime) > 5) {
+                    $execution = $this->etlExecutionRepository->find($execution->getId());
+                    $execution->setStepStats(serialize($operationStates));
+                    $this->etlExecutionRepository->save($execution);
+
+                    $observerProcessTime = time();
+                }
+            });
             $execution = $this->etlExecutionRepository->find($execution->getId());
             $execution->setStatus(EtlExecution::STATUS_SUCCESS);
         } catch (\Throwable $exception) {
@@ -136,7 +148,6 @@ public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator
         } finally {
             $execution->setEndTime(new \DateTime());
             $execution->setRunTime(time() - $execution->getStartTime()->getTimestamp());
-            $execution->setStepStats('[]'); // To be developped
             $this->etlExecutionRepository->save($execution);
         }
     }