vendor/shopware/core/Framework/DataAbstractionLayer/Indexing/EntityIndexerRegistry.php line 96

Open in your IDE?
  1. <?php declare(strict_types=1);
  2. namespace Shopware\Core\Framework\DataAbstractionLayer\Indexing;
  3. use Shopware\Core\Framework\Context;
  4. use Shopware\Core\Framework\DataAbstractionLayer\Event\EntityWrittenContainerEvent;
  5. use Shopware\Core\Framework\DataAbstractionLayer\Indexing\MessageQueue\IterateEntityIndexerMessage;
  6. use Shopware\Core\Framework\Event\ProgressAdvancedEvent;
  7. use Shopware\Core\Framework\Event\ProgressFinishedEvent;
  8. use Shopware\Core\Framework\Event\ProgressStartedEvent;
  9. use Shopware\Core\Framework\MessageQueue\Handler\AbstractMessageHandler;
  10. use Shopware\Core\Framework\Struct\ArrayStruct;
  11. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  12. use Symfony\Component\Messenger\MessageBusInterface;
  13. use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
  14. class EntityIndexerRegistry extends AbstractMessageHandler implements EventSubscriberInterface
  15. {
  16.     public const EXTENSION_INDEXER_SKIP 'indexer-skip';
  17.     /**
  18.      * @deprecated tag:v6.5.0 - `$context->addExtension(EntityIndexerRegistry::USE_INDEXING_QUEUE, ...)` will be ignored, use `context->addState(EntityIndexerRegistry::USE_INDEXING_QUEUE)` instead
  19.      */
  20.     public const USE_INDEXING_QUEUE 'use-queue-indexing';
  21.     /**
  22.      * @deprecated tag:v6.5.0 - `$context->addExtension(EntityIndexerRegistry::DISABLE_INDEXING, ...)` will be ignored, use `context->addState(EntityIndexerRegistry::DISABLE_INDEXING)` instead
  23.      */
  24.     public const DISABLE_INDEXING 'disable-indexing';
  25.     /**
  26.      * @var EntityIndexer[]
  27.      */
  28.     private iterable $indexer;
  29.     private MessageBusInterface $messageBus;
  30.     private bool $working false;
  31.     private EventDispatcherInterface $dispatcher;
  32.     public function __construct(iterable $indexerMessageBusInterface $messageBusEventDispatcherInterface $dispatcher)
  33.     {
  34.         $this->indexer $indexer;
  35.         $this->messageBus $messageBus;
  36.         $this->dispatcher $dispatcher;
  37.     }
  38.     public static function getSubscribedEvents(): array
  39.     {
  40.         return [
  41.             EntityWrittenContainerEvent::class => [
  42.                 ['refresh'1000],
  43.             ],
  44.         ];
  45.     }
  46.     public static function getHandledMessages(): iterable
  47.     {
  48.         return [
  49.             EntityIndexingMessage::class,
  50.             IterateEntityIndexerMessage::class,
  51.         ];
  52.     }
  53.     public function index(bool $useQueue, array $skip = []): void
  54.     {
  55.         foreach ($this->indexer as $indexer) {
  56.             if (\in_array($indexer->getName(), $skiptrue)) {
  57.                 continue;
  58.             }
  59.             $offset null;
  60.             $this->dispatcher->dispatch(new ProgressStartedEvent($indexer->getName(), $indexer->getTotal()));
  61.             while ($message $indexer->iterate($offset)) {
  62.                 $message->setIndexer($indexer->getName());
  63.                 $message->addSkip(...$skip);
  64.                 $this->sendOrHandle($message$useQueue);
  65.                 $offset $message->getOffset();
  66.                 try {
  67.                     $count = \is_array($message->getData()) ? \count($message->getData()) : 1;
  68.                     $this->dispatcher->dispatch(new ProgressAdvancedEvent($count));
  69.                 } catch (\Exception $e) {
  70.                 }
  71.             }
  72.             $this->dispatcher->dispatch(new ProgressFinishedEvent($indexer->getName()));
  73.         }
  74.     }
  75.     public function refresh(EntityWrittenContainerEvent $event): void
  76.     {
  77.         $context $event->getContext();
  78.         if ($this->working) {
  79.             return;
  80.         }
  81.         $this->working true;
  82.         if ($this->disabled($context)) {
  83.             $this->working false;
  84.             return;
  85.         }
  86.         $useQueue $this->useQueue($context);
  87.         foreach ($this->indexer as $indexer) {
  88.             $message $indexer->update($event);
  89.             if (!$message) {
  90.                 continue;
  91.             }
  92.             $message->setIndexer($indexer->getName());
  93.             self::addSkips($message$context);
  94.             $this->sendOrHandle($message$useQueue);
  95.         }
  96.         $this->working false;
  97.     }
  98.     public static function addSkips(EntityIndexingMessage $messageContext $context): void
  99.     {
  100.         if (!$context->hasExtension(self::EXTENSION_INDEXER_SKIP)) {
  101.             return;
  102.         }
  103.         /** @var ArrayStruct $skip */
  104.         $skip $context->getExtension(self::EXTENSION_INDEXER_SKIP);
  105.         $message->addSkip(...$skip->all());
  106.     }
  107.     public function handle($message): void
  108.     {
  109.         if ($message instanceof EntityIndexingMessage) {
  110.             $indexer $this->getIndexer($message->getIndexer());
  111.             if ($indexer) {
  112.                 $indexer->handle($message);
  113.             }
  114.             return;
  115.         }
  116.         if ($message instanceof IterateEntityIndexerMessage) {
  117.             $next $this->iterateIndexer($message->getIndexer(), $message->getOffset(), true$message->getSkip());
  118.             if (!$next) {
  119.                 return;
  120.             }
  121.             $this->messageBus->dispatch(new IterateEntityIndexerMessage($message->getIndexer(), $next->getOffset(), $message->getSkip()));
  122.         }
  123.     }
  124.     public function sendIndexingMessage(array $indexer = [], array $skip = []): void
  125.     {
  126.         if (empty($indexer)) {
  127.             $indexer = [];
  128.             foreach ($this->indexer as $loop) {
  129.                 $indexer[] = $loop->getName();
  130.             }
  131.         }
  132.         if (empty($indexer)) {
  133.             return;
  134.         }
  135.         foreach ($indexer as $name) {
  136.             if (\in_array($name$skiptrue)) {
  137.                 continue;
  138.             }
  139.             $this->messageBus->dispatch(new IterateEntityIndexerMessage($namenull$skip));
  140.         }
  141.     }
  142.     public function has(string $name): bool
  143.     {
  144.         return $this->getIndexer($name) !== null;
  145.     }
  146.     public function getIndexer(string $name): ?EntityIndexer
  147.     {
  148.         foreach ($this->indexer as $indexer) {
  149.             if ($indexer->getName() === $name) {
  150.                 return $indexer;
  151.             }
  152.         }
  153.         return null;
  154.     }
  155.     private function useQueue(Context $context): bool
  156.     {
  157.         return $context->hasExtension(self::USE_INDEXING_QUEUE) || $context->hasState(self::USE_INDEXING_QUEUE);
  158.     }
  159.     private function disabled(Context $context): bool
  160.     {
  161.         return $context->hasExtension(self::DISABLE_INDEXING) || $context->hasState(self::DISABLE_INDEXING);
  162.     }
  163.     private function sendOrHandle(EntityIndexingMessage $messagebool $useQueue): void
  164.     {
  165.         if ($useQueue || $message->forceQueue()) {
  166.             $this->messageBus->dispatch($message);
  167.             return;
  168.         }
  169.         $this->handle($message);
  170.     }
  171.     private function iterateIndexer(string $name$offsetbool $useQueue, array $skip): ?EntityIndexingMessage
  172.     {
  173.         $indexer $this->getIndexer($name);
  174.         if (!$indexer instanceof EntityIndexer) {
  175.             throw new \RuntimeException(sprintf('Entity indexer with name %s not found'$name));
  176.         }
  177.         $message $indexer->iterate($offset);
  178.         if (!$message) {
  179.             return null;
  180.         }
  181.         $message->setIndexer($indexer->getName());
  182.         $message->addSkip(...$skip);
  183.         $this->sendOrHandle($message$useQueue);
  184.         return $message;
  185.     }
  186. }