oliverde8 / phpEtlBundle

Services/ChainProcessorsManager.php

<?php

namespace Oliverde8\PhpEtlBundle\Services;

use Oliverde8\Component\PhpEtl\ChainProcessor;
use Oliverde8\PhpEtlBundle\Entity\EtlExecution;
use Oliverde8\PhpEtlBundle\Exception\UnknownChainException;
use Oliverde8\PhpEtlBundle\Factory\ChainFactory;
use Oliverde8\PhpEtlBundle\Repository\EtlExecutionRepository;

class ChainProcessorsManager
{
    protected EtlExecutionRepository $etlExecutionRepository;
    protected LoggerFactory $loggerFactory;
    protected ChainFactory $chainFactory;
    protected array $definitions;
    protected array $rewDefinitions;

    public function __construct(
        EtlExecutionRepository $etlExecutionRepository,
        LoggerFactory $loggerFactory,
        ChainFactory $chainFactory,
        array $definitions,
        array $rawDefinitions
    ) {
        $this->etlExecutionRepository = $etlExecutionRepository;
        $this->loggerFactory = $loggerFactory;
        $this->chainFactory = $chainFactory;
        $this->definitions = $definitions;
        $this->rewDefinitions = $rawDefinitions;
    }

    /**
     * @throws UnknownChainException
     */
    public function getRawDefinition(string $chainName): string
    {
        if (!isset($this->rewDefinitions[$chainName])) {
            $alternatives = [];
            foreach (array_keys($this->rewDefinitions) as $knownId) {
                $lev = levenshtein($chainName, $knownId);
                if ($lev <= \strlen($chainName) / 3 || str_contains($knownId, $chainName)) {
                    $alternatives[] = $knownId;
                }
            }

            throw new UnknownChainException("Unknown chain '$chainName', did you mean: " . implode(", ", $alternatives));
        }

        return $this->rewDefinitions[$chainName];
    }

    public function getRewDefinitions(): array
    {
        return $this->rewDefinitions;
    }

    public function getProcessor(string $chainName, array $options): ChainProcessor
    {
        $this->getRawDefinition($chainName);
        $definition = $this->definitions[$chainName];
        $chain = $definition['chain'];
        $maxAsynchronousItems = $definition['maxAsynchronousItems'] ?? 20;
        $defaultOptions = $definition['defaultOptions'] ?? [];

        $options = array_merge($defaultOptions, $options);

        return $this->chainFactory->create($chain, $options, $maxAsynchronousItems);
    }

    /**
     * Execute a particular chanin
     *
     * @param string $chainName
     * @param iterable $iterator
     * @param array $params
     *
     * @throws \Exception
     */
    public function execute(string $chainName, iterable $iterator, array $params)
    {
        $definition = $this->getRawDefinition($chainName);

        $inputData = ["Iterator! Can't show input data"];
        if (is_array($iterator)) {
            $inputData = $iterator;
            $iterator = new \ArrayIterator($iterator);
        }

        $execution = new EtlExecution($chainName, $definition, $inputData, $params);
        $execution->setStatus(EtlExecution::STATUS_RUNNING);
        $this->etlExecutionRepository->save($execution);

        $this->executeFromEtlEntity($execution, $iterator);
    }

    /**
     * Execute a chain from it's entity.
     *
     */
    public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator = null): void
    {
        $chainName = $execution->getName();
        $logger = $this->loggerFactory->get($execution);

The assignment to $logger is dead and can be removed.

        try {
            // Update execution object with new status.
            $execution->setStatus(EtlExecution::STATUS_RUNNING);
            $execution->setStartTime(new \DateTime());
            $execution->setWaitTime(time() - $execution->getCreateTime()->getTimestamp());
            $this->etlExecutionRepository->save($execution);

            // Build the processor.
            $params = json_decode($execution->getInputOptions(), true);

It seems like $execution->getInputOptions() can also be of type null; however, parameter $json of json_decode() does only seem to accept string, maybe add an additional type check?

            $processor = $this->getProcessor($chainName, $params);

            if (is_null($iterator)) {
                $iterator = new \ArrayIterator(json_decode($execution->getInputData(), true));
            }
            $params['etl'] = [
                'chain' => $chainName,
                'startTime' => new \DateTime(),
                'execution' => $execution
            ];

            // Start the process.
            $processor->process($iterator, $params);
            $execution = $this->etlExecutionRepository->find($execution->getId());
            $execution->setStatus(EtlExecution::STATUS_SUCCESS);
        } catch (\Throwable $exception) {
            $execution = $this->etlExecutionRepository->find($execution->getId());
            $execution->setFailTime(new \DateTime());
            $execution->setStatus(EtlExecution::STATUS_FAILURE);
            $execution->setErrorMessage($this->getFullExceptionTrace($exception));
            throw $exception;
        } finally {
            $execution->setEndTime(new \DateTime());
            $execution->setRunTime(time() - $execution->getStartTime()->getTimestamp());
            $execution->setStepStats('[]'); // To be developped
            $this->etlExecutionRepository->save($execution);

It seems like $execution can also be of type null; however, parameter $execution of Oliverde8\PhpEtlBundle\Repository\EtlExecutionRepository::save() does only seem to accept Oliverde8\PhpEtlBundle\Entity\EtlExecution, maybe add an additional type check?

        }
    }

    protected function getFullExceptionTrace(\Throwable $exception): string
    {
        $message = '';
        do {
            $message .= $exception->getMessage() . "\n" . $exception->getTraceAsString() . "\n\n";
        } while ($exception = $exception->getPrevious());

        return $message;
    }
}