diff --git a/README.md b/README.md index c25da2f..d84ec70 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,30 @@ valantic_elastica_bridge: # If true, when a document fails to be indexed, it will be skipped and indexing continue with the next document. If false, indexing that index will be aborted. should_skip_failing_documents: false ``` +## Events + +This project uses Symfony's event dispatcher. Here are the events that you can listen to: + +| Description | Example Usage | Event Constant (`ElasticaBridgeEvents::`) | Event Object (`Model\Event\`) | +|-----------------------------------------------------|----------------------------------------------------------------------|-------------------------------------------|--------------------------------| +| After an element has been refreshed in an index. | Log Event, send notification | `POST_REFRESH_ELEMENT_IN_INDEX` | `RefreshedElementInIndexEvent` | +| Before an element is refreshed in an index. | Stop propagation of element in specific index | `PRE_REFRESH_ELEMENT_IN_INDEX` | `RefreshedElementInIndexEvent` | +| After an element has been refreshed in all indices. | Clear caches, refresh related Objects, Log Event, send notification | `POST_REFRESH_ELEMENT` | `RefreshedElementEvent` | +| Before an element is refreshed in all indices. | Stop propagation of element in all indices | `PRE_REFRESH_ELEMENT` | `RefreshedElementEvent` | + +You can create an event subscriber or an event listener to listen to these events. Please refer to the [Symfony documentation](https://symfony.com/doc/current/event_dispatcher.html) for more information on how to use the event dispatcher. + +### Possible Use Cases for Events + +- clear cache after an element has been refreshed +- send a notification after an element has been refreshed +- log the event +- update related elements in the index + +### Event Propagation + +When refreshing multiple elements, each refresh triggers an event that could potentially lead to another refresh, resulting in an endless loop. To prevent this, you can disable event propagation during the refresh process. +You can disable event propagation by setting `$stopPropagateEvents` to `true` in the `RefreshElement` Message constructor or by calling `stopEventPropagation()` on the message before you add it to the queue. ## Queue diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon index 50cfd7a..860dd8e 100644 --- a/phpstan-baseline.neon +++ b/phpstan-baseline.neon @@ -1,10 +1,5 @@ parameters: ignoreErrors: - - - message: "#^Method Valantic\\\\ElasticaBridgeBundle\\\\EventListener\\\\Pimcore\\\\ChangeListener\\:\\:getFreshElement\\(\\) should return TElement of Pimcore\\\\Model\\\\Asset\\|Pimcore\\\\Model\\\\DataObject\\\\AbstractObject\\|Pimcore\\\\Model\\\\Document but returns Pimcore\\\\Model\\\\Asset\\|Pimcore\\\\Model\\\\DataObject\\\\AbstractObject\\|Pimcore\\\\Model\\\\Document\\.$#" - count: 1 - path: src/EventListener/Pimcore/ChangeListener.php - - message: "#^Call to function is_subclass_of\\(\\) with Pimcore\\\\Model\\\\Element\\\\AbstractElement and mixed will always evaluate to true\\.$#" count: 1 diff --git a/src/Enum/ElementInIndexOperation.php b/src/Enum/ElementInIndexOperation.php new file mode 100644 index 0000000..c9dbfb3 --- /dev/null +++ b/src/Enum/ElementInIndexOperation.php @@ -0,0 +1,13 @@ + $elementClass */ $elementClass = $element::class; $e = new PimcoreElementNotFoundException($element->getId(), $elementClass); @@ -117,6 +118,7 @@ private function getFreshElement(AbstractElement $element): AbstractElement throw $e; } + /** @var TElement */ return $elementClass::getById($element->getId(), ['force' => true]) ?? throw $e; } diff --git a/src/Messenger/Handler/RefreshElementHandler.php b/src/Messenger/Handler/RefreshElementHandler.php index db251c7..acda8b3 100644 --- a/src/Messenger/Handler/RefreshElementHandler.php +++ b/src/Messenger/Handler/RefreshElementHandler.php @@ -23,6 +23,10 @@ public function __invoke(RefreshElement $message): void { $element = $this->resolveElement($message); + if ($message->isEventPropagationStopped()) { + PropagateChanges::stopPropagation(); + } + $this->propagateChanges->handle($element); } } diff --git a/src/Messenger/Handler/RefreshElementInIndexHandler.php b/src/Messenger/Handler/RefreshElementInIndexHandler.php index 1e715fd..6c63eba 100644 --- a/src/Messenger/Handler/RefreshElementInIndexHandler.php +++ b/src/Messenger/Handler/RefreshElementInIndexHandler.php @@ -28,6 +28,10 @@ public function __invoke(RefreshElementInIndex $message): void $index = $this->indexRepository->flattenedGet($message->index); $element = $this->resolveElement($message); + if ($message->isEventPropagationStopped()) { + PropagateChanges::stopPropagation(); + } + if ($index->usesBlueGreenIndices() && !$this->lockService->getIndexingLock($index)->acquire()) { $this->propagateChanges->handleIndex($element, $index, $index->getBlueGreenInactiveElasticaIndex()); } diff --git a/src/Messenger/Message/AbstractRefresh.php b/src/Messenger/Message/AbstractRefresh.php index 1c9d33e..22d03e5 100644 --- a/src/Messenger/Message/AbstractRefresh.php +++ b/src/Messenger/Message/AbstractRefresh.php @@ -11,10 +11,26 @@ abstract class AbstractRefresh /** @var class-string */ public string $className; public int $id; + private bool $shouldStopEventPropagation = false; + + public function isEventPropagationStopped(): bool + { + return $this->shouldStopEventPropagation; + } + + public function stopEventPropagation(): void + { + $this->setShouldStopEventPropagation(true); + } protected function setElement(ElementInterface $element): void { $this->className = $element::class; $this->id = $element->getId() ?? throw new \InvalidArgumentException('Pimcore ID is null.'); } + + protected function setShouldStopEventPropagation(bool $stopEventPropagation): void + { + $this->shouldStopEventPropagation = $stopEventPropagation; + } } diff --git a/src/Messenger/Message/RefreshElement.php b/src/Messenger/Message/RefreshElement.php index b4785d0..5621d13 100644 --- a/src/Messenger/Message/RefreshElement.php +++ b/src/Messenger/Message/RefreshElement.php @@ -8,8 +8,11 @@ class RefreshElement extends AbstractRefresh { - public function __construct(ElementInterface $element) - { + public function __construct( + ElementInterface $element, + bool $stopEventPropagation = false, + ) { $this->setElement($element); + $this->setShouldStopEventPropagation($stopEventPropagation); } } diff --git a/src/Messenger/Message/RefreshElementInIndex.php b/src/Messenger/Message/RefreshElementInIndex.php index ebdcdd0..f21ed3d 100644 --- a/src/Messenger/Message/RefreshElementInIndex.php +++ b/src/Messenger/Message/RefreshElementInIndex.php @@ -11,7 +11,9 @@ class RefreshElementInIndex extends AbstractRefresh public function __construct( ElementInterface $element, public readonly string $index, + bool $stopEventPropagation = false, ) { + $this->setShouldStopEventPropagation($stopEventPropagation); $this->setElement($element); } } diff --git a/src/Model/Event/ElasticaBridgeEvents.php b/src/Model/Event/ElasticaBridgeEvents.php new file mode 100644 index 0000000..2cb2c6b --- /dev/null +++ b/src/Model/Event/ElasticaBridgeEvents.php @@ -0,0 +1,13 @@ + $indices + */ + public function __construct( + private readonly AbstractElement $element, + private readonly array $indices, + ) {} + + public function getElement(): AbstractElement + { + return $this->element; + } + + /** + * @return array + */ + public function getIndices(): array + { + return $this->indices; + } +} diff --git a/src/Model/Event/RefreshedElementInIndexEvent.php b/src/Model/Event/RefreshedElementInIndexEvent.php new file mode 100644 index 0000000..ba1f4a6 --- /dev/null +++ b/src/Model/Event/RefreshedElementInIndexEvent.php @@ -0,0 +1,41 @@ +element; + } + + public function getIndex(): IndexInterface + { + return $this->index; + } + + public function getElasticaIndex(): Index + { + return $this->elasticaIndex; + } + + public function getOperation(): ElementInIndexOperation + { + return $this->operation; + } +} diff --git a/src/Service/PropagateChanges.php b/src/Service/PropagateChanges.php index 141a113..4ea30e8 100644 --- a/src/Service/PropagateChanges.php +++ b/src/Service/PropagateChanges.php @@ -8,18 +8,26 @@ use Elastica\Index; use Pimcore\Model\DataObject\AbstractObject; use Pimcore\Model\Element\AbstractElement; +use Symfony\Component\EventDispatcher\EventDispatcherInterface; use Symfony\Component\Messenger\MessageBusInterface; use Valantic\ElasticaBridgeBundle\Document\DocumentInterface; +use Valantic\ElasticaBridgeBundle\Enum\ElementInIndexOperation; use Valantic\ElasticaBridgeBundle\Index\IndexInterface; use Valantic\ElasticaBridgeBundle\Messenger\Message\RefreshElementInIndex; +use Valantic\ElasticaBridgeBundle\Model\Event\ElasticaBridgeEvents; +use Valantic\ElasticaBridgeBundle\Model\Event\RefreshedElementEvent; +use Valantic\ElasticaBridgeBundle\Model\Event\RefreshedElementInIndexEvent; use Valantic\ElasticaBridgeBundle\Repository\IndexRepository; class PropagateChanges { + private static bool $isPropagationStopped = false; + public function __construct( private readonly IndexRepository $indexRepository, private readonly DocumentHelper $documentHelper, private readonly MessageBusInterface $messageBus, + private readonly EventDispatcherInterface $eventDispatcher, ) {} /** @@ -32,8 +40,24 @@ public function handle(AbstractElement $element): void { $indices = $this->matchingIndicesForElement($this->indexRepository->flattenedAll(), $element); + $event = new RefreshedElementEvent($element, $indices); + + if (!self::$isPropagationStopped) { + $this->eventDispatcher->dispatch($event, ElasticaBridgeEvents::PRE_REFRESH_ELEMENT); + } + foreach ($indices as $index) { - $this->messageBus->dispatch(new RefreshElementInIndex($element, $index->getName())); + $this->messageBus->dispatch( + new RefreshElementInIndex( + $element, + $index->getName(), + self::$isPropagationStopped || $event->isPropagationStopped() + ) + ); + } + + if (!self::$isPropagationStopped && !$event->isPropagationStopped()) { + $this->eventDispatcher->dispatch($event, ElasticaBridgeEvents::POST_REFRESH_ELEMENT); } } @@ -45,6 +69,11 @@ public function handleIndex( $this->doHandleIndex($element, $index, $elasticaIndex ?? $index->getElasticaIndex()); } + public static function stopPropagation(): void + { + self::$isPropagationStopped = true; + } + private function doHandleIndex( AbstractElement $element, IndexInterface $index, @@ -69,19 +98,30 @@ private function doHandleIndex( } $isPresent = $this->isIdInIndex($document::getElasticsearchId($element), $elasticaIndex); + $shouldIndex = $document->shouldIndex($element); - if ($document->shouldIndex($element)) { - if ($isPresent) { - $this->updateElementInIndex($element, $elasticaIndex, $document); - } + $operation = match (true) { + !$isPresent && $shouldIndex => ElementInIndexOperation::INSERT, + $isPresent && $shouldIndex => ElementInIndexOperation::UPDATE, + $isPresent && !$shouldIndex => ElementInIndexOperation::DELETE, + default => ElementInIndexOperation::NOTHING, + }; - if (!$isPresent) { - $this->addElementToIndex($element, $elasticaIndex, $document); - } + $event = new RefreshedElementInIndexEvent($element, $index, $elasticaIndex, $operation); + + if (!self::$isPropagationStopped) { + $this->eventDispatcher->dispatch($event, ElasticaBridgeEvents::PRE_REFRESH_ELEMENT_IN_INDEX); } - if ($isPresent && !$document->shouldIndex($element)) { - $this->deleteElementFromIndex($element, $elasticaIndex, $document); + match ($operation) { + ElementInIndexOperation::INSERT => $this->addElementToIndex($element, $elasticaIndex, $document), + ElementInIndexOperation::UPDATE => $this->updateElementInIndex($element, $elasticaIndex, $document), + ElementInIndexOperation::DELETE => $this->deleteElementFromIndex($element, $elasticaIndex, $document), + ElementInIndexOperation::NOTHING => null, + }; + + if (!self::$isPropagationStopped && !$event->isPropagationStopped()) { + $this->eventDispatcher->dispatch($event, ElasticaBridgeEvents::POST_REFRESH_ELEMENT_IN_INDEX); } $this->documentHelper->resetTenantIfNeeded($document, $index);