vendor/pimcore/data-hub-simple-rest/src/Queue/QueueService.php line 39

Open in your IDE?
  1. <?php
  3. /**
  4.  * Pimcore
  5.  *
  6.  * This source file is available under following license:
  7.  * - Pimcore Commercial License (PCL)
  8.  *
  9.  *  @copyright  Copyright (c) Pimcore GmbH (http://www.pimcore.org)
  10.  *  @license    http://www.pimcore.org/license     PCL
  11.  */
  13. namespace Pimcore\Bundle\DataHubSimpleRestBundle\Queue;
  15. use Carbon\Carbon;
  16. use Doctrine\DBAL\Exception\TableNotFoundException;
  17. use Pimcore\Db;
  19. class QueueService
  20. {
  21.     const QUEUE_TABLE_NAME = 'bundle_data_hub_rest_index_queue';
  23.     /**
  24.      * @return Db\Connection|Db\ConnectionInterface
  25.      */
  26.     protected function getDb()
  27.     {
  28.         return Db::get();
  29.     }
  31.     protected function getCurrentQueueTableOperationTime(): int
  32.     {
  33.         /** @var Carbon $carbonNow */
  34.         $carbonNow = Carbon::now();
  36.         return (int)($carbonNow->getTimestamp() . str_pad((string)$carbonNow->milli, 3, '0'));
  37.     }
  39.     public function addItemToQueue($elementId, string $entityType, string $configName = null)
  40.     {
  41.         try {
  42.             $this->getDb()->executeQuery(sprintf(
  43.                 'INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE timestamp = VALUES(timestamp)',
  44.                 self::QUEUE_TABLE_NAME,
  45.                 implode(',', ['elementId', 'timestamp', 'configName', 'entityType']),
  46.                 implode(',', [
  47.                     $elementId,
  48.                     $this->getCurrentQueueTableOperationTime(),
  49.                     $this->getDb()->quote($configName),
  50.                     $this->getDb()->quote($entityType)
  51.                 ])
  52.             ));
  53.         } catch (TableNotFoundException $exception) {
  54.             $this->createQueueTableIfNotExisting(function () use ($elementId, $entityType, $configName) {
  55.                 $this->addItemToQueue($elementId, $entityType, $configName);
  56.             });
  57.         }
  58.     }
  60.     protected function createQueueTableIfNotExisting(\Closure $callable = null)
  61.     {
  62.         $this->getDb()->executeQuery(sprintf('CREATE TABLE IF NOT EXISTS %s (
  63.             elementId bigint NOT NULL,
  64.             timestamp bigint NULL,
  65.             configName varchar(50) NOT NULL,
  66.             entityType varchar(50) NOT NULL,
  67.             PRIMARY KEY (elementId, configName, entityType),
  68.             KEY `bundle_index_queue_configName_index` (`configName`))
  69.         ', self::QUEUE_TABLE_NAME));
  71.         if ($callable) {
  72.             return $callable();
  73.         }
  74.     }
  76.     public function getAllQueueEntries($limit = 100000): array
  77.     {
  78.         try {
  79.             $results = $this->getDb()->fetchAll(
  80.                 sprintf('SELECT * FROM %s', self::QUEUE_TABLE_NAME)
  81.             );
  83.             return $results ?? [];
  84.         } catch (TableNotFoundException $exception) {
  85.             return $this->createQueueTableIfNotExisting(function () use ($limit) {
  86.                 return $this->getAllQueueEntries($limit);
  87.             });
  88.         }
  89.     }
  91.     /**
  92.      * @return int
  93.      */
  94.     public function getQueueItemCount(): int
  95.     {
  96.         try {
  97.             return $this->getDb()->fetchOne(
  98.                 sprintf('SELECT count(*) as count FROM %s', self::QUEUE_TABLE_NAME)
  99.             ) ?? 0;
  100.         } catch (TableNotFoundException $exception) {
  101.             return $this->createQueueTableIfNotExisting(function () {
  102.                 return $this->getQueueItemCount();
  103.             });
  104.         }
  105.     }
  107.     public function markQueueEntryAsProcessed($elementId, $entityType, $configName)
  108.     {
  109.         try {
  110.             $this->getDb()->executeQuery(
  111.                 sprintf('DELETE FROM %s WHERE elementId = ? AND entityType = ? AND configName = ?', self::QUEUE_TABLE_NAME),
  112.                 [$elementId, $entityType, $configName]
  113.             );
  114.         } catch (TableNotFoundException $exception) {
  115.             $this->createQueueTableIfNotExisting();
  116.         }
  117.     }
  118. }