xero_sync-8.x-1.x-dev/src/Plugin/QueueWorker/WorkerBase.php

<?php

namespace Drupal\xero_sync\Plugin\QueueWorker;

use Drupal\Component\Datetime\TimeInterface;
use Drupal\Core\Entity\EntityTypeBundleInfoInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\Queue\SuspendQueueException;
use Drupal\Core\TypedData\TypedDataManagerInterface;
use Drupal\xero_sync\XeroSyncEntityHandlerInterface;
use Drupal\xero_sync\XeroSyncJobManagerInterface;
use Drupal\xero_sync\XeroSyncQueueJobTrait;
use Drupal\xero_sync\XeroSyncSynchronizerInterface;
use Psr\Log\LoggerInterface;
use Drupal\Core\Logger\RfcLogLevel;
use Symfony\Component\Serializer\SerializerInterface;

/**
 * A base class for Xero Sync queue workers.
 */
class WorkerBase extends QueueWorkerBase implements ContainerFactoryPluginInterface {

  use XeroSyncQueueJobTrait;

  /**
   * Respect Xero rate limits by waiting until this time before processing.
   *
   * @var int
   */
  protected static $waitUntil = 0;

  /**
   * Construct a Xero Sync queue worker plugin.
   *
   * This method is not in XeroSyncQueueJobTrait, in order to facilitate unit
   * testing that trait in tests/src/Unit/QueueJobTest.php.
   *
   * @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 \Psr\Log\LoggerInterface $logger
   *   A logger instance.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entity_type_manager
   *   The entity type manager.
   * @param \Drupal\Core\Entity\EntityTypeBundleInfoInterface $entity_type_bundle_info
   *   The entity type bundle manager.
   * @param \Drupal\xero_sync\XeroSyncSynchronizerInterface $synchronizer
   *   The synchronizer.
   * @param \Drupal\Core\TypedData\TypedDataManagerInterface $typed_data_manager
   *   The typed data manager service.
   * @param \Drupal\xero_sync\XeroSyncJobManagerInterface $job_manager
   *   The job manager.
   * @param \Symfony\Component\Serializer\SerializerInterface $serializer
   *   The serializer.
   * @param \Drupal\Component\Datetime\TimeInterface $time
   *   The time service.
   * @param \Drupal\xero_sync\XeroSyncEntityHandlerInterface $entity_handler
   *   The entity handler.
   */
  public function __construct(array $configuration, $plugin_id, $plugin_definition, LoggerInterface $logger, EntityTypeManagerInterface $entity_type_manager, EntityTypeBundleInfoInterface $entity_type_bundle_info, XeroSyncSynchronizerInterface $synchronizer, TypedDataManagerInterface $typed_data_manager, XeroSyncJobManagerInterface $job_manager, SerializerInterface $serializer, TimeInterface $time, XeroSyncEntityHandlerInterface $entity_handler) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
    $this->setProperties($logger, $entity_type_manager, $entity_type_bundle_info, $synchronizer, $typed_data_manager, $job_manager, $serializer, $time, $entity_handler);
  }

  /**
   * {@inheritdoc}
   */
  public function processItem($data) {
    try {
      $wait = self::$waitUntil - $this->time->getCurrentTime();
      if ($wait > 60) {
        throw new SuspendQueueException("Xero API rate limit exceeded, Xero Sync queue processing paused.");
      }
      if ($wait > 0) {
        // There's no point to throwing a DelayedRequeueException, as subsequent
        // items will only fail in this same way. Better to simply wait.
        sleep($wait);
      }

      $message = $this->doProcessItem($data);
      $this->logger->log(RfcLogLevel::INFO, $message);
    }
    catch (\Exception $e) {
      // Nothing to do, error logging already happens in ::doProcessItem.
    }
  }

}