add configuration.

Author: makasim

DependencyInjection/Configuration.php

@@ -0,0 +1,31 @@
+<?php
+
+namespace Enqueue\Bundle\DependencyInjection;
+
+use Symfony\Component\Config\Definition\Builder\TreeBuilder;
+use Symfony\Component\Config\Definition\ConfigurationInterface;
+
+class Configuration implements ConfigurationInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function getConfigTreeBuilder()
+    {
+        $tb = new TreeBuilder();
+        $rootNode = $tb->root('enqueue_elastica');
+        $rootNode
+            ->children()
+                ->arrayNode('doctrine_queue_listeners')
+                    ->prototype('array')
+                        ->addDefaultsIfNotSet()
+                        ->children()
+                            ->booleanNode('on_insert')->defaultTrue()->end()
+                            ->booleanNode('on_update')->defaultTrue()->end()
+                            ->booleanNode('on_remove')->defaultTrue()->end()
+                            ->scalarNode('index_name')->isRequired()->cannotBeEmpty()->end()
+                            ->scalarNode('type_name')->isRequired()->cannotBeEmpty()->end()
+                            ->scalarNode('model_class')->isRequired()->cannotBeEmpty()->end()
+        ;
+    }
+}

DependencyInjection/EnqueueElasticaExtension.php

@@ -2,9 +2,11 @@
 
 namespace Enqueue\ElasticaBundle\DependencyInjection;
 
+use Enqueue\ElasticaBundle\Doctrine\SyncIndexWithObjectChangeListener;
 use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Loader\YamlFileLoader;
+use Symfony\Component\DependencyInjection\Reference;
 use Symfony\Component\HttpKernel\DependencyInjection\Extension;
 
 class EnqueueElasticaExtension extends Extension
@@ -14,7 +16,27 @@ class EnqueueElasticaExtension extends Extension
      */
     public function load(array $configs, ContainerBuilder $container)
     {
+        $configuration = $this->getConfiguration($configs, $container);
+        $config = $this->processConfiguration($configuration, $configs);
+
         $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('services.yml');
+
+        if (isset($config['doctrine_queue_listeners'])) {
+            foreach ($config['doctrine_queue_listeners'] as $listenerConfig) {
+                $listenerId = sprintf(
+                    'enqueue_elastica.doctrine_queue_listener.%s.%s',
+                    $listenerConfig['index_name'],
+                    $listenerConfig['type_name']
+                );
+
+                $container->register($listenerId, SyncIndexWithObjectChangeListener::class)
+                    ->addArgument(new Reference('enqueue.transport.context'))
+                    ->addArgument($listenerConfig['modelClass'])
+                    ->addArgument($listenerConfig)
+                    ->addTag('doctrine.event_subscriber')
+                ;
+            }
+        }
     }
 }

Queue/SyncIndexWithDoctrineORMObjectChangeProcessor.php

@@ -4,7 +4,7 @@
 use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Consumption\QueueSubscriberInterface;
 use Enqueue\Util\JSON;
-use FOS\ElasticaBundle\Persister\ObjectPersisterInterface;
+use FOS\ElasticaBundle\Persister\PersisterRegistry;
 use FOS\ElasticaBundle\Provider\IndexableInterface;
 use Interop\Queue\PsrContext;
 use Interop\Queue\PsrMessage;
@@ -14,9 +14,9 @@
 final class SyncIndexWithDoctrineORMObjectChangeProcessor implements PsrProcessor, CommandSubscriberInterface, QueueSubscriberInterface
 {
     /**
-     * @var ObjectPersisterInterface
+     * @var PersisterRegistry
      */
-    private $objectPersister;
+    private $persisterRegistry;
 
     /**
      * @var IndexableInterface
@@ -28,9 +28,9 @@ final class SyncIndexWithDoctrineORMObjectChangeProcessor implements PsrProcesso
      */
     private $doctrine;
 
-    public function __construct(RegistryInterface $doctrine, ObjectPersisterInterface $objectPersister, IndexableInterface $indexable)
+    public function __construct(RegistryInterface $doctrine, PersisterRegistry $persisterRegistry, IndexableInterface $indexable)
     {
-        $this->objectPersister = $objectPersister;
+        $this->persisterRegistry = $persisterRegistry;
         $this->indexable = $indexable;
         $this->doctrine = $doctrine;
     }
@@ -46,42 +46,43 @@ public function process(PsrMessage $message, PsrContext $context)
             return self::REJECT;
         }
 
-        $indexName = $data['indexName'];
-        $typeName = $data['typeName'];
+        $index = $data['indexName'];
+        $type = $data['typeName'];
 
-        $objectRepository = $this->doctrine->getManagerForClass($data['modelClass'])->getRepository($data['modelClass']);
+        $repository = $this->doctrine->getManagerForClass($data['modelClass'])->getRepository($data['modelClass']);
+        $persister = $this->persisterRegistry->getPersister($index, $type);
 
         switch ($data['action']) {
             case 'update':
-                if (false == $object = $objectRepository->find($data['id'])) {
-                    $this->objectPersister->deleteById($data['id']);
+                if (false == $object = $repository->find($data['id'])) {
+                    $persister->deleteById($data['id']);
 
                     return self::REJECT;
                 }
 
-                if ($this->objectPersister->handlesObject($object)) {
-                    if ($this->indexable->isObjectIndexable($indexName, $typeName, $object)) {
-                        $this->objectPersister->replaceOne($object);
+                if ($persister->handlesObject($object)) {
+                    if ($this->indexable->isObjectIndexable($index, $type, $object)) {
+                        $persister->replaceOne($object);
                     } else {
-                        $this->objectPersister->deleteOne($object);
+                        $persister->deleteOne($object);
                     }
                 }
 
                 break;
             case 'insert':
-                if (false == $object = $objectRepository->find($data['id'])) {
-                    $this->objectPersister->deleteById($data['id']);
+                if (false == $object = $repository->find($data['id'])) {
+                    $persister->deleteById($data['id']);
 
                     return self::REJECT;
                 }
 
-                if ($this->objectPersister->handlesObject($object) && $this->indexable->isObjectIndexable($indexName, $typeName, $object)) {
-                    $this->objectPersister->insertOne($object);
+                if ($persister->handlesObject($object) && $this->indexable->isObjectIndexable($index, $type, $object)) {
+                    $persister->insertOne($object);
                 }
 
                 break;
             case 'delete':
-                $this->objectPersister->deleteById($data['id']);
+                $persister->deleteById($data['id']);
 
                 break;
             default:

Resources/config/services.yml

@@ -7,6 +7,15 @@ services:
       tags:
           - { name: "enqueue.client.processor" }
 
+  enqueue_elastica.sync_index_with_doctrine_orm_object_change_processor:
+      class: 'Enqueue\ElasticaBundle\Queue\SyncIndexWithDoctrineORMObjectChangeProcessor'
+      arguments:
+          - '@doctrine'
+          - '@fos_elastica.persister_registry'
+          - '@fos_elastica.indexable'
+      tags:
+          - { name: "enqueue.client.processor" }
+
   enqueue_elastica.purge_populate_queue_listener:
       class: 'Enqueue\ElasticaBundle\Persister\Listener\PurgePopulateQueueListener'
       arguments: