queue_stats-8.x-1.0-beta1/src/Plugin/QueueStatistic/ProcessingTime.php

<?php

namespace Drupal\queue_stats\Plugin\QueueStatistic;

use Drupal\Core\Datetime\DateFormatterInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\State\StateInterface;
use Drupal\queue_stats\Event\QueueItemEvent;
use Drupal\queue_stats\MonitoredQueueInterface;
use Drupal\queue_stats\Plugin\DateIntervalStatistic;
use Drupal\queue_stats\Plugin\QueueStatisticBase;
use Drupal\queue_stats\Plugin\StatefulStatistic;
use Drupal\queue_stats\Statistics\ExponentialMovingAverage;
use Symfony\Component\DependencyInjection\ContainerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;

/**
 * Calculate how long it takes to process each time.
 *
 * @QueueStatistic(
 *   id = "processing_time",
 *   label = "Average processing time"
 * )
 */
class ProcessingTime extends QueueStatisticBase implements EventSubscriberInterface, ContainerFactoryPluginInterface {
  use StatefulStatistic, DateIntervalStatistic;

  /**
   * Events for items which have started processing.
   *
   * @var \Drupal\queue_stats\Event\QueueItemEvent[]
   */
  protected $itemStarts;

  /**
   * ProcessingTime constructor.
   *
   * @param array $configuration
   *   The plugin configuration.
   * @param string $plugin_id
   *   The plugin id.
   * @param mixed $plugin_definition
   *   The plugin definition.
   * @param \Drupal\Core\State\StateInterface $state
   *   The site state.
   * @param \Drupal\Core\Datetime\DateFormatterInterface $date_formatter
   *   The date formatter.
   */
  public function __construct(
    array $configuration,
    string $plugin_id,
    $plugin_definition,
    StateInterface $state,
    DateFormatterInterface $date_formatter
  ) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
    $this->state = $state;
    $this->dateFormatter = $date_formatter;
  }

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

  /**
   * {@inheritdoc}
   */
  public static function getSubscribedEvents() {
    return [
      QueueItemEvent::PROCESSING_STARTED => 'trackStart',
      QueueItemEvent::PROCESSING_COMPLETED => 'trackCompletion'
    ];
  }

  /**
   * Event handler when events start processing.
   *
   * @param \Drupal\queue_stats\Event\QueueItemEvent $event
   *   The event.
   */
  public function trackStart(QueueItemEvent $event) {
    $this->itemStarts[$event->getItem()->item_id] = $event;
  }

  /**
   * Event handler when events have finished processing.
   *
   * @param \Drupal\queue_stats\Event\QueueItemEvent $event
   *   The event.
   */
  public function trackCompletion(QueueItemEvent $event) {
    if (!empty($this->itemStarts[$event->getItem()->item_id])) {
      $startEvent = $this->itemStarts[$event->getItem()->item_id];

      $processing_time = $event->getTimestamp() - $startEvent->getTimestamp();
      if ($processing_time > 0) {
        /** @var \Drupal\queue_stats\Statistics\ExponentialMovingAverage $value */
        $value = $this->retrieveValue($event->getQueue(), 'value', new ExponentialMovingAverage(10));
        $value->add($processing_time);
        $this->storeValue($event->getQueue(), 'value', $value);
      }
    }
  }

  /**
   * {@inheritdoc}
   */
  public function getValue(MonitoredQueueInterface $queue) {
    /** @var \Drupal\queue_stats\Statistics\ExponentialMovingAverage $average */
    $average = $this->retrieveValue($queue, 'value', new ExponentialMovingAverage(10));
    return $average->average();
  }

  /**
   * {@inheritdoc}
   */
  public function reset(MonitoredQueueInterface $queue) {
    $this->deleteValue($queue, 'value');
  }

}