Merge pull request #38 from maciejmrozinski/queue-index-alias-support

Support index alias when populating and consuming

Author: makasim

DependencyInjection/EnqueueElasticaExtension.php

@@ -34,6 +34,7 @@ public function load(array $configs, ContainerBuilder $container)
         $container->register('enqueue_elastica.populate_processor', PopulateProcessor::class)
             ->addArgument(new Reference('fos_elastica.pager_provider_registry'))
             ->addArgument(new Reference('fos_elastica.pager_persister_registry'))
+            ->addArgument(new Reference('fos_elastica.index_manager'))
 
             ->addTag('enqueue.command_subscriber', ['client' => $transport])
             ->addTag('enqueue.transport.processor', ['transport' => $transport])
@@ -49,6 +50,7 @@ public function load(array $configs, ContainerBuilder $container)
             ->addArgument(new Reference('enqueue_elastica.context'))
             ->addArgument(new Reference('fos_elastica.persister_registry'))
             ->addArgument(new Reference('event_dispatcher'))
+            ->addArgument(new Reference('fos_elastica.index_manager'))
 
             ->addTag('fos_elastica.pager_persister', ['persisterName' => 'queue'])
             ->setPublic(true)

Persister/QueuePagerPersister.php

@@ -3,6 +3,7 @@
 
 use Enqueue\ElasticaBundle\Queue\Commands;
 use Enqueue\Util\JSON;
+use FOS\ElasticaBundle\Index\IndexManager;
 use FOS\ElasticaBundle\Persister\Event\PostAsyncInsertObjectsEvent;
 use FOS\ElasticaBundle\Persister\Event\PostPersistEvent;
 use FOS\ElasticaBundle\Persister\Event\PrePersistEvent;
@@ -28,11 +29,21 @@ final class QueuePagerPersister implements PagerPersisterInterface
      */
     private $dispatcher;
 
-    public function __construct(Context $context, PersisterRegistry $registry, EventDispatcherInterface $dispatcher)
-    {
+    /**
+     * @var IndexManager
+     */
+    private $indexManager;
+
+    public function __construct(
+        Context $context,
+        PersisterRegistry $registry,
+        EventDispatcherInterface $dispatcher,
+        IndexManager $indexManager
+    ) {
         $this->context = $context;
         $this->dispatcher = $dispatcher;
         $this->registry = $registry;
+        $this->indexManager = $indexManager;
     }
 
     /**
@@ -42,15 +53,21 @@ public function insert(PagerInterface $pager, array $options = array())
     {
         $pager->setMaxPerPage(empty($options['max_per_page']) ? 100 : $options['max_per_page']);
 
-        $options = array_replace([
+        $defaultOptions = [
             'max_per_page' => $pager->getMaxPerPage(),
             'first_page' => $pager->getCurrentPage(),
             'last_page' => $pager->getNbPages(),
             'populate_queue' => Commands::POPULATE,
             'populate_reply_queue' => null,
             'reply_receive_timeout' => 5000, // ms
-            'limit_overall_reply_time' => 180, // sec
-        ], $options);
+            'limit_overall_reply_time' => 180 // sec
+        ];
+        $index = $this->indexManager->getIndex($options['indexName']);
+        if ($index->getName() !== $index->getOriginalName()) {
+            $defaultOptions['realIndexName'] = $index->getName();
+        }
+
+        $options = array_replace($defaultOptions, $options);
 
         $pager->setCurrentPage($options['first_page']);
 

Queue/PopulateProcessor.php

@@ -4,6 +4,7 @@
 use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Consumption\QueueSubscriberInterface;
 use Enqueue\Consumption\Result;
+use FOS\ElasticaBundle\Index\IndexManager;
 use FOS\ElasticaBundle\Persister\InPlacePagerPersister;
 use FOS\ElasticaBundle\Persister\PagerPersisterRegistry;
 use FOS\ElasticaBundle\Provider\PagerProviderRegistry;
@@ -18,12 +19,16 @@ final class PopulateProcessor implements Processor, CommandSubscriberInterface,
 
     private $pagerPersisterRegistry;
 
+    private $indexManager;
+
     public function __construct(
         PagerProviderRegistry $pagerProviderRegistry,
-        PagerPersisterRegistry $pagerPersisterRegistry
+        PagerPersisterRegistry $pagerPersisterRegistry,
+        IndexManager $indexManager
     ) {
         $this->pagerPersisterRegistry = $pagerPersisterRegistry;
         $this->pagerProviderRegistry = $pagerProviderRegistry;
+        $this->indexManager = $indexManager;
     }
 
     public function process(Message $message, Context $context): Result
@@ -53,6 +58,10 @@ public function process(Message $message, Context $context): Result
             $options['first_page'] = $data['page'];
             $options['last_page'] = $data['page'];
 
+            if (isset($options['realIndexName'])) {
+                $this->indexManager->getIndex($options['indexName'])->overrideName($options['realIndexName']);
+            }
+
             $provider = $this->pagerProviderRegistry->getProvider($options['indexName']);
             $pager = $provider->provide($options);
             $pager->setMaxPerPage($options['max_per_page']);