vendor/shopware/platform/src/Core/Framework/DataAbstractionLayer/Indexing/EntityIndexerRegistry.php line 69

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Framework\DataAbstractionLayer\Indexing;
  5. use Shopware\Core\Framework\DataAbstractionLayer\Event\EntityWrittenContainerEvent;
  6. use Shopware\Core\Framework\DataAbstractionLayer\Indexing\MessageQueue\IterateEntityIndexerMessage;
  7. use Shopware\Core\Framework\MessageQueue\Handler\AbstractMessageHandler;
  8. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  9. use Symfony\Component\Messenger\MessageBusInterface;
  11. class EntityIndexerRegistry extends AbstractMessageHandler implements EventSubscriberInterface
  12. {
  13.     public const USE_INDEXING_QUEUE = 'use-queue-indexing';
  14.     public const DISABLE_INDEXING = 'disable-indexing';
  16.     /**
  17.      * @var EntityIndexer[]
  18.      */
  19.     private $indexer;
  21.     /**
  22.      * @var MessageBusInterface
  23.      */
  24.     private $messageBus;
  26.     /**
  27.      * @var bool
  28.      */
  29.     private $working = false;
  31.     public function __construct(iterable $indexer, MessageBusInterface $messageBus)
  32.     {
  33.         $this->indexer = $indexer;
  34.         $this->messageBus = $messageBus;
  35.     }
  37.     public static function getSubscribedEvents(): array
  38.     {
  39.         return [
  40.             EntityWrittenContainerEvent::class => [
  41.                 ['refresh', 1000],
  42.             ],
  43.         ];
  44.     }
  46.     public static function getHandledMessages(): iterable
  47.     {
  48.         return [
  49.             EntityIndexingMessage::class,
  50.             IterateEntityIndexerMessage::class,
  51.         ];
  52.     }
  54.     public function index(bool $useQueue): void
  55.     {
  56.         foreach ($this->indexer as $indexer) {
  57.             $offset = null;
  59.             while ($message = $indexer->iterate($offset)) {
  60.                 $message->setIndexer($indexer->getName());
  62.                 $this->sendOrHandle($message, $useQueue);
  64.                 $offset = $message->getOffset();
  65.             }
  66.         }
  67.     }
  69.     public function refresh(EntityWrittenContainerEvent $event): void
  70.     {
  71.         if ($this->working) {
  72.             return;
  73.         }
  74.         $this->working = true;
  76.         if ($event->getContext()->hasExtension(self::DISABLE_INDEXING)) {
  77.             $this->working = false;
  79.             return;
  80.         }
  82.         $useQueue = $event->getContext()->hasExtension(self::USE_INDEXING_QUEUE);
  84.         foreach ($this->indexer as $indexer) {
  85.             $message = $indexer->update($event);
  87.             if (!$message) {
  88.                 continue;
  89.             }
  91.             $message->setIndexer($indexer->getName());
  93.             $this->sendOrHandle($message, $useQueue);
  94.         }
  96.         $this->working = false;
  97.     }
  99.     public function handle($message): void
  100.     {
  101.         if ($message instanceof EntityIndexingMessage) {
  102.             $indexer = $this->getIndexer($message->getIndexer());
  104.             if ($indexer) {
  105.                 $indexer->handle($message);
  106.             }
  108.             return;
  109.         }
  111.         if ($message instanceof IterateEntityIndexerMessage) {
  112.             $next = $this->iterateIndexer($message->getIndexer(), $message->getOffset(), true);
  114.             if (!$next) {
  115.                 return;
  116.             }
  118.             $this->messageBus->dispatch(new IterateEntityIndexerMessage($message->getIndexer(), $next->getOffset()));
  120.             return;
  121.         }
  122.     }
  124.     public function sendIndexingMessage(array $indexer = []): void
  125.     {
  126.         if (empty($indexer)) {
  127.             $indexer = [];
  128.             foreach ($this->indexer as $loop) {
  129.                 $indexer[] = $loop->getName();
  130.             }
  131.         }
  133.         if (empty($indexer)) {
  134.             return;
  135.         }
  137.         foreach ($indexer as $name) {
  138.             $this->messageBus->dispatch(new IterateEntityIndexerMessage($name, null));
  139.         }
  140.     }
  142.     public function has(string $name): bool
  143.     {
  144.         return $this->getIndexer($name) !== null;
  145.     }
  147.     private function sendOrHandle(EntityIndexingMessage $message, bool $useQueue): void
  148.     {
  149.         if ($useQueue || $message->forceQueue()) {
  150.             $this->messageBus->dispatch($message);
  152.             return;
  153.         }
  154.         $this->handle($message);
  155.     }
  157.     private function getIndexer(string $name): ?EntityIndexer
  158.     {
  159.         foreach ($this->indexer as $indexer) {
  160.             if ($indexer->getName() === $name) {
  161.                 return $indexer;
  162.             }
  163.         }
  165.         return null;
  166.     }
  168.     private function iterateIndexer(string $name, $offset, bool $useQueue): ?EntityIndexingMessage
  169.     {
  170.         $indexer = $this->getIndexer($name);
  172.         if (!$indexer instanceof EntityIndexer) {
  173.             throw new \RuntimeException(sprintf('Entity indexer with name %s not found', $name));
  174.         }
  176.         $message = $indexer->iterate($offset);
  177.         if (!$message) {
  178.             return null;
  179.         }
  181.         $message->setIndexer($indexer->getName());
  183.         $this->sendOrHandle($message, $useQueue);
  185.         return $message;
  186.     }
  187. }