Improve index queue handling

config/pimcore/messenger.yaml

@@ -2,9 +2,11 @@ framework:
     messenger:
         enabled: true
         transports:
-            pimcore_index_queues: 'doctrine://default?queue_name=pimcore_index_queues'
+            pimcore_generic_data_index_queue: 'doctrine://default?queue_name=pimcore_index_queues'
+            pimcore_generic_data_index_sync: 'sync://'
         routing:
-            Pimcore\Bundle\GenericDataIndexBundle\Message\IndexUpdateQueueMessage: pimcore_index_queues
+            Pimcore\Bundle\GenericDataIndexBundle\Message\IndexUpdateQueueMessage: pimcore_generic_data_index_queue
+            Pimcore\Bundle\GenericDataIndexBundle\Message\DispatchQueueMessagesMessage: pimcore_generic_data_index_queue
         buses:
             messenger.bus.pimcore-generic-data-index:
                 middleware:

src/Command/Update/IndexUpdateCommand.php

@@ -83,11 +83,6 @@ protected function configure(): void
     }
 
     /**
-     * @param InputInterface $input
-     * @param OutputInterface $output
-     *
-     * @return int
-     *
      * @throws \Doctrine\DBAL\Exception
      * @throws RuntimeException
      */
@@ -159,7 +154,12 @@ protected function execute(InputInterface $input, OutputInterface $output): int
                 ->updateAll();
         }
 
-        $this->indexQueueService->dispatchQueueMessages($output);
+        $this->output->writeln(
+            '<info>Dispatch queue messages</info>',
+            OutputInterface::VERBOSITY_VERBOSE
+        );
+
+        $this->indexQueueService->dispatchQueueMessages(true);
 
         $this->release();
 

src/Enum/Messenger/TransportName.php

@@ -0,0 +1,10 @@
+<?php
+declare(strict_types=1);
+
+namespace Pimcore\Bundle\GenericDataIndexBundle\Enum\Messenger;
+
+enum TransportName: string
+{
+    case INDEX_QUEUE = 'pimcore_generic_data_index_queue';
+    case SYNC = 'pimcore_generic_data_index_sync';
+}

src/Enum/SearchIndex/IndexName.php

@@ -0,0 +1,10 @@
+<?php
+declare(strict_types=1);
+
+namespace Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex;
+
+enum IndexName: string
+{
+    case ASSET = 'asset';
+    case DATA_OBJECT = 'data-object';
+}

src/EventSubscriber/IndexUpdateSubscriber.php

@@ -14,6 +14,7 @@
 namespace Pimcore\Bundle\GenericDataIndexBundle\EventSubscriber;
 
 use Exception;
+use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexName;
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexQueueOperation;
 use Pimcore\Bundle\GenericDataIndexBundle\Installer;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueueService;
@@ -116,9 +117,10 @@ public function addDataObjectMapping(ClassDefinitionEvent $event): void
         }
 
         $classDefinition = $event->getClassDefinition();
+
         $this->dataObjectIndexService
             ->updateMapping($classDefinition, true)
