vendor/shopware/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.         }
  75.         $this->working = true;
  77.         if ($event->getContext()->hasExtension(self::DISABLE_INDEXING)) {
  78.             return;
  79.         }
  81.         $useQueue = $event->getContext()->hasExtension(self::USE_INDEXING_QUEUE);
  83.         foreach ($this->indexer as $indexer) {
  84.             $message = $indexer->update($event);
  86.             if (!$message) {
  87.                 continue;
  88.             }
  90.             $message->setIndexer($indexer->getName());
  92.             $this->sendOrHandle($message, $useQueue);
  93.         }
  95.         $this->working = false;
  96.     }
  98.     public function handle($message): void
  99.     {
  100.         if ($message instanceof EntityIndexingMessage) {
  101.             $indexer = $this->getIndexer($message->getIndexer());
  103.             if ($indexer) {
  104.                 $indexer->handle($message);
  105.             }
  107.             return;
  108.         }
  110.         if ($message instanceof IterateEntityIndexerMessage) {
  111.             $next = $this->iterateIndexer($message->getIndexer(), $message->getOffset(), true);
  113.             if (!$next) {
  114.                 return;
  115.             }
  117.             $this->messageBus->dispatch(new IterateEntityIndexerMessage($message->getIndexer(), $next->getOffset()));
  119.             return;
  120.         }
  121.     }
  123.     public function sendIndexingMessage(array $indexer = []): void
  124.     {
  125.         if (empty($indexer)) {
  126.             $indexer = [];
  127.             foreach ($this->indexer as $loop) {
  128.                 $indexer[] = $loop->getName();
  129.             }
  130.         }
  132.         if (empty($indexer)) {
  133.             return;
  134.         }
  136.         foreach ($indexer as $name) {
  137.             $this->messageBus->dispatch(new IterateEntityIndexerMessage($name, null));
  138.         }
  139.     }
  141.     public function has(string $name): bool
  142.     {
  143.         return $this->getIndexer($name) !== null;
  144.     }
  146.     private function sendOrHandle(EntityIndexingMessage $message, bool $useQueue): void
  147.     {
  148.         if ($useQueue || $message->forceQueue()) {
  149.             $this->messageBus->dispatch($message);
  151.             return;
  152.         }
  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. }