feat: added support for php-etl v1.2

Author: oliverde8

Command/ExecuteCommand.php

@@ -2,17 +2,20 @@
 
 namespace Oliverde8\PhpEtlBundle\Command;
 
+use Oliverde8\Component\PhpEtl\Output\SymfonyConsoleOutput;
 use Oliverde8\PhpEtlBundle\Services\ChainProcessorsManager;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
 class ExecuteCommand extends Command
 {
     const ARGUMENT_NAME = "name";
     const ARGUMENT_DATA = "data";
     const ARGUMENT_PARAMS = "params";
+    const OPTION_PRETTY = "pretty";
 
     protected ChainProcessorsManager $chainProcessorsManager;
 
@@ -32,6 +35,7 @@ protected function configure()
         $this->addArgument(self::ARGUMENT_NAME, InputArgument::REQUIRED);
         $this->addArgument(self::ARGUMENT_DATA, InputArgument::OPTIONAL, "json with the input array");
         $this->addArgument(self::ARGUMENT_PARAMS, InputArgument::OPTIONAL, "json with all the additional parameters");
+        $this->addOption(self::OPTION_PRETTY, "p", InputOption::VALUE_NONE, "Disables pretty output");
     }
 
     /**
@@ -43,7 +47,15 @@ protected function execute(InputInterface $input, OutputInterface $output)
         $options = json_decode($input->getArgument(self::ARGUMENT_PARAMS) ?? '[]', true);
         $data = json_decode($input->getArgument(self::ARGUMENT_DATA) ?? '[]', true);
 
-        $this->chainProcessorsManager->execute($chainName, $data, $options);
+        $processorOutput = new SymfonyConsoleOutput($output, 0);
+        $observation = function (array $operationStates) use ($processorOutput) {
+            $processorOutput->output($operationStates);
+        };
+        if ($input->getOption(self::OPTION_PRETTY)) {
+            $observation = function (array $operationStates)  {};
+        }
+
+        $this->chainProcessorsManager->execute($chainName, $data, $options, $observation);
         return 0;
     }
 }

Services/ChainProcessorsManager.php

@@ -77,7 +77,7 @@ public function getProcessor(string $chainName, array $options): ChainProcessor
      *
      * @throws \Exception
      */
-    public function execute(string $chainName, iterable $iterator, array $params)
+    public function execute(string $chainName, iterable $iterator, array $params,  ?callable $observerCallback = null)
     {
         $definition = $this->getRawDefinition($chainName);
 
@@ -91,14 +91,14 @@ public function execute(string $chainName, iterable $iterator, array $params)
         $execution->setStatus(EtlExecution::STATUS_RUNNING);
         $this->etlExecutionRepository->save($execution);
 
-        $this->executeFromEtlEntity($execution, $iterator);
+        $this->executeFromEtlEntity($execution, $iterator, $observerCallback);
     }
 
     /**
      * Execute a chain from it's entity.
      *
      */
-    public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator = null): void
+    public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator = null, ?callable $observerCallback = null): void
     {
         $chainName = $execution->getName();
         $logger = $this->loggerFactory->get($execution);
@@ -124,7 +124,7 @@ public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator
             ];
 
             // Start the process.
-            $processor->process($iterator, $params);
+            $processor->process($iterator, $params, $observerCallback);
             $execution = $this->etlExecutionRepository->find($execution->getId());
             $execution->setStatus(EtlExecution::STATUS_SUCCESS);
         } catch (\Throwable $exception) {

composer.json

@@ -11,9 +11,9 @@
   ],
   "require": {
     "php": ">=8.0",
-    "oliverde8/php-etl": "^v1.1.0-alpha1",
-    "symfony/framework-bundle": "^5.4|^6.0",
-    "symfony/messenger": "^5.4|^6.0",
+    "oliverde8/php-etl": "^v1.2.0-alpha1",
+    "symfony/framework-bundle": "^5.4|^6.0|^7.0",
+    "symfony/messenger": "^5.4|^6.0|^7.0",
     "ext-json": "*"
   },
   "autoload": {
@@ -23,6 +23,6 @@
   },
   "require-dev": {
     "symfony/monolog-bundle": "^3.7",
-    "symfony/console": "^5.4|^6.0"
+    "symfony/console": "^6.0|^7.0"
   }
 }