-            ->addClassDefinitionToAlias($classDefinition, 'class_definitions');
+            ->addClassDefinitionToAlias($classDefinition, IndexName::DATA_OBJECT->value);
     }
 
     /**
@@ -131,10 +133,14 @@ public function updateDataObjectMapping(ClassDefinitionEvent $event): void
         }
 
         $classDefinition = $event->getClassDefinition();
+
         $this->dataObjectIndexService
             ->updateMapping($classDefinition)
-            ->addClassDefinitionToAlias($classDefinition, 'class_definitions');
-        $this->indexQueueService->updateDataObjects($classDefinition);
+            ->addClassDefinitionToAlias($classDefinition, IndexName::DATA_OBJECT->value);
+
+        $this->indexQueueService
+            ->updateDataObjects($classDefinition)
+            ->dispatchQueueMessages();
     }
 
     public function deleteDataObjectIndex(ClassDefinitionEvent $event): void
@@ -148,7 +154,7 @@ public function deleteDataObjectIndex(ClassDefinitionEvent $event): void
         try {
             $this->dataObjectIndexService
                 ->deleteIndex($classDefinition)
-                ->removeClassDefinitionFromAlias($classDefinition, 'class_definitions')
+                ->removeClassDefinitionFromAlias($classDefinition, IndexName::DATA_OBJECT->value)
             ;
         } catch (Exception $e) {
             $this->logger->error($e->getMessage());
@@ -195,7 +201,9 @@ public function deleteTag(TagEvent $event): void
             return;
         }
 
-        $this->indexQueueService->updateByTag($event->getTag());
+        $this->indexQueueService
+            ->updateByTag($event->getTag())
+            ->dispatchQueueMessages();
     }
 
     public function updateTagAssignment(TagEvent $event): void

src/Message/DispatchQueueMessagesMessage.php

@@ -0,0 +1,9 @@
+<?php
+declare(strict_types=1);
+
+namespace Pimcore\Bundle\GenericDataIndexBundle\Message;
+
+class DispatchQueueMessagesMessage
+{
+
+}

src/MessageHandler/DispatchQueueMessagesHandler.php

@@ -0,0 +1,47 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * Pimcore
+ *
+ * This source file is available under following license:
+ * - Pimcore Commercial License (PCL)
+ *
+ *  @copyright  Copyright (c) Pimcore GmbH (http://www.pimcore.org)
+ *  @license    http://www.pimcore.org/license     PCL
+ */
+
+namespace Pimcore\Bundle\GenericDataIndexBundle\MessageHandler;
+
+use Pimcore\Bundle\GenericDataIndexBundle\Message\DispatchQueueMessagesMessage;
+use Pimcore\Bundle\GenericDataIndexBundle\Message\IndexUpdateQueueMessage;
+use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueueService;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Symfony\Component\Messenger\MessageBusInterface;
+
+#[AsMessageHandler]
+class DispatchQueueMessagesHandler
+{
+    public function __construct(
+        protected readonly IndexQueueService $indexQueueService,
+        protected readonly MessageBusInterface $messageBus
+    ) {
+    }
+
+    public function __invoke(DispatchQueueMessagesMessage $message): void
+    {
+        $batchSize = 20;
+        while (true) {
+            $entries = $this->indexQueueService->getUnhandledIndexQueueEntries(true, $batchSize);
+            $amountOfEntries = count($entries);
+
+            if ($amountOfEntries > 0) {
+                $this->messageBus->dispatch(new IndexUpdateQueueMessage($entries));
+            }
+
+            if ($amountOfEntries < $batchSize) {
+                break;
+            }
+        }
+    }
+}

src/MessageHandler/IndexUpdateQueueHandler.php

@@ -16,15 +16,13 @@
 use Pimcore\Bundle\GenericDataIndexBundle\Message\IndexUpdateQueueMessage;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueueService;
 use Symfony\Component\Messenger\Attribute\AsMessageHandler;
