Merge pull request #9 from ne0h12/master

Supports mongodb persistence driver

Author: makasim

DependencyInjection/Configuration.php

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

DependencyInjection/EnqueueElasticaExtension.php

@@ -23,6 +23,7 @@ public function load(array $configs, ContainerBuilder $container)
 
         $container->setAlias('enqueue_elastica.context', $config['context']);
 
+        $doctrineDriver = $config['doctrine']['driver'];
         if (false == empty($config['doctrine']['queue_listeners'])) {
             foreach ($config['doctrine']['queue_listeners'] as $listenerConfig) {
                 $listenerId = sprintf(
@@ -36,9 +37,49 @@ public function load(array $configs, ContainerBuilder $container)
                     ->addArgument(new Reference('enqueue_elastica.context'))
                     ->addArgument($listenerConfig['model_class'])
                     ->addArgument($listenerConfig)
-                    ->addTag('doctrine.event_subscriber', ['connection' => $listenerConfig['connection']])
+                    ->addTag($this->getEventSubscriber($doctrineDriver), ['connection' => $listenerConfig['connection']])
                 ;
             }
         }
+
+        $serviceId       = 'enqueue_elastica.doctrine.sync_index_with_object_change_processor';
+        $managerRegistry = $this->getManagerRegistry($doctrineDriver);
+        $container
+            ->getDefinition($serviceId)
+            ->replaceArgument(0, new Reference($managerRegistry));
+    }
+
+    /**
+     * @param string $driver
+     *
+     * @return string
+     */
+    private function getManagerRegistry(string $driver): string
+    {
+        switch ($driver) {
+            case 'mongodb':
+                return 'doctrine_mongodb';
+                break;
+            case 'orm':
+            default:
+                return 'doctrine';
+        }
+    }
+
+    /**
+     * @param string $driver
+     *
+     * @return string
+     */
+    private function getEventSubscriber(string $driver): string
+    {
+        switch ($driver) {
+            case 'mongodb':
+                return 'doctrine_mongodb.odm.event_subscriber';
+                break;
+            case 'orm':
+            default:
+                return 'doctrine.event_subscriber';
+        }
     }
 }

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;