You are here

Home » API reference » Advanced Queue 8 » Database.php

class Database in Advanced Queue 8

Provides the database queue backend.

Plugin annotation


@AdvancedQueueBackend(
  id = "database",
  label = @Translation("Database"),
)

Hierarchy

Expanded class hierarchy of Database

File

src/Plugin/AdvancedQueue/Backend/Database.php, line 18

Namespace

Drupal\advancedqueue\Plugin\AdvancedQueue\Backend
View source 
class Database extends BackendBase implements SupportsDeletingJobsInterface, SupportsListingJobsInterface, SupportsReleasingJobsInterface {

  /**
   * The database connection.
   *
   * @var \Drupal\Core\Database\Connection
   */
  protected $connection;

  /**
   * Constructs a new Database 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\Component\Datetime\TimeInterface $time
   *   The time.
   * @param \Drupal\Core\Database\Connection $connection
   *   The database connection to use.
   */
  public function __construct(array $configuration, $plugin_id, $plugin_definition, TimeInterface $time, Connection $connection) {
    parent::__construct($configuration, $plugin_id, $plugin_definition, $time);
    $this->connection = $connection;
  }

  /**
   * {@inheritdoc}
   */
  public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition) {
    return new static($configuration, $plugin_id, $plugin_definition, $container
      ->get('datetime.time'), $container
      ->get('database'));
  }

  /**
   * {@inheritdoc}
   */
  public function createQueue() {

    // No need to do anything, all database queues share the same table.
  }

  /**
   * {@inheritdoc}
   */
  public function deleteQueue() {

    // Delete all jobs in the current queue.
    $this->connection
      ->delete('advancedqueue')
      ->condition('queue_id', $this->queueId)
      ->execute();
  }

  /**
   * {@inheritdoc}
   */
  public function cleanupQueue() {

    // Reset expired jobs.
    $this->connection
      ->update('advancedqueue')
      ->fields([
      'state' => Job::STATE_QUEUED,
      'expires' => 0,
    ])
      ->condition('expires', 0, '<>')
      ->condition('expires', $this->time
      ->getCurrentTime(), '<')
      ->execute();
  }

  /**
   * {@inheritdoc}
   */
  public function countJobs() {

    // Ensure each state gets a count, even if it's 0.
    $jobs = [
      Job::STATE_QUEUED => 0,
      Job::STATE_PROCESSING => 0,
      Job::STATE_SUCCESS => 0,
      Job::STATE_FAILURE => 0,
    ];
    $query = 'SELECT state, COUNT(job_id) FROM {advancedqueue} WHERE queue_id = :queue_id GROUP BY state';
    $counts = $this->connection
      ->query($query, [
      ':queue_id' => $this->queueId,
    ])
      ->fetchAllKeyed();
    foreach ($counts as $state => $count) {
      $jobs[$state] = $count;
    }
    return $jobs;
  }

  /**
   * {@inheritdoc}
   */
  public function enqueueJob(Job $job, $delay = 0) {
    $this
      ->enqueueJobs([
      $job,
    ], $delay);
  }

  /**
   * {@inheritdoc}
   */
  public function enqueueJobs(array $jobs, $delay = 0) {
    if (count($jobs) > 1) {

      // Make the inserts atomic, and improve performance on certain engines.
      $this->connection
        ->startTransaction();
    }

    /** @var \Drupal\advancedqueue\Job $job */
    foreach ($jobs as $job) {
      $job
        ->setQueueId($this->queueId);
      $job
        ->setState(Job::STATE_QUEUED);
      if (!$job
        ->getAvailableTime()) {
        $job
          ->setAvailableTime($this->time
          ->getCurrentTime() + $delay);
      }
      $fields = $job
        ->toArray();
      unset($fields['id']);
      $fields['payload'] = json_encode($fields['payload']);

      // InsertQuery supports inserting multiple rows at once, which is faster,
      // but that doesn't give us the inserted job IDs.
      $query = $this->connection
        ->insert('advancedqueue')
        ->fields($fields);
      $job_id = $query
        ->execute();
      $job
        ->setId($job_id);
    }
  }

  /**
   * {@inheritdoc}
   */
  public function retryJob(Job $job, $delay = 0) {
    if ($job
      ->getState() != Job::STATE_FAILURE) {
      throw new \InvalidArgumentException('Only failed jobs can be retried.');
    }
    $job
      ->setNumRetries($job
      ->getNumRetries() + 1);
    $job
      ->setAvailableTime($this->time
      ->getCurrentTime() + $delay);
    $job
      ->setState(Job::STATE_QUEUED);
    $this
      ->updateJob($job);
  }

  /**
   * {@inheritdoc}
   */
  public function claimJob() {

    // Claim a job by updating its expire fields. If the claim is not successful
    // another thread may have claimed the job in the meantime. Therefore loop
    // until a job is successfully claimed or we are reasonably sure there
    // are no unclaimed jobs left.
    while (TRUE) {
      $query = 'SELECT * FROM {advancedqueue}
        WHERE queue_id = :queue_id AND state = :state AND available <= :now AND expires = 0
        ORDER BY available, job_id ASC';
      $params = [
        ':queue_id' => $this->queueId,
        ':state' => Job::STATE_QUEUED,
        ':now' => $this->time
          ->getCurrentTime(),
      ];
      $job_definition = $this->connection
        ->queryRange($query, 0, 1, $params)
        ->fetchAssoc();
      if (!$job_definition) {

        // No jobs left to claim.
        return NULL;
      }

      // Try to update the item. Only one thread can succeed in updating the
      // same row. We cannot rely on the request time because items might be
      // claimed by a single consumer which runs longer than 1 second. If we
      // continue to use request time instead of current time, we steal
      // time from the lease, and will tend to reset items before the lease
      // should really expire.
      $state = Job::STATE_PROCESSING;
      $expires = $this->time
        ->getCurrentTime() + $this->configuration['lease_time'];
      $update = $this->connection
        ->update('advancedqueue')
        ->fields([
        'state' => $state,
        'expires' => $expires,
      ])
        ->condition('job_id', $job_definition['job_id'])
        ->condition('expires', 0);

      // If there are affected rows, the claim succeeded.
      if ($update
        ->execute()) {
        $job_definition['id'] = $job_definition['job_id'];
        unset($job_definition['job_id']);
        $job_definition['payload'] = json_decode($job_definition['payload'], TRUE);
        $job_definition['state'] = $state;
        $job_definition['expires'] = $expires;
        return new Job($job_definition);
      }
    }
  }

  /**
   * {@inheritdoc}
   */
  public function onSuccess(Job $job) {
    $job
      ->setProcessedTime($this->time
      ->getCurrentTime());
    $this
      ->updateJob($job);
  }

  /**
   * {@inheritdoc}
   */
  public function onFailure(Job $job) {
    $job
      ->setProcessedTime($this->time
      ->getCurrentTime());
    $this
      ->updateJob($job);
  }

  /**
   * {@inheritdoc}
   */
  public function releaseJob($job_id) {
    $this->connection
      ->update('advancedqueue')
      ->fields([
      'state' => Job::STATE_QUEUED,
      'expires' => 0,
    ])
      ->condition('job_id', $job_id)
      ->execute();
  }

  /**
   * {@inheritdoc}
   */
  public function deleteJob($job_id) {
    $this->connection
      ->delete('advancedqueue')
      ->condition('job_id', $job_id)
      ->execute();
  }

  /**
   * Updates the given job.
   *
   * @param \Drupal\advancedqueue\Job $job
   *   The job.
   */
  protected function updateJob(Job $job) {
    $this->connection
      ->update('advancedqueue')
      ->fields([
      'payload' => json_encode($job
        ->getPayload()),
      'state' => $job
        ->getState(),
      'message' => $job
        ->getMessage(),
      'num_retries' => $job
        ->getNumRetries(),
      'available' => $job
        ->getAvailableTime(),
      'processed' => $job
        ->getProcessedTime(),
      'expires' => $job
        ->getExpiresTime(),
    ])
      ->condition('job_id', $job
      ->getId())
      ->execute();
  }

}