-use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Serializer\Exception\ExceptionInterface;
 
 #[AsMessageHandler]
 class IndexUpdateQueueHandler
 {
     public function __construct(
-        protected readonly IndexQueueService $indexQueueService,
-        protected readonly MessageBusInterface $messageBus
+        protected readonly IndexQueueService $indexQueueService
     ) {
     }
 

src/Service/SearchIndex/IndexQueueService.php

@@ -18,8 +18,11 @@
 use Exception;
 use InvalidArgumentException;
 use Pimcore\Bundle\GenericDataIndexBundle\Entity\IndexQueue;
+use Pimcore\Bundle\GenericDataIndexBundle\Enum\Messenger\TransportName;
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\ElementType;
+use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexName;
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexQueueOperation;
+use Pimcore\Bundle\GenericDataIndexBundle\Message\DispatchQueueMessagesMessage;
 use Pimcore\Bundle\GenericDataIndexBundle\Message\IndexUpdateQueueMessage;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexService\AbstractIndexService;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexService\AssetIndexService;
@@ -35,9 +38,8 @@
 use Pimcore\Model\DataObject\Service;
 use Pimcore\Model\Element\ElementInterface;
 use Pimcore\Model\Element\Tag;
-use Symfony\Component\Console\Helper\ProgressBar;
-use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\Stamp\TransportNamesStamp;
 use Symfony\Component\Serializer\Exception\ExceptionInterface;
 use Symfony\Component\Serializer\Normalizer\DenormalizerInterface;
 use UnhandledMatchError;
@@ -248,7 +250,7 @@ public function updateAssets(): IndexQueueService
     {
         $selectQuery = sprintf("SELECT id, '%s', '%s', '%s', '%s', 0 FROM %s",
             ElementType::ASSET->value,
-            'asset',
+            IndexName::ASSET->value,
             IndexQueueOperation::UPDATE->value,
             $this->getCurrentQueueTableOperationTime(),
             'assets'
@@ -266,7 +268,7 @@ public function updateByTag(Tag $tag): IndexQueueService
         //assets
         $selectQuery = sprintf("SELECT id, '%s', '%s', '%s', '%s', 0 FROM assets where id in (select cid from tags_assignment where ctype='asset' and tagid = %s)",
             ElementType::ASSET->value,
-            'asset',
+            IndexName::ASSET->value,
             IndexQueueOperation::UPDATE->value,
             $this->getCurrentQueueTableOperationTime(),
             $tag->getId()
@@ -479,7 +481,7 @@ protected function getElementIndexName(ElementInterface $element): string
     {
         return match (true) {
             $element instanceof Concrete => $element->getClassName(),
-            $element instanceof Asset => 'asset',
+            $element instanceof Asset => IndexName::ASSET->value,
         };
     }
 
@@ -509,20 +511,16 @@ public function countQueuedItems(): int
         }
     }
 
-    public function dispatchQueueMessages(OutputInterface $output): void
+    public function dispatchQueueMessages(bool $synchronously = false): void
     {
-        $entries = $this->getUnhandledIndexQueueEntries(true);
+        $stamps = [];
 
-        $progressBar = new ProgressBar($output, count($entries));
-        $progressBar->start();
-
-        foreach(array_chunk($entries, 400) as $entriesBatch) {
-            $message = new IndexUpdateQueueMessage($entries);
-            $this->messageBus->dispatch($message);
-            $progressBar->advance(count($entriesBatch));
+        if ($synchronously) {
+            $stamps[] = new TransportNamesStamp(TransportName::SYNC->value);
         }
 
-        $progressBar->finish();
+        $message = new DispatchQueueMessagesMessage();
+        $this->messageBus->dispatch($message, $stamps);
     }
 
     public function commit(): IndexQueueService

src/Service/SearchIndex/IndexService/AssetIndexService.php

@@ -19,6 +19,7 @@
 use OpenSearch\Namespaces\IndicesNamespace;
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\FieldCategory;
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\FieldCategory\SystemField;
+use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexName;
 use Pimcore\Model\Asset;
 use Pimcore\Model\Element\ElementInterface;
 use Pimcore\Model\Element\Service;
@@ -42,7 +43,7 @@ protected function getIndexName(ElementInterface $element): string
 
     protected function getAssetIndexName(): string
     {
-        return $this->searchIndexConfigService->getIndexName('asset');
+        return $this->searchIndexConfigService->getIndexName(IndexName::ASSET->value);
     }
 
     protected function getCurrentFullIndexName(): string

src/Service/SearchIndex/IndexService/DataObjectIndexService.php

@@ -113,8 +113,6 @@ public function extractMapping(ClassDefinition $classDefinition): array
      *  - if that fails, create new index and reindex on ES side
      *  - if that also fails, throws exception
      *
-     * @return $this
-     *
      * @throws Exception
      */
     public function updateMapping(ClassDefinition $classDefinition, bool $forceCreateIndex = false): DataObjectIndexService