Import.php

Go to the documentation of this file.

<?php
 
namespace Drupal\dkan_datastore\Plugin\QueueWorker;
 
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\Logger\RfcLogLevel;
use Procrastinator\Result;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;
 
class Import extends QueueWorkerBase implements ContainerFactoryPluginInterface {
 
  use \Drupal\Core\Logger\LoggerChannelTrait;
 
 
  private $container;
 
  public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition) {
    return new Import($configuration, $plugin_id, $plugin_definition, $container);
  }
 
  public function __construct(array $configuration, $plugin_id, $plugin_definition, ContainerInterface $container) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
    $this->container = $container;
  }
 
  public function processItem($data) {
 
    try {
      $datastore = $this->container->get('dkan_datastore.service');
 
      $results = $datastore->import($data['uuid']);
 
      foreach ($results as $result) {
        $this->processResult($result, $data);
      }
    }
    catch (\Exception $e) {
      $this->log(RfcLogLevel::ERROR,
        "Import for {$data['uuid']} returned an error: {$e->getMessage()}");
    }
  }
 
  private function processResult(Result $result, $data) {
    $level = RfcLogLevel::INFO;
    $message = "";
    $status = $result->getStatus();
    switch ($status) {
      case Result::STOPPED:
        $newQueueItemId = $this->requeue($data);
        $message = "Import for {$data['uuid']} is requeueing for iteration No. {$data['queue_iteration']}. (ID:{$newQueueItemId}).";
        break;
 
      case Result::IN_PROGRESS:
      case Result::ERROR:
        $level = RfcLogLevel::ERROR;
        $message = "Import for {$data['uuid']} returned an error: {$result->getError()}";
        break;
 
      case Result::DONE:
        $message = "Import for {$data['uuid']} completed.";
        break;
    }
    $this->log($level, $message);
  }
 
  protected function log($level, $message, array $context = []) {
    $this->getLogger($this->getPluginId())
      ->log($level, $message, $context);
  }
 
  protected function requeue(array $data) {
    return $this->container->get('queue')
      ->get($this->getPluginId())
      ->createItem($data);
  }
 
}