data_pipelines-1.x-dev/src/Plugin/QueueWorker/DestinationWorker.php

<?php

declare(strict_types=1);

namespace Drupal\data_pipelines\Plugin\QueueWorker;

use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Queue\Attribute\QueueWorker;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\data_pipelines\Plugin\Derivative\DatasetQueueWorkerDeriver;
use Drupal\Tests\data_pipelines\Exception\DestinationNotFoundException;
use Drupal\Tests\data_pipelines\Exception\DestinationPluginNotFoundException;
use Drupal\data_pipelines\Destination\DatasetDestinationPluginManager;
use Drupal\data_pipelines\Destination\ProcessingOperation;
use Drupal\data_pipelines\Entity\DatasetInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Defines a queue worker for saving or deleting from a destination.
 */
#[QueueWorker(
  id: 'data_pipelines_process',
  title: new TranslatableMarkup('Dataset Destination queue worker'),
  deriver: DatasetQueueWorkerDeriver::class,
)]
final class DestinationWorker extends QueueWorkerBase implements ContainerFactoryPluginInterface {

  /**
   * Entity type manager.
   *
   * @var \Drupal\Core\Entity\EntityTypeManagerInterface
   */
  protected $entityTypeManager;

  /**
   * The destination plugin manager.
   *
   * @var \Drupal\data_pipelines\Destination\DatasetDestinationPluginManager
   */
  protected $destinationPluginManager;

  /**
   * The logger.
   *
   * @var \Psr\Log\LoggerInterface
   */
  protected $logger;

  /**
   * Constructs a \Drupal\Component\Plugin\PluginBase object.
   *
   * @param array $configuration
   *   A configuration array containing information about the plugin instance.
   * @param string $plugin_id
   *   The plugin_id for the plugin instance.
   * @param mixed $plugin_definition
   *   The plugin implementation definition.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\data_pipelines\Destination\DatasetDestinationPluginManager $destinationPluginManager
   *   The destination plugin manager.
   * @param \Psr\Log\LoggerInterface $logger
   *   The logger.
   */
  public function __construct(array $configuration, $plugin_id, $plugin_definition, EntityTypeManagerInterface $entityTypeManager, DatasetDestinationPluginManager $destinationPluginManager, LoggerInterface $logger) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
    $this->entityTypeManager = $entityTypeManager;
    $this->destinationPluginManager = $destinationPluginManager;
    $this->logger = $logger;
  }

  /**
   * {@inheritdoc}
   */
  public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition) {
    return new static(
      $configuration,
      $plugin_id,
      $plugin_definition,
      $container->get('entity_type.manager'),
      $container->get('plugin.manager.data_pipelines_destination'),
      $container->get('logger.channel.data_pipelines')
    );
  }

  /**
   * {@inheritdoc}
   */
  public function processItem($data): void {
    if (!$data instanceof ProcessingOperation) {
      $this->logger->error("Invalid queue data of type %type", ['%type' => get_class($data)]);
      return;
    }
    $pipeline_id = $this->pluginDefinition['data_pipelines_id'];
    $dataset = $this->entityTypeManager
      ->getStorage('data_pipelines')
      ->load($pipeline_id);

    if (!$dataset instanceof DatasetInterface) {
      $this->logger->warning(sprintf("Failed to load dataset for pipeline %s", $pipeline_id));
      return;
    }

    /** @var \Drupal\data_pipelines\Entity\DestinationInterface|null $destination */
    $destination = $this->entityTypeManager
      ->getStorage('dataset_destination')
      ->load($data->getDestinationId());
    if (!$destination) {
      $this->logger->error(sprintf("Destination not found for dataset %s", $dataset->id()));
      throw new DestinationNotFoundException(sprintf("Destination not found for dataset %s", $dataset->id()));
    }

    $destinationPlugin = $destination->getDestinationPlugin();
    if (!$destinationPlugin) {
      $this->logger->error(sprintf("Destination plugin not provided for destination %s", $destination->id()));
      throw new DestinationPluginNotFoundException(sprintf("Destination plugin not provided for destination %s", $destination->id()));
    }

    $destinationPlugin->mergeConfiguration($dataset->getPipeline()->getDestinationSettings($destination->getDestinationPluginId()));

    $result = $data->getOperation()->process($destinationPlugin, $dataset, $data);
    if (!$result) {
      $this->logger->error(sprintf("Destination plugin processing error during %s operation for %s plugin", $data->getOperation()->id(), $destination->id()));
    }
  }

}