[bundle][dbal] Use doctrine bundle configured connections

pkg/enqueue-bundle/EnqueueBundle.php

@@ -5,6 +5,7 @@
 use Enqueue\AsyncEventDispatcher\DependencyInjection\AsyncEventDispatcherExtension;
 use Enqueue\AsyncEventDispatcher\DependencyInjection\AsyncEventsPass;
 use Enqueue\AsyncEventDispatcher\DependencyInjection\AsyncTransformersPass;
+use Enqueue\Doctrine\DoctrineSchemaCompilerPass;
 use Enqueue\Symfony\Client\DependencyInjection\AnalyzeRouteCollectionPass;
 use Enqueue\Symfony\Client\DependencyInjection\BuildClientExtensionsPass;
 use Enqueue\Symfony\Client\DependencyInjection\BuildCommandSubscriberRoutesPass as BuildClientCommandSubscriberRoutesPass;
@@ -39,5 +40,7 @@ public function build(ContainerBuilder $container): void
             $container->addCompilerPass(new AsyncEventsPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
             $container->addCompilerPass(new AsyncTransformersPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
         }
+
+        $container->addCompilerPass(new DoctrineSchemaCompilerPass());
     }
 }

pkg/enqueue-bundle/Tests/Functional/App/config/custom-config.yml

@@ -14,6 +14,14 @@ framework:
     router:          { resource: '%kernel.root_dir%/config/routing.yml' }
     default_locale:  '%locale%'
 
+doctrine:
+    dbal:
+        connections:
+            custom:
+                url:   "%env(DOCTRINE_DSN)%"
+                driver: pdo_mysql
+                charset:  UTF8
+
 services:
     test_enqueue.client.default.driver:
         alias: 'enqueue.client.default.driver'

pkg/enqueue-bundle/Tests/Functional/UseCasesTest.php

@@ -136,6 +136,13 @@ public function provideEnqueueConfigs()
                 'transport' => getenv('MONGO_DSN'),
             ],
         ]];
+
+        yield 'doctrine' => [[
+            'default' => [
+                'transport' => 'doctrine://custom',
+            ],
+        ]];
+
 //
 //        yield 'gps' => [[
 //            'transport' => [

pkg/enqueue/Doctrine/DoctrineConnectionFactoryFactory.php

@@ -0,0 +1,54 @@
+<?php
+
+namespace Enqueue\Doctrine;
+
+use Enqueue\ConnectionFactoryFactoryInterface;
+use Enqueue\Dbal\ManagerRegistryConnectionFactory;
+use Enqueue\Dsn\Dsn;
+use Interop\Queue\ConnectionFactory;
+use Symfony\Bridge\Doctrine\RegistryInterface;
+
+class DoctrineConnectionFactoryFactory implements ConnectionFactoryFactoryInterface
+{
+    /**
+     * @var RegistryInterface
+     */
+    private $doctrine;
+
+    /**
+     * @var ConnectionFactoryFactoryInterface
+     */
+    private $fallbackFactory;
+
+    public function __construct(RegistryInterface $doctrine, ConnectionFactoryFactoryInterface $fallbackFactory)
+    {
+        $this->doctrine = $doctrine;
+        $this->fallbackFactory = $fallbackFactory;
+    }
+
+    public function create($config): ConnectionFactory
+    {
+        if (is_string($config)) {
+            $config = ['dsn' => $config];
+        }
+
+        if (false == is_array($config)) {
+            throw new \InvalidArgumentException('The config must be either array or DSN string.');
+        }
+
+        if (false == array_key_exists('dsn', $config)) {
+            throw new \InvalidArgumentException('The config must have dsn key set.');
+        }
+
+        $dsn = Dsn::parseFirst($config['dsn']);
+
+        if ('doctrine' === $dsn->getScheme()) {
+            $config = $dsn->getQuery();
+            $config['connection_name'] = $dsn->getHost();
+
+            return new ManagerRegistryConnectionFactory($this->doctrine, $config);
+        }
+
+        return $this->fallbackFactory->create($config);
+    }
+}

pkg/enqueue/Doctrine/DoctrineDriverFactory.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace Enqueue\Doctrine;
+
+use Enqueue\Client\Config;
+use Enqueue\Client\Driver\DbalDriver;
+use Enqueue\Client\DriverFactoryInterface;
+use Enqueue\Client\DriverInterface;
+use Enqueue\Client\RouteCollection;
+use Enqueue\Dsn\Dsn;
+use Interop\Queue\ConnectionFactory;
+
+class DoctrineDriverFactory implements DriverFactoryInterface
+{
+    /**
+     * @var DriverFactoryInterface
+     */
+    private $fallbackFactory;
+
+    public function __construct(DriverFactoryInterface $fallbackFactory)
+    {
+        $this->fallbackFactory = $fallbackFactory;
+    }
+
+    public function create(ConnectionFactory $factory, Config $config, RouteCollection $collection): DriverInterface
+    {
+        $dsn = $config->getTransportOption('dsn');
+
+        if (empty($dsn)) {
+            throw new \LogicException('This driver factory relies on dsn option from transport config. The option is empty or not set.');
+        }
+
+        $dsn = Dsn::parseFirst($dsn);
+
+        if ('doctrine' === $dsn->getScheme()) {
+            return new DbalDriver($factory->createContext(), $config, $collection);
+        }
+
+        return $this->fallbackFactory->create($factory, $config, $collection);
+    }
+}

pkg/enqueue/Doctrine/DoctrineSchemaCompilerPass.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace Enqueue\Doctrine;
+
+use Enqueue\Symfony\Client\DependencyInjection\ClientFactory;
+use Enqueue\Symfony\DependencyInjection\TransportFactory;
+use Enqueue\Symfony\DiUtils;
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Reference;
+
+class DoctrineSchemaCompilerPass implements CompilerPassInterface
+{
+    public function process(ContainerBuilder $container)
+    {
+        if (false === $container->hasDefinition('doctrine')) {
+            return;
+        }
+
+        foreach ($container->getParameter('enqueue.transports') as $name) {
+            $diUtils = DiUtils::create(TransportFactory::MODULE, $name);
+
+            $container->register($diUtils->format('connection_factory_factory.outer'), DoctrineConnectionFactoryFactory::class)
+                ->setDecoratedService($diUtils->format('connection_factory_factory'), $diUtils->format('connection_factory_factory.inner'))
+                ->addArgument(new Reference('doctrine'))
+                ->addArgument(new Reference($diUtils->format('connection_factory_factory.inner')))
+            ;
+        }
+
+        foreach ($container->getParameter('enqueue.clients') as $name) {
+            $diUtils = DiUtils::create(ClientFactory::MODULE, $name);
+
+            $container->register($diUtils->format('driver_factory.outer'), DoctrineDriverFactory::class)
+                ->setDecoratedService($diUtils->format('driver_factory'), $diUtils->format('driver_factory.inner'))
+                ->addArgument(new Reference($diUtils->format('driver_factory.inner')))
+            ;
+        }
+    }
+}