Feature/symfony messenger (#86)

Co-authored-by: Stefan Hagspiel <info@solverat.com>
dachcom-digital · Apr 3, 2024 · da046ad · da046ad
1 parent ce9751f
commit da046ad
Show file tree

Hide file tree

Showing 31 changed files with 793 additions and 569 deletions.
diff --git a/README.md b/README.md
@@ -8,11 +8,12 @@
 [![PhpStan](https://img.shields.io/github/actions/workflow/status/dachcom-digital/pimcore-dynamic-search/.github/workflows/php-stan.yml?branch=master&style=flat-square&logo=github&label=phpstan%20level%204)](https://github.com/dachcom-digital/pimcore-dynamic-search/actions?query=workflow%3A"PHP+Stan"+branch%3Amaster)
 
 ### Release Plan
-| Release | Supported Pimcore Versions | Supported Symfony Versions | Release Date | Maintained      | Branch                                                                    |
-|---------|----------------------------|----------------------------|--------------|-----------------|---------------------------------------------------------------------------|
-| **3.x** | `11.0`                     | `^6.2`                     | 28.09.2023   | Feature Branch  | master                                                                    |
-| **2.x** | `10.0` - `10.6`            | `^5.4`                     | 19.12.2021   | No              | [2.x](https://github.com/dachcom-digital/pimcore-dynamic-search/tree/2.x) |
-| **1.x** | `6.6` - `6.9`              | `^4.4`                     | 18.04.2021   | No              | [1.x](https://github.com/dachcom-digital/pimcore-dynamic-search/tree/1.x) |
+| Release | Supported Pimcore Versions | Supported Symfony Versions | Release Date | Maintained     | Branch                                                                    |
+|---------|----------------------------|----------------------------|--------------|----------------|---------------------------------------------------------------------------|
+| **4.x** | `11.0`                     | `^6.2`                     | --           | Feature Branch | master                                                                    |
+| **3.x** | `11.0`                     | `^6.2`                     | 28.09.2023   | Bugfixes       | [3.x](https://github.com/dachcom-digital/pimcore-dynamic-search/tree/3.x) |
+| **2.x** | `10.0` - `10.6`            | `^5.4`                     | 19.12.2021   | No             | [2.x](https://github.com/dachcom-digital/pimcore-dynamic-search/tree/2.x) |
+| **1.x** | `6.6` - `6.9`              | `^4.4`                     | 18.04.2021   | No             | [1.x](https://github.com/dachcom-digital/pimcore-dynamic-search/tree/1.x) |
 
 ## Introduction
 The Dynamic Search Bundle allows you to redefine your search strategy. 
@@ -34,7 +35,7 @@ There are several data- and index providers available:
 
 ```json
 "require" : {
-    "dachcom-digital/dynamic-search" : "~3.0.0"
+    "dachcom-digital/dynamic-search" : "~4.0.0"
 }
 ```
 
@@ -46,6 +47,7 @@ return [
 ```
 
 - Execute: `$ bin/console pimcore:bundle:install DynamicSearchBundle`
+- Execute optionally: `$ bin/console messenger:setup-transports`
 
 ## Upgrading
 - Execute: `$ bin/console doctrine:migrations:migrate --prefix 'DynamicSearchBundle\Migrations'`
@@ -61,6 +63,14 @@ dynamic_search_frontend:
     resource: '@DynamicSearchBundle/config/pimcore/routing/frontend_routing.yaml'
 ```
 
+## Start Queue Worker
+```
+$ bin/console messenger:consume dynamic_search_queue
+```
+
+Read more details about the queue worker and the recommended setup [here](docs/01_DispatchWorkflow.md#queue-worker).
+
+
 ## Dispatch Dynamic Search
 After you've added [a definition](docs/0_ExampleSetup.md), you're ready to start the engine.
 Always use the verbose `-v` flag, otherwise you won't get any process information about the ongoing data / index providing process.

diff --git a/UPGRADE.md b/UPGRADE.md
@@ -1,25 +1,12 @@
 # Upgrade Notes
 
-## 3.0.1
-- [BUGFIX] allow empty strings being submitted [#81](https://github.com/dachcom-digital/pimcore-dynamic-search/issues/81)
-- Use `microtime` for envelope queue to avoid wrong processing order. Execute `bin/console dynamic-search:check-queue'` before updating to this version
+## Migrating from Version 3.x to Version 4.x
 
-## Migrating from Version 2.x to Version 3.0.
-
-### Breaking Changes
-- filters will be passed to the view as associative array, having the filter names as keys (#59)  
-  filter names must match the pattern `/^[a-z0-9_\-\.]+$/i`
-
-### Global Changes
-- Recommended folder structure by symfony adopted
-- [ROUTE] Route include changed from `DynamicSearchBundle/Resources/config/pimcore/routing/frontend_routing.yml` to `DynamicSearchBundle/config/pimcore/routing/frontend_routing.yaml`
-
-### Fixes
--- 
-
-### New Features
---
+### Breaking Changes 
+- The queue is now based on symfony messenger. ([#83](https://github.com/dachcom-digital/pimcore-dynamic-search/issues/83)).   
+  Execute `bin/console messenger:setup-transports`  
+  read [how to setup the queue worker](docs/01_DispatchWorkflow.md#queue-worker)
 
 ***
 
-DynamicSearch 2.x Upgrade Notes: https://github.com/dachcom-digital/pimcore-dynamic-search/blob/2.x/UPGRADE.md
+DynamicSearch 3.x Upgrade Notes: https://github.com/dachcom-digital/pimcore-dynamic-search/blob/3.x/UPGRADE.md
diff --git a/composer.json b/composer.json
@@ -25,7 +25,7 @@
     },
     "extra": {
         "branch-alias": {
-            "dev-master": "3.x-dev"
+            "dev-master": "4.x-dev"
         },
         "pimcore": {
             "bundles": [

diff --git a/config/pimcore/config.yaml b/config/pimcore/config.yaml
@@ -1,3 +1,6 @@
+imports:
+    - { resource: messenger.yaml }
+
 doctrine_migrations:
     migrations_paths:
         'DynamicSearchBundle\Migrations': '@DynamicSearchBundle/src/Migrations'
@@ -28,4 +31,4 @@ monolog:
                 VERBOSITY_VERY_VERBOSE: DEBUG
                 VERBOSITY_DEBUG: DEBUG
             channels: ['dynamic_search']
-            formatter: dynamic_search.log.formatter.console.provider
+            formatter: dynamic_search.log.formatter.console.provider
diff --git a/config/pimcore/messenger.yaml b/config/pimcore/messenger.yaml
@@ -0,0 +1,21 @@
+parameters:
+    dynamic_search.queue.table_name: messenger_dynamic_search
+
+framework:
+    messenger:
+        transports:
+            dynamic_search_queue:
+                dsn: 'microsecs-doctrine://default'
+                options:
+                    table_name: '%dynamic_search.queue.table_name%'
+        routing:
+            DynamicSearchBundle\Queue\Message\QueueResourceMessage: dynamic_search_queue
+            DynamicSearchBundle\Queue\Message\ProcessResourceMessage: dynamic_search_queue
+        buses:
+            dynamic_search.bus: ~
+
+services:
+    DynamicSearchBundle\Queue\Transport\ExtendedDoctrineTransportFactory:
+        arguments:
+            - '@Doctrine\Persistence\ConnectionRegistry'
+        tags: [ messenger.transport_factory ]
diff --git a/config/services/command.yaml b/config/services/command.yaml
@@ -8,7 +8,3 @@ services:
     DynamicSearchBundle\Command\SearchCommand:
         tags:
             - { name: console.command }
-
-    DynamicSearchBundle\Command\QueuedDataCommand:
-        tags:
-            - { name: console.command }
diff --git a/config/services/maintenance.yaml b/config/services/maintenance.yaml
@@ -4,7 +4,3 @@ services:
         autowire: true
         autoconfigure: true
         public: false
-
-    DynamicSearchBundle\EventListener\Maintenance\QueuedDataTask:
-        tags:
-            - { name: pimcore.maintenance.task, type: dynamic_search_check_modified_data }
diff --git a/config/services/manager.yaml b/config/services/manager.yaml
@@ -27,4 +27,6 @@ services:
     DynamicSearchBundle\Manager\NormalizerManager: ~
 
     DynamicSearchBundle\Manager\QueueManagerInterface: '@DynamicSearchBundle\Manager\QueueManager'
-    DynamicSearchBundle\Manager\QueueManager: ~
+    DynamicSearchBundle\Manager\QueueManager:
+        arguments:
+            $tableName: '%dynamic_search.queue.table_name%'
diff --git a/config/services/queue.yaml b/config/services/queue.yaml
@@ -5,8 +5,20 @@ services:
         autoconfigure: true
         public: false
 
+    DynamicSearchBundle\Queue\MessageHandler\QueuedResourcesHandler:
+        arguments:
+            $messageBus: '@dynamic_search.bus'
+        tags:
+            - { name: messenger.message_handler, bus: dynamic_search.bus }
+
+    DynamicSearchBundle\Queue\MessageHandler\ProcessResourceHandler:
+        tags:
+            - { name: messenger.message_handler, bus: dynamic_search.bus }
+
     DynamicSearchBundle\Queue\DataCollectorInterface: '@DynamicSearchBundle\Queue\DataCollector'
-    DynamicSearchBundle\Queue\DataCollector: ~
+    DynamicSearchBundle\Queue\DataCollector:
+        arguments:
+            $messageBus: '@dynamic_search.bus'
 
     DynamicSearchBundle\Queue\DataProcessorInterface: '@DynamicSearchBundle\Queue\DataProcessor'
     DynamicSearchBundle\Queue\DataProcessor: ~
diff --git a/docs/01_DispatchWorkflow.md b/docs/01_DispatchWorkflow.md
@@ -23,24 +23,21 @@ At every modification/deletion event of every pimcore element,
 DynamicSearch will validate this element by calling [ResourceValidator](./40_ResourceValidator.md). 
 If the resource is still present, DynamicSearch creates an `Envelope` which will be passed to a dedicated queue.
 
-#### Element Processor
-Another maintenance task processes that queue (Interval depends on your maintenance cronjob).
-If available, the `Envelope` will be submitted to the index provider. 
+#### Queue Worker
+The queue is based on [symfony messenger](https://symfony.com/doc/current/messenger.html).
+Resources which are dispatched to the index are put into the queue and will be processed by the worker asynchronously.
+The queue is processed in batches.
+If available and resource validation passes, the transformed resource will be submitted to the index provider. 
 
-> DynamicSearch will sort envelopes by creation date. 
-> If you're updating your element multiple times before the element processor kicks in,
-> only the latest envelope will be used. This allows us to save some trees. 
-
-#### Element Process Command
-There is a secret command which allows you to dispatch the queue processor immediately. 
-This comes in handy if you're debugging your application:
-
-> Use the `-v flag to output some log information`
+To start the queue worker, execute
 
 ```bash
-$ bin/console dynamic-search:check-queue -v
+$ bin/console messenger:consume dynamic_search_queue
 ```
 
+> It is highly recommended to set up the worker to be always running with supervisor or systemd.  
+> More details about how to set up the queue worker are found [here](https://symfony.com/doc/current/messenger.html#consuming-messages-running-the-worker).
+
 #### Inheritance
 Inheritance is unknown to DynamicSearch. If you're updating an object, that very object will be transmitted to the queue.
 But in some cases, you're working with variants and those should get updated too.

diff --git a/phpstan.neon b/phpstan.neon
@@ -7,4 +7,9 @@ parameters:
     symfony:
         container_xml_path: %currentWorkingDirectory%/var/cache/test/TestKernelTestDebugContainer.xml
         constant_hassers: false
-    ignoreErrors:
+    excludePaths:
+        analyse:
+            - src/Queue/Transport/ExtendedDoctrineConnection.php
+    ignoreErrors:
+        -
+            message: "#^Method DynamicSearchBundle\\\\Queue\\\\MessageHandler\\\\(.*)Handler\\:\\:(getBatchSize|process)\\(\\) is unused\\.$#"
diff --git a/src/Command/QueuedDataCommand.php b/src/Command/QueuedDataCommand.php
diff --git a/src/Context/ContextDefinitionInterface.php b/src/Context/ContextDefinitionInterface.php
@@ -36,7 +36,7 @@ interface ContextDefinitionInterface
     /*
      * Allowed dispatch types for queue
      */
-    public const ALLOWED_QUEUE_DISPATCH_TYPES = ['insert', 'update', 'delete'];
+    public const ALLOWED_QUEUE_DISPATCH_TYPES = ['index', 'insert', 'update', 'delete'];
 
     public function getName(): string;
 

diff --git a/src/EventListener/Maintenance/QueuedDataTask.php b/src/EventListener/Maintenance/QueuedDataTask.php
diff --git a/src/EventSubscriber/DataProcessingEventSubscriber.php b/src/EventSubscriber/DataProcessingEventSubscriber.php
@@ -3,12 +3,13 @@
 namespace DynamicSearchBundle\EventSubscriber;
 
 use DynamicSearchBundle\Builder\ContextDefinitionBuilderInterface;
+use DynamicSearchBundle\Context\ContextDefinitionInterface;
 use DynamicSearchBundle\DynamicSearchEvents;
 use DynamicSearchBundle\Event\NewDataEvent;
 use DynamicSearchBundle\Logger\LoggerInterface;
 use DynamicSearchBundle\Processor\ResourceModificationProcessorInterface;
 use DynamicSearchBundle\Provider\DataProviderInterface;
-use DynamicSearchBundle\Validator\ResourceValidatorInterface;
+use DynamicSearchBundle\Queue\DataCollectorInterface;
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 
 class DataProcessingEventSubscriber implements EventSubscriberInterface
@@ -17,7 +18,7 @@ public function __construct(
         protected LoggerInterface $logger,
         protected ContextDefinitionBuilderInterface $contextDefinitionBuilder,
         protected ResourceModificationProcessorInterface $resourceModificationProcessor,
-        protected ResourceValidatorInterface $resourceValidator
+        protected DataCollectorInterface $dataCollector
     ) {
     }
 
@@ -32,32 +33,19 @@ public function dispatchResourceModification(NewDataEvent $event): void
     {
         $contextDefinition = $this->contextDefinitionBuilder->buildContextDefinition($event->getContextName(), $event->getContextDispatchType());
 
-        try {
-            // validate and allow rewriting resource based on current data behaviour
-            $isImmutableResource = $event->getProviderBehaviour() === DataProviderInterface::PROVIDER_BEHAVIOUR_SINGLE_DISPATCH;
-            $resourceCandidate = $this->resourceValidator->validateResource($event->getContextName(), $event->getContextDispatchType(), false, $isImmutableResource, $event->getData());
-        } catch (\Throwable $e) {
-            $this->logger->error(
-                sprintf(
-                    'Error while validate resource candidate: %s',
-                    $e->getMessage()), $contextDefinition->getDataProviderName(), $event->getContextName()
-            );
-
-            return;
-        }
-
-        if ($resourceCandidate->getResource() === null) {
-            $this->logger->debug(
-                sprintf(
-                    'Resource has been removed due to validation. Skipping...'),
-                $contextDefinition->getDataProviderName(), $contextDefinition->getName()
-            );
-
-            return;
-        }
-
         if ($event->getProviderBehaviour() === DataProviderInterface::PROVIDER_BEHAVIOUR_FULL_DISPATCH) {
-            $this->resourceModificationProcessor->process($contextDefinition, $resourceCandidate->getResource());
+            // data collector add to queue
+            $this->dataCollector->addToContextQueue(
+                $contextDefinition->getName(),
+                ContextDefinitionInterface::CONTEXT_DISPATCH_TYPE_INDEX,
+                $event->getData(),
+                [
+                    'resourceValidation' => [
+                        'unknownResource' => false,
+                        'immutableResource' => false
+                    ]
+                ]
+            );
         } elseif ($event->getProviderBehaviour() === DataProviderInterface::PROVIDER_BEHAVIOUR_SINGLE_DISPATCH) {
             $this->resourceModificationProcessor->processByResourceMeta($contextDefinition, $event->getResourceMeta(), $event->getData());
         } else {