Members

Namesort descending Modifiers Type Description Overrides
BackendBase::$queueId protected property The current queue ID.
BackendBase::$time protected property The time.
BackendBase::buildConfigurationForm public function Form constructor. Overrides PluginFormInterface::buildConfigurationForm
BackendBase::calculateDependencies public function
BackendBase::defaultConfiguration public function Gets default configuration for this plugin. Overrides ConfigurableInterface::defaultConfiguration
BackendBase::getConfiguration public function Gets this plugin's configuration. Overrides ConfigurableInterface::getConfiguration
BackendBase::getLabel public function Gets the backend label. Overrides BackendInterface::getLabel
BackendBase::setConfiguration public function Sets the configuration for this plugin instance. Overrides ConfigurableInterface::setConfiguration
BackendBase::submitConfigurationForm public function Form submission handler. Overrides PluginFormInterface::submitConfigurationForm
BackendBase::validateConfigurationForm public function Form validation handler. Overrides PluginFormInterface::validateConfigurationForm
Database::$connection protected property The database connection.
Database::claimJob public function Claims the next available job for processing. Overrides BackendInterface::claimJob
Database::cleanupQueue public function Cleans up the queue. Overrides BackendBase::cleanupQueue
Database::countJobs public function Gets an estimated number of jobs in the queue. Overrides BackendInterface::countJobs
Database::create public static function Creates an instance of the plugin. Overrides BackendBase::create
Database::createQueue public function Creates the queue. Overrides BackendInterface::createQueue
Database::deleteJob public function Deletes the job with the given ID. Overrides SupportsDeletingJobsInterface::deleteJob
Database::deleteQueue public function Deletes the queue. Overrides BackendInterface::deleteQueue
Database::enqueueJob public function Enqueues the given job. Overrides BackendInterface::enqueueJob
Database::enqueueJobs public function Enqueues the given jobs. Overrides BackendInterface::enqueueJobs
Database::onFailure public function Called when job processing has failed. Overrides BackendInterface::onFailure
Database::onSuccess public function Called when a job has been successfully processed. Overrides BackendInterface::onSuccess
Database::releaseJob public function Releases the job with the given ID. Overrides SupportsReleasingJobsInterface::releaseJob
Database::retryJob public function Retries the given job. Overrides BackendInterface::retryJob
Database::updateJob protected function Updates the given job.
Database::__construct public function Constructs a new Database object. Overrides BackendBase::__construct
DependencySerializationTrait::$_entityStorages protected property An array of entity type IDs keyed by the property name of their storages.
DependencySerializationTrait::$_serviceIds protected property An array of service IDs keyed by property name used for serialization.
DependencySerializationTrait::__sleep public function 1
DependencySerializationTrait::__wakeup public function 2
MessengerTrait::$messenger protected property The messenger. 29
MessengerTrait::messenger public function Gets the messenger. 29
MessengerTrait::setMessenger public function Sets the messenger.
PluginBase::$configuration protected property Configuration information passed into the plugin. 1
PluginBase::$pluginDefinition protected property The plugin implementation definition. 1
PluginBase::$pluginId protected property The plugin_id.
PluginBase::DERIVATIVE_SEPARATOR constant A string which is used to separate base plugin IDs from the derivative ID.
PluginBase::getBaseId public function Gets the base_plugin_id of the plugin instance. Overrides DerivativeInspectionInterface::getBaseId
PluginBase::getDerivativeId public function Gets the derivative_id of the plugin instance. Overrides DerivativeInspectionInterface::getDerivativeId
PluginBase::getPluginDefinition public function Gets the definition of the plugin implementation. Overrides PluginInspectionInterface::getPluginDefinition 3
PluginBase::getPluginId public function Gets the plugin_id of the plugin instance. Overrides PluginInspectionInterface::getPluginId
PluginBase::isConfigurable public function Determines if the plugin is configurable.
StringTranslationTrait::$stringTranslation protected property The string translation service. 1
StringTranslationTrait::formatPlural protected function Formats a string containing a count of items.
StringTranslationTrait::getNumberOfPlurals protected function Returns the number of plurals supported by a given language.
StringTranslationTrait::getStringTranslation protected function Gets the string translation service.
StringTranslationTrait::setStringTranslation public function Sets the string translation service to use. 2
StringTranslationTrait::t protected function Translates a string to the current language or to a given language.