data_pipelines-1.x-dev/src/TransformValidDataIterator.php

<?php

declare(strict_types=1);

namespace Drupal\data_pipelines;

use Drupal\data_pipelines\Entity\DatasetInterface;
use Drupal\data_pipelines\Exception\TransformSkipRecordException;
use Symfony\Component\Validator\ConstraintViolation;

/**
 * The valid data transform iterator.
 */
class TransformValidDataIterator extends \FilterIterator {

  /**
   * Current position.
   *
   * @var int
   */
  private int $position = 0;

  /**
   * Invalid deltas.
   *
   * @var array<int>
   */
  private array $invalidDeltas = [];

  /**
   * The valid data transform iterator constructor.
   *
   * @param \Iterator $iterator
   *   The iterator.
   * @param \Drupal\data_pipelines\Entity\DatasetInterface $dataset
   *   The dataset entity.
   * @param \Drupal\data_pipelines\DatasetPipelineInterface $pipeline
   *   The dataset pipeline.
   * @param int $seek_from
   *   Seek position.
   */
  public function __construct(\Iterator $iterator, private DatasetInterface $dataset, private DatasetPipelineInterface $pipeline, protected int $seek_from = 0) {
    parent::__construct($iterator);
  }

  /**
   * {@inheritdoc}
   */
  public function accept(): bool {
    if ($this->position < $this->seek_from) {
      $this->position++;
      return FALSE;
    }
    $delta = $this->position;
    $this->position++;
    $errors = $this->pipeline->validate($this->getInnerIterator()->current());
    if ($errors->count() > 0) {
      $this->invalidDeltas[] = $delta;
      foreach ($errors as $validation_error) {
        assert($validation_error instanceof ConstraintViolation);
        $this->dataset->addLogMessage(sprintf('Validation error in row %d: %s', $this->position, (string) $validation_error->getMessage()), FALSE);
      }
      $this->dataset->save();
      return FALSE;
    }

    try {
      $this->pipeline->transform($this->getInnerIterator()->current());
    }
    catch (TransformSkipRecordException $e) {
      $this->invalidDeltas[] = $delta;
      $this->dataset->addLogMessage(sprintf('Skip in row %d. %s', $this->position, $e->getMessage()), FALSE);
      $this->dataset->save();
      return FALSE;
    }

    return TRUE;
  }

  /**
   * Gets value of InvalidDeltas.
   *
   * @return int[]
   *   Value of InvalidDeltas.
   */
  public function getInvalidDeltas(): array {
    return $this->invalidDeltas;
  }

}