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.             dispatched bigint NULL,
  68.             workerId varchar(13) NULL,
  69.             PRIMARY KEY (elementId, configName, entityType),
  70.             KEY `bundle_index_queue_configName_workerId` (`workerId`),
  71.             KEY `bundle_index_queue_configName_index` (`configName`))
  72.         ', self::QUEUE_TABLE_NAME));
  74.         if ($callable) {
  75.             return $callable();
  76.         }
  77.     }
  79.     public function getAllQueueEntries($limit = 100000, bool $dispatch = false): array
  80.     {
  81.         try {
  82.             if ($dispatch === true) {
  83.                 $dispatchId = time();
  84.                 $workerId = uniqid();
  86.                 $this->getDb()->executeQuery('UPDATE ' . self::QUEUE_TABLE_NAME . ' SET dispatched = ?, workerId = ? WHERE (ISNULL(dispatched) OR dispatched < ?) LIMIT ' . intval($limit),
  87.                     [$dispatchId, $workerId, $dispatchId - 3000]);
  89.                 $results = $this->getDb()->fetchAll(
  90.                     sprintf('SELECT * FROM %s WHERE workerId = ?', self::QUEUE_TABLE_NAME),
  91.                     [$workerId]
  92.                 );
  93.             } else {
  94.                 $results = $this->getDb()->fetchAll(
  95.                     sprintf('SELECT * FROM %s', self::QUEUE_TABLE_NAME)
  96.                 );
  97.             }
  99.             return $results ?? [];
  100.         } catch (TableNotFoundException $exception) {
  101.             return $this->createQueueTableIfNotExisting(function () use ($limit) {
  102.                 return $this->getAllQueueEntries($limit);
  103.             });
  104.         }
  105.     }
  107.     /**
  108.      * @return int
  109.      */
  110.     public function getQueueItemCount(): int
  111.     {
  112.         try {
  113.             return $this->getDb()->fetchOne(
  114.                 sprintf('SELECT count(*) as count FROM %s', self::QUEUE_TABLE_NAME)
  115.             ) ?? 0;
  116.         } catch (TableNotFoundException $exception) {
  117.             return $this->createQueueTableIfNotExisting(function () {
  118.                 return $this->getQueueItemCount();
  119.             });
  120.         }
  121.     }
  123.     public function markQueueEntryAsProcessed($elementId, $entityType, $configName)
  124.     {
  125.         try {
  126.             $this->getDb()->executeQuery(
  127.                 sprintf('DELETE FROM %s WHERE elementId = ? AND entityType = ? AND configName = ?', self::QUEUE_TABLE_NAME),
  128.                 [$elementId, $entityType, $configName ?? '']
  129.             );
  130.         } catch (TableNotFoundException $exception) {
  131.             $this->createQueueTableIfNotExisting();
  132.         }
  133.     }
  134. }