diff --git a/.github/workflows/php.yml b/.github/workflows/php.yml
index 867f572..4b8327c 100644
--- a/.github/workflows/php.yml
+++ b/.github/workflows/php.yml
@@ -17,7 +17,6 @@ jobs:
php: [ '8.1', '8.2' ]
pimcore: [ '^11.0' ]
stability: [ prefer-lowest, prefer-stable ]
- elastica: [ '^7.1', '8.x-dev#0f6b04b' ]
steps:
- name: Checkout code
@@ -34,7 +33,7 @@ jobs:
- name: Install dependencies
run: |
- composer require "pimcore/pimcore:${{ matrix.pimcore }}" "ruflin/elastica:${{ matrix.elastica }}" --no-interaction --no-update
+ composer require "pimcore/pimcore:${{ matrix.pimcore }}" --no-interaction --no-update
composer update --${{ matrix.stability }} --prefer-dist --no-interaction
- name: List installed dependencies
diff --git a/README.md b/README.md
index 535dc66..c25da2f 100644
--- a/README.md
+++ b/README.md
@@ -12,8 +12,8 @@ The only job of the bundle is to store Pimcore elements (assets, documents, data
1. `composer require valantic/pimcore-elastica-bridge`
1. Edit `config/bundles.php` and add `\Valantic\ElasticaBridgeBundle\ValanticElasticaBridgeBundle::class => ['all' => true],`
-1. Configure the connection to your Elasticsearch cluster as seen in [`example/app/config/config.yml`](example/app/config/config.yml)
-1. Don't forget to register your newly created services (implementing `IndexInterface` etc.) in your `services.yml`
+1. Configure the connection to your Elasticsearch cluster as seen in [`example/app/config/config.yaml`](example/app/config/config.yaml)
+1. Don't forget to register your newly created services (implementing `IndexInterface` etc.) in your `services.yaml`
```yml
App\Elasticsearch\:
resource: '../../Elasticsearch'
@@ -53,11 +53,25 @@ See the [`ProductIndexDocument` provided in the example](docs/example/src/Elasti
```yaml
valantic_elastica_bridge:
client:
- host: localhost
- port: 9200
- addSentryBreadcrumbs: false
+
+ # The DSN to connect to the Elasticsearch cluster.
+ dsn: 'http://localhost:9200'
+
+ # If true, breadcrumbs are added to Sentry for every request made to Elasticsearch via Elastica.
+ should_add_sentry_breadcrumbs: false
+ indexing:
+
+ # To prevent overlapping indexing jobs. Set to a value higher than the slowest index. Value is specified in seconds.
+ lock_timeout: 300
+
+ # 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
```
+## Queue
+
+[Set up a worker](https://symfony.com/doc/current/messenger.html#consuming-messages-running-the-worker) to process `elastica_bridge_index`. Alternatively you can route the transport to use the `sync` handler: `framework.messenger.transports.elastica_bridge_index: 'sync'`.
+
## Indexing
### Bulk
@@ -86,6 +100,8 @@ The bridge automatically listens to Pimcore events and updates documents as need
This can be globally disabled by calling `\Valantic\ElasticaBridgeBundle\EventListener\Pimcore\ChangeListener::disableListener();`.
+You can also dispatch a `Valantic\ElasticaBridgeBundle\Messenger\Message\RefreshElement` message to handle updates to related objects which are not triggered by the `ChangeListener`.
+
## Status
```
diff --git a/UPGRADE.md b/UPGRADE.md
index 657db9f..46d3091 100644
--- a/UPGRADE.md
+++ b/UPGRADE.md
@@ -1,14 +1,26 @@
-# Upgrade from v1 to v2
+# UPGRADE
+## Upgrade from v3 to v4
-## Migration
+- Remove deprecated options `valantic_elastica_bridge.client.host` and `valantic_elastica_bridge.client.port`. Use `valantic_elastica_bridge.client.dsn` instead, e.g. `http://localhost:9200`
+- Renamed `valantic_elastica_bridge.client.addSentryBreadcrumbs` to `valantic_elastica_bridge.client.should_add_sentry_breadcrumbs`
+- See [README#Queue](./README#queue) to set up the **required** Symfony Messenger workers.
+
+## Upgrade from v2 to v3
+
+- no code changes necessary
+
+## Upgrade from v1 to v2
+
+
+### Migration
- `IndexDocumentInterface` implementations should now extend `\Valantic\ElasticaBridgeBundle\Document\AbstractDocument`. `getType()` should now return one of `\Valantic\ElasticaBridgeBundle\Enum\DocumentType`
- `Valantic\ElasticaBridgeBundle\DocumentType\Index\ListingTrait` was removed, remove any references to it [#30](https://github.com/valantic/pimcore-elastica-bridge/issues/30)
- Update references to renamed classes and interfaces (see next section)
- see also the example in [`docs/example/`](./docs/example/)
-## Breaking Changes
+### Breaking Changes
- PHP 8.1+ [#26](https://github.com/valantic/pimcore-elastica-bridge/issues/26)
- `\Valantic\ElasticaBridgeBundle\EventListener\Pimcore\AbstractListener` was renamed to `\Valantic\ElasticaBridgeBundle\EventListener\Pimcore\ChangeListener`
@@ -23,14 +35,14 @@
- `Valantic\ElasticaBridgeBundle\Document\TenantAwareTrait` has been replaced by `Valantic\ElasticaBridgeBundle\Document\AbstractTenantAwareDocument`
- `Valantic\ElasticaBridgeBundle\DocumentType\Index\DataObjectNormalizerTrait` was moved to `Valantic\ElasticaBridgeBundle\Document\DataObjectNormalizerTrait`
-## New Features
+### New Features
- PHPStan generics annotations for `\Valantic\ElasticaBridgeBundle\Document\DocumentInterface` and related helper traits [#32](https://github.com/valantic/pimcore-elastica-bridge/issues/32)
- Added `\Valantic\ElasticaBridgeBundle\Service\PropagateChanges::handle` to programmatically update an element in all indices [#33](https://github.com/valantic/pimcore-elastica-bridge/issues/33)
- Added support for assets [#34](https://github.com/valantic/pimcore-elastica-bridge/issues/34)
- Allow `\Valantic\ElasticaBridgeBundle\Document\DocumentInterface::getSubType` to return `null` for generic, element-level indices [#42](https://github.com/valantic/pimcore-elastica-bridge/issues/42)
-## Other changes
+### Other changes
- `:cleanup` now defaults to only cleaning up bundle indices [#27](https://github.com/valantic/pimcore-elastica-bridge/issues/27)
- Removed `--check` from `:index` [#41](https://github.com/valantic/pimcore-elastica-bridge/issues/41)
diff --git a/composer.json b/composer.json
index bd3533f..d417634 100644
--- a/composer.json
+++ b/composer.json
@@ -8,7 +8,7 @@
"ext-json": "*",
"pimcore/pimcore": "^11.0",
"psr/log": "^3.0",
- "ruflin/elastica": "^7.1 || 8.x-dev#0f6b04b",
+ "ruflin/elastica": "8.x-dev",
"symfony/console": "^6.2",
"symfony/lock": "^6.2"
},
@@ -18,7 +18,7 @@
"phpstan/phpstan": "^1.10.58",
"phpstan/phpstan-deprecation-rules": "^1.1.4",
"phpstan/phpstan-strict-rules": "^1.5.2",
- "rector/rector": "^0.18.13",
+ "rector/rector": "^1.0.1",
"roave/security-advisories": "dev-latest",
"sentry/sentry": "^3.22.1",
"symfony/http-client": "^6.4.3"
diff --git a/docs/example/config/config.yaml b/docs/example/config/config.yaml
new file mode 100644
index 0000000..7f91c51
--- /dev/null
+++ b/docs/example/config/config.yaml
@@ -0,0 +1,3 @@
+valantic_elastica_bridge:
+ client:
+ dsn: 'http://localhost:9200'
diff --git a/docs/example/config/config.yml b/docs/example/config/config.yml
deleted file mode 100644
index ff5b19a..0000000
--- a/docs/example/config/config.yml
+++ /dev/null
@@ -1,4 +0,0 @@
-valantic_elastica_bridge:
- client:
- host: 'localhost'
- port: 9200
diff --git a/rector.php b/rector.php
index c7ab87e..d60fd3e 100644
--- a/rector.php
+++ b/rector.php
@@ -2,30 +2,39 @@
declare(strict_types=1);
+use Rector\CodeQuality\Rector\Foreach_\SimplifyForeachToCoalescingRector;
+use Rector\CodeQuality\Rector\FuncCall\SingleInArrayToCompareRector;
use Rector\CodingStyle\Rector\FuncCall\CountArrayToEmptyArrayComparisonRector;
use Rector\Config\RectorConfig;
use Rector\Php55\Rector\String_\StringClassNameToClassConstantRector;
use Rector\Php80\Rector\FunctionLike\MixedTypeRector;
-use Rector\Set\ValueObject\LevelSetList;
-use Rector\Set\ValueObject\SetList;
-return static function (RectorConfig $rectorConfig): void {
- $rectorConfig->paths([
+return RectorConfig::configure()
+ ->withPhpSets()
+ ->withPreparedSets(
+ codeQuality: true,
+ )
+ ->withAttributesSets(
+ symfony: true,
+ doctrine: true,
+ )
+ ->withPaths([
__DIR__ . '/src',
- ]);
-
- $rectorConfig->skip([
+ ])
+ ->withSkip([
CountArrayToEmptyArrayComparisonRector::class,
StringClassNameToClassConstantRector::class => [
'src/Elastica/Client/ElasticsearchClientFactory.php',
],
MixedTypeRector::class => [
'src/Document/DataObjectNormalizerTrait.php',
+ 'src/Index/IndexInterface.php',
],
- ]);
-
- $rectorConfig->sets([
- LevelSetList::UP_TO_PHP_81,
- SetList::CODE_QUALITY,
- ]);
-};
+ SingleInArrayToCompareRector::class => [
+ 'src/Command/NonBundleIndexTrait.php',
+ ],
+ SimplifyForeachToCoalescingRector::class => [
+ 'src/Repository/IndexRepository.php',
+ ],
+ ])
+ ->withRootFiles();
diff --git a/src/Command/Cleanup.php b/src/Command/Cleanup.php
index b327f02..1d55788 100644
--- a/src/Command/Cleanup.php
+++ b/src/Command/Cleanup.php
@@ -4,7 +4,7 @@
namespace Valantic\ElasticaBridgeBundle\Command;
-use Elastica\Exception\ResponseException;
+use Elastic\Elasticsearch\Exception\ElasticsearchException;
use Symfony\Component\Console\Helper\QuestionHelper;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
@@ -15,6 +15,7 @@
class Cleanup extends BaseCommand
{
+ use NonBundleIndexTrait;
private const OPTION_ALL_IN_CLUSTER = 'all';
public function __construct(
@@ -54,15 +55,23 @@ protected function execute(InputInterface $input, OutputInterface $output): int
$indices = $this->getIndices();
foreach ($indices as $index) {
+ if (!$this->shouldProcessNonBundleIndex($index)) {
+ continue;
+ }
+
$client = $this->esClient->getIndex($index);
+ if ($client->getSettings()->getBool('hidden')) {
+ continue;
+ }
+
foreach ($client->getAliases() as $alias) {
$client->removeAlias($alias);
}
try {
$client->delete();
- } catch (ResponseException $e) {
+ } catch (ElasticsearchException $e) {
$this->output->writeln(sprintf('%s', $e->getMessage()));
}
}
@@ -81,7 +90,7 @@ private function getIndices(): array
$indices = [];
- foreach ($this->indexRepository->flattened() as $indexConfig) {
+ foreach ($this->indexRepository->flattenedAll() as $indexConfig) {
if ($indexConfig->usesBlueGreenIndices()) {
$indices[] = $indexConfig->getBlueGreenActiveElasticaIndex()->getName();
$indices[] = $indexConfig->getBlueGreenInactiveElasticaIndex()->getName();
diff --git a/src/Command/Index.php b/src/Command/Index.php
index 023e27f..157829c 100644
--- a/src/Command/Index.php
+++ b/src/Command/Index.php
@@ -5,22 +5,20 @@
namespace Valantic\ElasticaBridgeBundle\Command;
use Elastica\Index as ElasticaIndex;
-use Elastica\Request;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\ConsoleOutput;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\HttpKernel\KernelInterface;
-use Symfony\Component\Lock\LockFactory;
-use Symfony\Component\Lock\LockInterface;
use Symfony\Component\Process\Process;
use Valantic\ElasticaBridgeBundle\Elastica\Client\ElasticsearchClient;
use Valantic\ElasticaBridgeBundle\Enum\IndexBlueGreenSuffix;
use Valantic\ElasticaBridgeBundle\Exception\Index\BlueGreenIndicesIncorrectlySetupException;
use Valantic\ElasticaBridgeBundle\Index\IndexInterface;
-use Valantic\ElasticaBridgeBundle\Repository\ConfigurationRepository;
use Valantic\ElasticaBridgeBundle\Repository\IndexRepository;
+use Valantic\ElasticaBridgeBundle\Service\LockService;
+use Valantic\ElasticaBridgeBundle\Util\ElasticsearchResponse;
class Index extends BaseCommand
{
@@ -34,8 +32,7 @@ public function __construct(
private readonly IndexRepository $indexRepository,
private readonly ElasticsearchClient $esClient,
private readonly KernelInterface $kernel,
- private readonly LockFactory $lockFactory,
- private readonly ConfigurationRepository $configurationRepository,
+ private readonly LockService $lockService,
) {
parent::__construct();
}
@@ -73,7 +70,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
{
$skippedIndices = [];
- foreach ($this->indexRepository->flattened() as $indexConfig) {
+ foreach ($this->indexRepository->flattenedAll() as $indexConfig) {
if (
is_array($this->input->getArgument(self::ARGUMENT_INDEX))
&& count($this->input->getArgument(self::ARGUMENT_INDEX)) > 0
@@ -84,7 +81,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
continue;
}
- $lock = $this->getLock($indexConfig);
+ $lock = $this->lockService->getIndexingLock($indexConfig);
if (!$lock->acquire()) {
if ($this->input->getOption(self::OPTION_LOCK_RELEASE) === true) {
@@ -229,7 +226,7 @@ private function ensureCorrectBlueGreenIndexSetup(IndexInterface $indexConfig):
// In case an index with the same name as the blue/green alias exists, delete it
if (
$nonAliasIndex->exists()
- && !$this->esClient->request('_alias/' . $indexConfig->getName(), Request::HEAD)->isOk()
+ && !ElasticsearchResponse::getResponse($this->esClient->indices()->existsAlias(['name' => $indexConfig->getName()]))->asBool()
) {
$nonAliasIndex->delete();
$this->output->writeln('-> Deleted non-blue/green index to prepare for blue/green usage');
@@ -259,13 +256,4 @@ private function ensureCorrectBlueGreenIndexSetup(IndexInterface $indexConfig):
$this->output->writeln('-> Ensured indices are correctly set up with alias');
}
-
- private function getLock(mixed $indexConfig): LockInterface
- {
- return $this->lockFactory
- ->createLock(
- __METHOD__ . '->' . $indexConfig->getName(),
- ttl: $this->configurationRepository->getIndexingLockTimeout()
- );
- }
}
diff --git a/src/Command/NonBundleIndexTrait.php b/src/Command/NonBundleIndexTrait.php
new file mode 100644
index 0000000..47f94f4
--- /dev/null
+++ b/src/Command/NonBundleIndexTrait.php
@@ -0,0 +1,13 @@
+indexRepository->flattened() as $indexConfig) {
+ foreach ($this->indexRepository->flattenedAll() as $indexConfig) {
if ($indexConfig->getName() === $this->input->getOption(self::OPTION_CONFIG)) {
return $indexConfig;
}
@@ -92,13 +95,21 @@ private function populateIndex(IndexInterface $indexConfig, ElasticaIndex $esInd
$listing->setLimit($indexConfig->getBatchSize());
foreach ($listing->getData() ?? [] as $dataObject) {
- $progressBar->advance();
+ try {
+ $progressBar->advance();
- if (!$documentInstance->shouldIndex($dataObject)) {
- continue;
- }
+ if (!$documentInstance->shouldIndex($dataObject)) {
+ continue;
+ }
+
+ $esDocuments[] = $this->documentHelper->elementToDocument($documentInstance, $dataObject);
+ } catch (\Throwable $throwable) {
+ $this->displayDocumentError($indexConfig, $document, $dataObject, $throwable);
- $esDocuments[] = $this->documentHelper->elementToDocument($documentInstance, $dataObject);
+ if (!$this->configurationRepository->shouldSkipFailingDocuments()) {
+ throw new DocumentFailedException($throwable);
+ }
+ }
}
if (count($esDocuments) > 0) {
@@ -116,22 +127,7 @@ private function populateIndex(IndexInterface $indexConfig, ElasticaIndex $esInd
}
}
} catch (\Throwable $throwable) {
- $this->output->writeln('');
- $this->output->writeln(sprintf(
- 'Error while populating index %s, processing documents of type %s, last processed element ID %s.',
- $indexConfig::class,
- $document ?? '(N/A)',
- isset($dataObject) && $dataObject instanceof AbstractElement ? $dataObject->getId() : '(N/A)'
- ));
- $this->output->writeln('');
- $this->output->writeln(sprintf('In %s line %d', $throwable->getFile(), $throwable->getLine()));
- $this->output->writeln('');
-
- $this->output->writeln($throwable->getMessage());
- $this->output->writeln('');
-
- $this->output->writeln($throwable->getTraceAsString());
- $this->output->writeln('');
+ $this->displayIndexError($indexConfig, $throwable);
throw new IndexingFailedException($throwable);
} finally {
@@ -143,4 +139,44 @@ private function populateIndex(IndexInterface $indexConfig, ElasticaIndex $esInd
$progressBar->finish();
$this->output->writeln('');
}
+
+ private function displayDocumentError(
+ IndexInterface $indexConfig,
+ string $document,
+ AbstractElement $dataObject,
+ \Throwable $throwable,
+ ): void {
+ $this->output->writeln('');
+ $this->output->writeln(sprintf(
+ 'Error while populating index %s, processing documents of type %s, last processed element ID %s.',
+ $indexConfig::class,
+ $document,
+ $dataObject->getId()
+ ));
+ $this->displayThrowable($throwable);
+ }
+
+ private function displayIndexError(IndexInterface $indexConfig, \Throwable $throwable): void
+ {
+ $this->output->writeln('');
+ $this->output->writeln(sprintf(
+ 'Error while populating index %s.',
+ $indexConfig::class,
+ ));
+
+ $this->displayThrowable($throwable);
+ }
+
+ private function displayThrowable(\Throwable $throwable): void
+ {
+ $this->output->writeln('');
+ $this->output->writeln(sprintf('In %s line %d', $throwable->getFile(), $throwable->getLine()));
+ $this->output->writeln('');
+
+ $this->output->writeln($throwable->getMessage());
+ $this->output->writeln('');
+
+ $this->output->writeln($throwable->getTraceAsString());
+ $this->output->writeln('');
+ }
}
diff --git a/src/Command/Status.php b/src/Command/Status.php
index af616fb..6d66c96 100644
--- a/src/Command/Status.php
+++ b/src/Command/Status.php
@@ -14,6 +14,7 @@
class Status extends BaseCommand
{
+ use NonBundleIndexTrait;
/**
* @var array>
*/
@@ -55,7 +56,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
$this->output->writeln('');
- foreach ($this->indexRepository->flattened() as $indexConfig) {
+ foreach ($this->indexRepository->flattenedAll() as $indexConfig) {
$this->processBundleIndex($indexConfig);
}
@@ -66,11 +67,19 @@ protected function execute(InputInterface $input, OutputInterface $output): int
->setHeaderTitle('Indices (managed by this bundle)');
$table->render();
+ $otherIndexNames = [];
+
foreach ($this->esClient->getCluster()->getIndexNames() as $indexName) {
- if (in_array($indexName, $this->skipOtherIndices, true)) {
+ if (in_array($indexName, $this->skipOtherIndices, true) || !$this->shouldProcessNonBundleIndex($indexName)) {
continue;
}
- $this->processOtherIndex($indexName);
+ $otherIndexNames[] = $indexName;
+ }
+
+ sort($otherIndexNames);
+
+ foreach ($otherIndexNames as $otherIndexName) {
+ $this->processOtherIndex($otherIndexName);
}
$this->output->writeln('');
diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php
index 00bacd7..d3598df 100644
--- a/src/DependencyInjection/Configuration.php
+++ b/src/DependencyInjection/Configuration.php
@@ -7,16 +7,8 @@
use Symfony\Component\Config\Definition\Builder\TreeBuilder;
use Symfony\Component\Config\Definition\ConfigurationInterface;
-/**
- * This is the class that validates and merges configuration from your app/config files.
- *
- * To learn more see {@link http://symfony.com/doc/current/cookbook/bundles/configuration.html}
- */
class Configuration implements ConfigurationInterface
{
- /**
- * {@inheritDoc}
- */
public function getConfigTreeBuilder()
{
$treeBuilder = new TreeBuilder('valantic_elastica_bridge');
@@ -24,16 +16,15 @@ public function getConfigTreeBuilder()
->children()
->arrayNode('client')
->children()
- ->scalarNode('host')->defaultValue('localhost')->setDeprecated('valantic/pimcore-elastica-bridge', '3.1.0', 'Use the "dsn" option instead, e.g. http://username:password@localhost:9200')->end()
- ->integerNode('port')->defaultValue(9200)->setDeprecated('valantic/pimcore-elastica-bridge', '3.1.0', 'Use the "dsn" option instead, e.g. http://username:password@localhost:9200')->end()
- ->scalarNode('dsn')->defaultNull()->end()
- ->booleanNode('addSentryBreadcrumbs')->defaultValue(false)->end()
+ ->scalarNode('dsn')->defaultValue('http://localhost:9200')->info('The DSN to connect to the Elasticsearch cluster.')->end()
+ ->booleanNode('should_add_sentry_breadcrumbs')->defaultFalse()->info('If true, breadcrumbs are added to Sentry for every request made to Elasticsearch via Elastica.')->end()
->end()
->end()
->arrayNode('indexing')
->addDefaultsIfNotSet()
->children()
->integerNode('lock_timeout')->defaultValue(5 * 60)->info('To prevent overlapping indexing jobs. Set to a value higher than the slowest index. Value is specified in seconds.')->end()
+ ->booleanNode('should_skip_failing_documents')->defaultFalse()->info('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.')->end()
->end()
->end()
->end()
diff --git a/src/DependencyInjection/ValanticElasticaBridgeExtension.php b/src/DependencyInjection/ValanticElasticaBridgeExtension.php
index 3e02a4e..655a846 100644
--- a/src/DependencyInjection/ValanticElasticaBridgeExtension.php
+++ b/src/DependencyInjection/ValanticElasticaBridgeExtension.php
@@ -37,7 +37,7 @@ public function load(array $configs, ContainerBuilder $container): void
$config = $this->processConfiguration($configuration, $configs);
$loader = new Loader\YamlFileLoader($container, new FileLocator(__DIR__ . '/../Resources/config'));
- $loader->load('services.yml');
+ $loader->load('services.yaml');
$container->setParameter('valantic_elastica_bridge', $config);
}
diff --git a/src/Document/AbstractDocument.php b/src/Document/AbstractDocument.php
index eb674b5..a125b3e 100644
--- a/src/Document/AbstractDocument.php
+++ b/src/Document/AbstractDocument.php
@@ -5,11 +5,8 @@
namespace Valantic\ElasticaBridgeBundle\Document;
use Pimcore\Model\Asset;
-use Pimcore\Model\Asset\Listing as AssetListing;
use Pimcore\Model\DataObject;
-use Pimcore\Model\DataObject\Listing as DataObjectListing;
use Pimcore\Model\Document as PimcoreDocument;
-use Pimcore\Model\Document\Listing as DocumentListing;
use Pimcore\Model\Element\AbstractElement;
use Pimcore\Model\Listing\AbstractListing;
use Valantic\ElasticaBridgeBundle\Enum\DocumentType;
@@ -42,12 +39,12 @@ public function getListingInstance(IndexInterface $index): AbstractListing
}
if (in_array($this->getType(), DocumentType::casesPublishedState(), true)) {
- /** @var DocumentListing|DataObjectListing $listingInstance */
+ /** @var PimcoreDocument\Listing|DataObject\Listing $listingInstance */
$listingInstance->setUnpublished($this->includeUnpublishedElementsInListing());
}
if ($this->getType() === DocumentType::DATA_OBJECT) {
- /** @var DataObjectListing $listingInstance */
+ /** @var DataObject\Listing $listingInstance */
if ($this->treatObjectVariantsAsDocuments()) {
$listingInstance->setObjectTypes([
DataObject\AbstractObject::OBJECT_TYPE_OBJECT,
@@ -137,8 +134,8 @@ protected function getListingClass(): string
{
try {
return match ($this->getType()) {
- DocumentType::ASSET => AssetListing::class,
- DocumentType::DOCUMENT => DocumentListing::class,
+ DocumentType::ASSET => Asset\Listing::class,
+ DocumentType::DOCUMENT => PimcoreDocument\Listing::class,
DocumentType::DATA_OBJECT, DocumentType::VARIANT => $this->getDataObjectListingClass(),
};
} catch (\UnhandledMatchError) {
@@ -214,7 +211,7 @@ private function getDataObjectListingClass(): string
$subType = $this->getSubType();
if ($subType === null) {
- return DataObjectListing::class;
+ return DataObject\Listing::class;
}
$className = $subType . '\Listing';
diff --git a/src/Document/DataObjectNormalizerTrait.php b/src/Document/DataObjectNormalizerTrait.php
index c51149a..16ab6df 100644
--- a/src/Document/DataObjectNormalizerTrait.php
+++ b/src/Document/DataObjectNormalizerTrait.php
@@ -9,6 +9,7 @@
use Pimcore\Model\DataObject\Concrete;
use Pimcore\Model\DataObject\Localizedfield;
use Pimcore\Tool;
+use Symfony\Contracts\Service\Attribute\Required;
/**
* Collection of helpers for normalizing a DataObject.
@@ -19,11 +20,7 @@ trait DataObjectNormalizerTrait
{
protected LocaleService $localeService;
- /**
- * Injects the LocaleService using Symfony's DI.
- *
- * @required
- */
+ #[Required]
public function setLocaleService(LocaleService $localeService): void
{
$this->localeService = $localeService;
diff --git a/src/Document/DocumentInterface.php b/src/Document/DocumentInterface.php
index 2dc0c1f..84fce99 100644
--- a/src/Document/DocumentInterface.php
+++ b/src/Document/DocumentInterface.php
@@ -63,7 +63,7 @@ public function getType(): DocumentType;
*
* Returning null will result in all elements of getType() being included.
*
- * @return ?class-string
+ * @return ?class-string
*/
public function getSubType(): ?string;
diff --git a/src/Document/DocumentNormalizerTrait.php b/src/Document/DocumentNormalizerTrait.php
index 84508e8..b0a5d8b 100644
--- a/src/Document/DocumentNormalizerTrait.php
+++ b/src/Document/DocumentNormalizerTrait.php
@@ -37,10 +37,10 @@ protected function editables(Document\Page $document): array
$data = [];
$editableNames = array_merge(
array_map(fn (Document\Editable $editable): string => $editable->getName(), $document->getEditables()),
- $document->getContentMasterDocument() instanceof Document\PageSnippet
+ $document->getContentMainDocument() instanceof Document\PageSnippet
? array_map(
fn (Document\Editable $editable): string => $editable->getName(),
- $document->getContentMasterDocument()->getEditables()
+ $document->getContentMainDocument()->getEditables()
)
: []
);
@@ -84,7 +84,9 @@ protected function editableRelation(
[$contents->getId()],
array_map(
fn (Concrete $obj): int => $obj->getId(),
- $contents->getChildren([AbstractObject::OBJECT_TYPE_OBJECT])
+ $contents
+ ->getChildren([AbstractObject::OBJECT_TYPE_OBJECT])
+ ->getData() ?? []
)
);
diff --git a/src/Elastica/Client/ElasticsearchClientFactory.php b/src/Elastica/Client/ElasticsearchClientFactory.php
index 85cda05..adcc074 100644
--- a/src/Elastica/Client/ElasticsearchClientFactory.php
+++ b/src/Elastica/Client/ElasticsearchClientFactory.php
@@ -15,9 +15,9 @@ public function __construct(
public function __invoke(
): ElasticsearchClient {
- $esClient = new ElasticsearchClient($this->configurationRepository->getClient());
+ $esClient = new ElasticsearchClient($this->configurationRepository->getClientDsn());
- if ($this->configurationRepository->getAddSentryBreadcrumbs() && class_exists('\Sentry\Breadcrumb')) {
+ if ($this->configurationRepository->shouldAddSentryBreadcrumbs() && class_exists('\Sentry\Breadcrumb')) {
$esClient->setLogger(new SentryBreadcrumbLogger());
}
diff --git a/src/EventListener/Pimcore/ChangeListener.php b/src/EventListener/Pimcore/ChangeListener.php
index 8ff61fa..0489a26 100644
--- a/src/EventListener/Pimcore/ChangeListener.php
+++ b/src/EventListener/Pimcore/ChangeListener.php
@@ -15,8 +15,9 @@
use Pimcore\Model\Document;
use Pimcore\Model\Element\AbstractElement;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\MessageBusInterface;
use Valantic\ElasticaBridgeBundle\Exception\EventListener\PimcoreElementNotFoundException;
-use Valantic\ElasticaBridgeBundle\Service\PropagateChanges;
+use Valantic\ElasticaBridgeBundle\Messenger\Message\RefreshElement;
/**
* An abstract listener for DataObject and Document listeners.
@@ -28,7 +29,7 @@ class ChangeListener implements EventSubscriberInterface
private static bool $isEnabled = true;
public function __construct(
- private readonly PropagateChanges $propagateChanges,
+ private readonly MessageBusInterface $messageBus,
) {}
public function handle(AssetEvent|DataObjectEvent|DocumentEvent $event): void
@@ -45,7 +46,7 @@ public function handle(AssetEvent|DataObjectEvent|DocumentEvent $event): void
return;
}
- $this->propagateChanges->handle($this->getFreshElement($element));
+ $this->messageBus->dispatch(new RefreshElement($this->getFreshElement($element)));
}
public static function enableListener(): void
diff --git a/src/Exception/Command/DocumentFailedException.php b/src/Exception/Command/DocumentFailedException.php
new file mode 100644
index 0000000..9870def
--- /dev/null
+++ b/src/Exception/Command/DocumentFailedException.php
@@ -0,0 +1,15 @@
+ [
+ DocumentInterface::META_ID => [
+ 'type' => 'keyword',
+ ],
+ DocumentInterface::META_TYPE => [
+ 'type' => 'keyword',
+ ],
+ DocumentInterface::META_SUB_TYPE => [
+ 'type' => 'keyword',
+ ],
+ ],
+ ];
}
public function getSettings(): array
@@ -112,10 +126,14 @@ final public function getBlueGreenActiveSuffix(): IndexBlueGreenSuffix
throw new BlueGreenIndicesIncorrectlySetupException();
}
- $aliases = array_filter(
- $this->client->request('_aliases')->getData(),
- fn (array $datum): bool => array_key_exists($this->getName(), $datum['aliases'])
- );
+ try {
+ $aliases = array_filter(
+ ElasticsearchResponse::getResponse($this->client->indices()->getAlias(['name' => $this->getName()]))->asArray(),
+ fn (array $datum): bool => array_key_exists($this->getName(), $datum['aliases'])
+ );
+ } catch (ClientResponseException) {
+ throw new BlueGreenIndicesIncorrectlySetupException();
+ }
if (count($aliases) !== 1) {
throw new BlueGreenIndicesIncorrectlySetupException();
diff --git a/src/Index/IndexInterface.php b/src/Index/IndexInterface.php
index 96f1d37..774aef7 100644
--- a/src/Index/IndexInterface.php
+++ b/src/Index/IndexInterface.php
@@ -35,7 +35,7 @@ public function getBatchSize(): int;
* Defines the mapping to be used for this index.
* Passed 1:1 to Elasticsearch.
*
- * @return array>
+ * @return array{properties:array}
*/
public function getMapping(): array;
@@ -57,7 +57,7 @@ public function getCreateArguments(): array;
/**
* Defines the types of documents found in this index. Array of classes implementing DocumentInterface.
*
- * @return string[] Class names of DocumentInterface classes
+ * @return class-string[] Class names of DocumentInterface classes
*
* @see DocumentInterface
*/
@@ -66,7 +66,7 @@ public function getAllowedDocuments(): array;
/**
* The documents this index subscribes to i.e. the documents which are updated using event listeners.
*
- * @return string[] Class names of DocumentInterface instances
+ * @return class-string[] Class names of DocumentInterface instances
*
* @see DocumentInterface
*/
diff --git a/src/Messenger/Handler/AbstractRefreshHandler.php b/src/Messenger/Handler/AbstractRefreshHandler.php
new file mode 100644
index 0000000..471ba44
--- /dev/null
+++ b/src/Messenger/Handler/AbstractRefreshHandler.php
@@ -0,0 +1,45 @@
+ $className */
+ $className = $message->className;
+
+ try {
+ $element = $className::getById($message->id);
+ } catch (\Throwable) {
+ throw new PimcoreElementNotFoundException($message->id, $message->className);
+ }
+
+ if (!$element instanceof AbstractElement) {
+ // The element in question was deleted so we need a skeleton.
+ /** @var TModel $element */
+ $element = new ($className)();
+ $element->setId($message->id);
+
+ if ($element instanceof Concrete) {
+ $element->setPublished(false);
+ }
+ }
+
+ if ($element === null) {
+ throw new PimcoreElementNotFoundException($message->id, $message->className);
+ }
+
+ return $element;
+ }
+}
diff --git a/src/Messenger/Handler/RefreshElementHandler.php b/src/Messenger/Handler/RefreshElementHandler.php
new file mode 100644
index 0000000..db251c7
--- /dev/null
+++ b/src/Messenger/Handler/RefreshElementHandler.php
@@ -0,0 +1,28 @@
+
+ */
+class RefreshElementHandler extends AbstractRefreshHandler
+{
+ public function __construct(
+ private readonly PropagateChanges $propagateChanges,
+ ) {}
+
+ public function __invoke(RefreshElement $message): void
+ {
+ $element = $this->resolveElement($message);
+
+ $this->propagateChanges->handle($element);
+ }
+}
diff --git a/src/Messenger/Handler/RefreshElementInIndexHandler.php b/src/Messenger/Handler/RefreshElementInIndexHandler.php
new file mode 100644
index 0000000..1e715fd
--- /dev/null
+++ b/src/Messenger/Handler/RefreshElementInIndexHandler.php
@@ -0,0 +1,37 @@
+
+ */
+class RefreshElementInIndexHandler extends AbstractRefreshHandler
+{
+ public function __construct(
+ private readonly PropagateChanges $propagateChanges,
+ private readonly LockService $lockService,
+ private readonly IndexRepository $indexRepository,
+ ) {}
+
+ public function __invoke(RefreshElementInIndex $message): void
+ {
+ $index = $this->indexRepository->flattenedGet($message->index);
+ $element = $this->resolveElement($message);
+
+ if ($index->usesBlueGreenIndices() && !$this->lockService->getIndexingLock($index)->acquire()) {
+ $this->propagateChanges->handleIndex($element, $index, $index->getBlueGreenInactiveElasticaIndex());
+ }
+
+ $this->propagateChanges->handleIndex($element, $index);
+ }
+}
diff --git a/src/Messenger/Message/AbstractRefresh.php b/src/Messenger/Message/AbstractRefresh.php
new file mode 100644
index 0000000..1c9d33e
--- /dev/null
+++ b/src/Messenger/Message/AbstractRefresh.php
@@ -0,0 +1,20 @@
+ */
+ public string $className;
+ public int $id;
+
+ protected function setElement(ElementInterface $element): void
+ {
+ $this->className = $element::class;
+ $this->id = $element->getId() ?? throw new \InvalidArgumentException('Pimcore ID is null.');
+ }
+}
diff --git a/src/Messenger/Message/RefreshElement.php b/src/Messenger/Message/RefreshElement.php
new file mode 100644
index 0000000..b4785d0
--- /dev/null
+++ b/src/Messenger/Message/RefreshElement.php
@@ -0,0 +1,15 @@
+setElement($element);
+ }
+}
diff --git a/src/Messenger/Message/RefreshElementInIndex.php b/src/Messenger/Message/RefreshElementInIndex.php
new file mode 100644
index 0000000..ebdcdd0
--- /dev/null
+++ b/src/Messenger/Message/RefreshElementInIndex.php
@@ -0,0 +1,17 @@
+setElement($element);
+ }
+}
diff --git a/src/Repository/AbstractRepository.php b/src/Repository/AbstractRepository.php
index 3cad617..c9d6358 100644
--- a/src/Repository/AbstractRepository.php
+++ b/src/Repository/AbstractRepository.php
@@ -8,6 +8,8 @@
/**
* @template TItem
+ *
+ * @internal
*/
abstract class AbstractRepository
{
diff --git a/src/Repository/ConfigurationRepository.php b/src/Repository/ConfigurationRepository.php
index 112a76c..d08e277 100644
--- a/src/Repository/ConfigurationRepository.php
+++ b/src/Repository/ConfigurationRepository.php
@@ -6,34 +6,32 @@
use Symfony\Component\DependencyInjection\ParameterBag\ContainerBagInterface;
+/**
+ * @internal
+ */
class ConfigurationRepository
{
public function __construct(
private readonly ContainerBagInterface $containerBag,
) {}
- /**
- * @return array{host:string,port:int}|string
- */
- public function getClient(): array|string
+ public function getClientDsn(): string
{
- $config = $this->containerBag->get('valantic_elastica_bridge')['client'];
-
- return $config['dsn'] !== null && $config['dsn'] !== ''
- ? $config['dsn']
- : [
- 'host' => $config['host'],
- 'port' => $config['port'],
- ];
+ return $this->containerBag->get('valantic_elastica_bridge')['client']['dsn'];
}
- public function getAddSentryBreadcrumbs(): bool
+ public function shouldAddSentryBreadcrumbs(): bool
{
- return $this->containerBag->get('valantic_elastica_bridge')['client']['addSentryBreadcrumbs'];
+ return $this->containerBag->get('valantic_elastica_bridge')['client']['should_add_sentry_breadcrumbs'];
}
public function getIndexingLockTimeout(): int
{
return $this->containerBag->get('valantic_elastica_bridge')['indexing']['lock_timeout'];
}
+
+ public function shouldSkipFailingDocuments(): bool
+ {
+ return $this->containerBag->get('valantic_elastica_bridge')['indexing']['should_skip_failing_documents'];
+ }
}
diff --git a/src/Repository/DocumentRepository.php b/src/Repository/DocumentRepository.php
index 8427d6c..9c664ab 100644
--- a/src/Repository/DocumentRepository.php
+++ b/src/Repository/DocumentRepository.php
@@ -6,5 +6,9 @@
use Valantic\ElasticaBridgeBundle\Document\DocumentInterface;
-/** @extends AbstractRepository */
+/**
+ * @extends AbstractRepository
+ *
+ * @internal
+ */
class DocumentRepository extends AbstractRepository {}
diff --git a/src/Repository/IndexRepository.php b/src/Repository/IndexRepository.php
index 99da9ff..5ad406f 100644
--- a/src/Repository/IndexRepository.php
+++ b/src/Repository/IndexRepository.php
@@ -4,16 +4,21 @@
namespace Valantic\ElasticaBridgeBundle\Repository;
+use Valantic\ElasticaBridgeBundle\Exception\Repository\ItemNotFoundInRepositoryException;
use Valantic\ElasticaBridgeBundle\Index\IndexInterface;
use Valantic\ElasticaBridgeBundle\Index\TenantAwareInterface;
-/** @extends AbstractRepository */
+/**
+ * @extends AbstractRepository
+ *
+ * @internal
+ */
class IndexRepository extends AbstractRepository
{
/**
* @return \Generator
*/
- public function flattened(): \Generator
+ public function flattenedAll(): \Generator
{
foreach ($this->all() as $indexConfig) {
if ($indexConfig instanceof TenantAwareInterface) {
@@ -28,4 +33,15 @@ public function flattened(): \Generator
}
}
}
+
+ public function flattenedGet(string $key): IndexInterface
+ {
+ foreach ($this->flattenedAll() as $candidateKey => $index) {
+ if ($candidateKey === $key) {
+ return $index;
+ }
+ }
+
+ throw new ItemNotFoundInRepositoryException($key);
+ }
}
diff --git a/src/Resources/config/pimcore/config.yaml b/src/Resources/config/pimcore/config.yaml
new file mode 100644
index 0000000..53b51b1
--- /dev/null
+++ b/src/Resources/config/pimcore/config.yaml
@@ -0,0 +1,2 @@
+imports:
+ - { resource: messenger.yaml }
diff --git a/src/Resources/config/pimcore/messenger.yaml b/src/Resources/config/pimcore/messenger.yaml
new file mode 100644
index 0000000..66393aa
--- /dev/null
+++ b/src/Resources/config/pimcore/messenger.yaml
@@ -0,0 +1,8 @@
+framework:
+ messenger:
+ enabled: true
+ transports:
+ elastica_bridge_index: 'doctrine://default?queue_name=elastica_bridge_index'
+ routing:
+ Valantic\ElasticaBridgeBundle\Messenger\Message\RefreshElement: elastica_bridge_index
+ Valantic\ElasticaBridgeBundle\Messenger\Message\RefreshElementInIndex: elastica_bridge_index
diff --git a/src/Resources/config/services.yml b/src/Resources/config/services.yaml
similarity index 86%
rename from src/Resources/config/services.yml
rename to src/Resources/config/services.yaml
index acdf8b6..0de954b 100644
--- a/src/Resources/config/services.yml
+++ b/src/Resources/config/services.yaml
@@ -28,3 +28,8 @@ services:
Valantic\ElasticaBridgeBundle\Repository\DocumentRepository:
arguments:
- !tagged_iterator valantic.elastica_bridge.document
+
+ Valantic\ElasticaBridgeBundle\Messenger\Handler\:
+ resource: '../../Messenger/Handler/*'
+ tags:
+ - { name: messenger.message_handler }
diff --git a/src/Service/LockService.php b/src/Service/LockService.php
new file mode 100644
index 0000000..425832b
--- /dev/null
+++ b/src/Service/LockService.php
@@ -0,0 +1,29 @@
+lockFactory
+ ->createLock(
+ sprintf('%s:indexing:%s', self::LOCK_PREFIX, $indexConfig->getName()),
+ ttl: $this->configurationRepository->getIndexingLockTimeout()
+ );
+ }
+}
diff --git a/src/Service/PropagateChanges.php b/src/Service/PropagateChanges.php
index 24b97ae..a5fcb54 100644
--- a/src/Service/PropagateChanges.php
+++ b/src/Service/PropagateChanges.php
@@ -5,10 +5,13 @@
namespace Valantic\ElasticaBridgeBundle\Service;
use Elastica\Exception\NotFoundException;
+use Elastica\Index;
use Pimcore\Model\DataObject\AbstractObject;
use Pimcore\Model\Element\AbstractElement;
+use Symfony\Component\Messenger\MessageBusInterface;
use Valantic\ElasticaBridgeBundle\Document\DocumentInterface;
use Valantic\ElasticaBridgeBundle\Index\IndexInterface;
+use Valantic\ElasticaBridgeBundle\Messenger\Message\RefreshElementInIndex;
use Valantic\ElasticaBridgeBundle\Repository\IndexRepository;
class PropagateChanges
@@ -16,6 +19,7 @@ class PropagateChanges
public function __construct(
private readonly IndexRepository $indexRepository,
private readonly DocumentHelper $documentHelper,
+ private readonly MessageBusInterface $messageBus,
) {}
/**
@@ -26,15 +30,27 @@ public function __construct(
*/
public function handle(AbstractElement $element): void
{
- $indices = $this->matchingIndicesForElement($this->indexRepository->flattened(), $element);
+ $indices = $this->matchingIndicesForElement($this->indexRepository->flattenedAll(), $element);
foreach ($indices as $index) {
- $this->handleIndex($element, $index);
+ $this->messageBus->dispatch(new RefreshElementInIndex($element, $index->getName()));
}
}
- private function handleIndex(AbstractElement $element, IndexInterface $index): void
- {
+ public function handleIndex(
+ AbstractElement $element,
+ IndexInterface $index,
+ ?Index $elasticaIndex = null,
+ ): void {
+ $this->doHandleIndex($element, $index, $elasticaIndex ?? $index->getElasticaIndex());
+ }
+
+ private function doHandleIndex(
+ AbstractElement $element,
+ IndexInterface $index,
+ Index $elasticaIndex,
+ ): void {
+ // TODO: actually use $elasticaIndex
$document = $index->findDocumentInstanceByPimcore($element);
if (!$document instanceof DocumentInterface) {
@@ -52,20 +68,20 @@ private function handleIndex(AbstractElement $element, IndexInterface $index): v
return;
}
- $isPresent = $this->isIdInIndex($document::getElasticsearchId($element), $index);
+ $isPresent = $this->isIdInIndex($document::getElasticsearchId($element), $elasticaIndex);
if ($document->shouldIndex($element)) {
if ($isPresent) {
- $this->updateElementInIndex($element, $index, $document);
+ $this->updateElementInIndex($element, $elasticaIndex, $document);
}
if (!$isPresent) {
- $this->addElementToIndex($element, $index, $document);
+ $this->addElementToIndex($element, $elasticaIndex, $document);
}
}
if ($isPresent && !$document->shouldIndex($element)) {
- $this->deleteElementFromIndex($element, $index, $document);
+ $this->deleteElementFromIndex($element, $elasticaIndex, $document);
}
$this->documentHelper->resetTenantIfNeeded($document, $index);
@@ -76,11 +92,11 @@ private function handleIndex(AbstractElement $element, IndexInterface $index): v
*/
private function addElementToIndex(
AbstractElement $element,
- IndexInterface $index,
+ Index $index,
DocumentInterface $document,
): void {
$document = $this->documentHelper->elementToDocument($document, $element);
- $index->getElasticaIndex()->addDocument($document);
+ $index->addDocument($document);
}
/**
@@ -88,12 +104,12 @@ private function addElementToIndex(
*/
private function updateElementInIndex(
AbstractElement $element,
- IndexInterface $index,
+ Index $index,
DocumentInterface $document,
): void {
$document = $this->documentHelper->elementToDocument($document, $element);
// updateDocument() allows partial updates, hence the full replace here
- $index->getElasticaIndex()->addDocument($document);
+ $index->addDocument($document);
}
/**
@@ -101,11 +117,11 @@ private function updateElementInIndex(
*/
private function deleteElementFromIndex(
AbstractElement $element,
- IndexInterface $index,
+ Index $index,
DocumentInterface $document,
): void {
$elasticsearchId = $document::getElasticsearchId($element);
- $index->getElasticaIndex()->deleteById($elasticsearchId);
+ $index->deleteById($elasticsearchId);
}
/**
@@ -134,10 +150,10 @@ private function matchingIndicesForElement(
/**
* Checks whether a given ID is in an index.
*/
- private function isIdInIndex(string $id, IndexInterface $index): bool
+ private function isIdInIndex(string $id, Index $index): bool
{
try {
- $index->getElasticaIndex()->getDocument($id);
+ $index->getDocument($id);
} catch (NotFoundException) {
return false;
}
diff --git a/src/Util/ElasticsearchResponse.php b/src/Util/ElasticsearchResponse.php
new file mode 100644
index 0000000..92a1d14
--- /dev/null
+++ b/src/Util/ElasticsearchResponse.php
@@ -0,0 +1,28 @@
+