xero_sync-8.x-1.x-dev/src/Plugin/AdvancedQueue/JobType/JobBase.php

<?php

namespace Drupal\xero_sync\Plugin\AdvancedQueue\JobType;

use Drupal\advancedqueue\Job;
use Drupal\advancedqueue\JobResult;
use Drupal\advancedqueue\Plugin\AdvancedQueue\JobType\JobTypeBase;
use Drupal\Core\Entity\EntityTypeBundleInfoInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\TypedData\TypedDataManagerInterface;
use Drupal\Component\Datetime\TimeInterface;
use Drupal\xero_sync\Exception\SyncRetryException;
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 advanced queue job types.
 */
class JobBase extends JobTypeBase 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 advanced queue job type 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 process(Job $job) {
    try {
      $wait = self::$waitUntil - $this->time->getCurrentTime();
      // Advanced queue only allows 90 seconds of processing time per queue run
      // when not using cli, and if a rate limit is hit this suggests we will
      // already have run for 60 seconds by the time we resume, so it's
      // not really worth waiting, better to fail on any further job
      // in this queue run.
      if ($wait > 0 && $wait < 61 && PHP_SAPI != 'cli') {
        sleep($wait);
      }
      elseif ($wait > 0) {
        throw new SyncRetryException($wait);
      }

      $data = $job->getPayload();
      $message = $this->doProcessItem($data);
      $this->logger->log(RfcLogLevel::INFO, $message);
      return JobResult::success($message);
    }
    catch (SyncRetryException $e) {
      // Assume the need for a retry comes from rate limit, so is
      // not the responsibility of this job.
      return JobResult::failure($e->getMessage(), $this->getMaxRetries() + 1, $e->getDelay());
    }
    catch (\Throwable $e) {
      // Nothing to do, failure logging already happens in ::doProcessItem.
      return JobResult::failure($e->getMessage());
    }
  }

}