Supports mongodb persistence driver

Author: Klyachin Andrey

DependencyInjection/Configuration.php

@@ -19,6 +19,9 @@ public function getConfigTreeBuilder()
                 ->scalarNode('context')->defaultValue('enqueue.transport.context')->cannotBeEmpty()->end()
                 ->arrayNode('doctrine')
                     ->children()
+                        ->scalarNode('driver')->isRequired()->cannotBeEmpty()
+                            ->validate()->ifNotInArray(['orm', 'mongodb'])->thenInvalid('Invalid driver')
+                        ->end()->end()
                         ->arrayNode('queue_listeners')
                             ->prototype('array')
                                 ->addDefaultsIfNotSet()

DependencyInjection/EnqueueElasticaExtension.php

@@ -40,5 +40,23 @@ public function load(array $configs, ContainerBuilder $container)
                 ;
             }
         }
+
+        $serviceId       = 'enqueue_elastica.doctrine.sync_index_with_object_change_processor';
+        $managerRegistry = $this->getManagerRegistry($config['doctrine']['driver']);
+        $container
+            ->getDefinition($serviceId)
+            ->replaceArgument(0, new Reference($managerRegistry));
+    }
+
+    private function getManagerRegistry(string $driver): string
+    {
+        switch ($driver) {
+            case 'orm':
+                return 'doctrine';
+                break;
+            case 'mongodb':
+                return 'doctrine_mongodb';
+                break;
+        }
     }
 }

Doctrine/Queue/SyncIndexWithObjectChangeProcessor.php

@@ -10,7 +10,7 @@
 use Interop\Queue\PsrContext;
 use Interop\Queue\PsrMessage;
 use Interop\Queue\PsrProcessor;
-use Symfony\Bridge\Doctrine\RegistryInterface;
+use Doctrine\Common\Persistence\ManagerRegistry;
 
 final class SyncIndexWithObjectChangeProcessor implements PsrProcessor, CommandSubscriberInterface, QueueSubscriberInterface
 {
@@ -31,11 +31,11 @@ final class SyncIndexWithObjectChangeProcessor implements PsrProcessor, CommandS
     private $indexable;
 
     /**
-     * @var RegistryInterface
+     * @var ManagerRegistry
      */
     private $doctrine;
 
-    public function __construct(RegistryInterface $doctrine, PersisterRegistry $persisterRegistry, IndexableInterface $indexable)
+    public function __construct(ManagerRegistry $doctrine, PersisterRegistry $persisterRegistry, IndexableInterface $indexable)
     {
         $this->persisterRegistry = $persisterRegistry;
         $this->indexable = $indexable;

Resources/config/services.yml

@@ -12,7 +12,7 @@ services:
       class: 'Enqueue\ElasticaBundle\Doctrine\Queue\SyncIndexWithObjectChangeProcessor'
       public: true
       arguments:
-          - '@doctrine'
+          - null
           - '@fos_elastica.persister_registry'
           - '@fos_elastica.indexable'
       tags: