Make build processor registyry pass support multi transports.

makasim · makasim · commit bc571c23886b · 2018-10-21T22:07:19.000+03:00
diff --git a/pkg/enqueue-bundle/EnqueueBundle.php b/pkg/enqueue-bundle/EnqueueBundle.php
@@ -23,8 +23,8 @@ class EnqueueBundle extends Bundle
     public function build(ContainerBuilder $container): void
     {
         //transport passes
-        $container->addCompilerPass(new BuildConsumptionExtensionsPass('default'));
-        $container->addCompilerPass(new BuildProcessorRegistryPass('default'));
+        $container->addCompilerPass(new BuildConsumptionExtensionsPass());
+        $container->addCompilerPass(new BuildProcessorRegistryPass());
 
         //client passes
         $container->addCompilerPass(new BuildClientConsumptionExtensionsPass('default'));
diff --git a/pkg/enqueue/Symfony/Client/DependencyInjection/ClientFactory.php b/pkg/enqueue/Symfony/Client/DependencyInjection/ClientFactory.php
@@ -21,7 +21,6 @@
 use Enqueue\Consumption\QueueConsumer;
 use Enqueue\Rpc\RpcFactory;
 use Enqueue\Symfony\ContainerProcessorRegistry;
-use Enqueue\Symfony\DependencyInjection\FormatClientNameTrait;
 use Interop\Queue\Context;
 use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
diff --git a/pkg/enqueue/Symfony/Client/DependencyInjection/FormatClientNameTrait.php b/pkg/enqueue/Symfony/Client/DependencyInjection/FormatClientNameTrait.php
@@ -1,6 +1,6 @@
 <?php
 
-namespace Enqueue\Symfony\DependencyInjection;
+namespace Enqueue\Symfony\Client\DependencyInjection;
 
 use Symfony\Component\DependencyInjection\ContainerInterface;
 use Symfony\Component\DependencyInjection\Reference;
diff --git a/pkg/enqueue/Symfony/DependencyInjection/BuildConsumptionExtensionsPass.php b/pkg/enqueue/Symfony/DependencyInjection/BuildConsumptionExtensionsPass.php
@@ -8,55 +8,60 @@
 
 final class BuildConsumptionExtensionsPass implements CompilerPassInterface
 {
-    /**
-     * @var string
-     */
-    private $name;
+    use FormatTransportNameTrait;
 
-    public function __construct(string $transportName)
-    {
-        if (empty($transportName)) {
-            throw new \InvalidArgumentException('The name could not be empty.');
-        }
-
-        $this->name = $transportName;
-    }
+    protected $name;
 
     public function process(ContainerBuilder $container): void
     {
-        $extensionsId = sprintf('enqueue.transport.%s.consumption_extensions', $this->name);
-        if (false == $container->hasDefinition($extensionsId)) {
-            return;
+        if (false == $container->hasParameter('enqueue.transports')) {
+            throw new \LogicException('The "enqueue.transports" parameter must be set.');
         }
 
-        $tags = $container->findTaggedServiceIds('enqueue.transport.consumption_extension');
+        $names = $container->getParameter('enqueue.transports');
 
-        $groupByPriority = [];
-        foreach ($tags as $serviceId => $tagAttributes) {
-            foreach ($tagAttributes as $tagAttribute) {
-                $transport = $tagAttribute['transport'] ?? 'default';
+        foreach ($names as $name) {
+            $this->name = $name;
 
-                if ($transport !== $this->name && 'all' !== $transport) {
-                    continue;
-                }
+            $extensionsId = $this->format('consumption_extensions');
+            if (false == $container->hasDefinition($extensionsId)) {
+                throw new \LogicException(sprintf('Service "%s" not found', $extensionsId));
+            }
+
+            $tags = $container->findTaggedServiceIds('enqueue.transport.consumption_extension');
+
+            $groupByPriority = [];
+            foreach ($tags as $serviceId => $tagAttributes) {
+                foreach ($tagAttributes as $tagAttribute) {
+                    $transport = $tagAttribute['transport'] ?? 'default';
 
-                $priority = (int) ($tagAttribute['priority'] ?? 0);
+                    if ($transport !== $this->name && 'all' !== $transport) {
+                        continue;
+                    }
 
-                $groupByPriority[$priority][] = new Reference($serviceId);
+                    $priority = (int) ($tagAttribute['priority'] ?? 0);
+
+                    $groupByPriority[$priority][] = new Reference($serviceId);
+                }
             }
-        }
 
-        krsort($groupByPriority, SORT_NUMERIC);
+            krsort($groupByPriority, SORT_NUMERIC);
 
-        $flatExtensions = [];
-        foreach ($groupByPriority as $extension) {
-            $flatExtensions = array_merge($flatExtensions, $extension);
+            $flatExtensions = [];
+            foreach ($groupByPriority as $extension) {
+                $flatExtensions = array_merge($flatExtensions, $extension);
+            }
+
+            $extensionsService = $container->getDefinition($extensionsId);
+            $extensionsService->replaceArgument(0, array_merge(
+                $extensionsService->getArgument(0),
+                $flatExtensions
+            ));
         }
+    }
 
-        $extensionsService = $container->getDefinition($extensionsId);
-        $extensionsService->replaceArgument(0, array_merge(
-            $extensionsService->getArgument(0),
-            $flatExtensions
-        ));
+    protected function getName(): string
+    {
+        return $this->name;
     }
 }
diff --git a/pkg/enqueue/Symfony/DependencyInjection/BuildProcessorRegistryPass.php b/pkg/enqueue/Symfony/DependencyInjection/BuildProcessorRegistryPass.php
@@ -9,44 +9,49 @@
 
 final class BuildProcessorRegistryPass implements CompilerPassInterface
 {
-    /**
-     * @var string
-     */
-    private $name;
+    use FormatTransportNameTrait;
 
-    public function __construct(string $transportName)
-    {
-        if (empty($transportName)) {
-            throw new \InvalidArgumentException('The name could not be empty.');
-        }
-
-        $this->name = $transportName;
-    }
+    protected $name;
 
     public function process(ContainerBuilder $container): void
     {
-        $processorRegistryId = sprintf('enqueue.transport.%s.processor_registry', $this->name);
-        if (false == $container->hasDefinition($processorRegistryId)) {
-            return;
+        if (false == $container->hasParameter('enqueue.transports')) {
+            throw new \LogicException('The "enqueue.transports" parameter must be set.');
         }
 
-        $tag = 'enqueue.transport.processor';
-        $map = [];
-        foreach ($container->findTaggedServiceIds($tag) as $serviceId => $tagAttributes) {
-            foreach ($tagAttributes as $tagAttribute) {
-                $transport = $tagAttribute['transport'] ?? 'default';
+        $names = $container->getParameter('enqueue.transports');
 
-                if ($transport !== $this->name && 'all' !== $transport) {
-                    continue;
-                }
+        foreach ($names as $name) {
+            $this->name = $name;
+
+            $processorRegistryId = $this->format('processor_registry');
+            if (false == $container->hasDefinition($processorRegistryId)) {
+                throw new \LogicException(sprintf('Service "%s" not found', $processorRegistryId));
+            }
 
-                $processor = $tagAttribute['processor'] ?? $serviceId;
+            $tag = 'enqueue.transport.processor';
+            $map = [];
+            foreach ($container->findTaggedServiceIds($tag) as $serviceId => $tagAttributes) {
+                foreach ($tagAttributes as $tagAttribute) {
+                    $transport = $tagAttribute['transport'] ?? 'default';
 
-                $map[$processor] = new Reference($serviceId);
+                    if ($transport !== $this->name && 'all' !== $transport) {
+                        continue;
+                    }
+
+                    $processor = $tagAttribute['processor'] ?? $serviceId;
+
+                    $map[$processor] = new Reference($serviceId);
+                }
             }
+
+            $registry = $container->getDefinition($processorRegistryId);
+            $registry->setArgument(0, ServiceLocatorTagPass::register($container, $map, $processorRegistryId));
         }
+    }
 
-        $registry = $container->getDefinition($processorRegistryId);
-        $registry->setArgument(0, ServiceLocatorTagPass::register($container, $map, $processorRegistryId));
+    protected function getName(): string
+    {
+        return $this->name;
     }
 }
diff --git a/pkg/enqueue/Symfony/DependencyInjection/FormatTransportNameTrait.php b/pkg/enqueue/Symfony/DependencyInjection/FormatTransportNameTrait.php
@@ -2,10 +2,25 @@
 
 namespace Enqueue\Symfony\DependencyInjection;
 
+use Symfony\Component\DependencyInjection\ContainerInterface;
+use Symfony\Component\DependencyInjection\Reference;
+
 trait FormatTransportNameTrait
 {
     abstract protected function getName(): string;
 
+    private function reference(string $serviceName, $invalidBehavior = ContainerInterface::EXCEPTION_ON_INVALID_REFERENCE): Reference
+    {
+        return new Reference($this->format($serviceName), $invalidBehavior);
+    }
+
+    private function parameter(string $serviceName): string
+    {
+        $fullName = $this->format($serviceName, false);
+
+        return "%$fullName%";
+    }
+
     private function format(string $serviceName, $parameter = false): string
     {
         $pattern = 'enqueue.transport.%s.'.$serviceName;
diff --git a/pkg/enqueue/Tests/Symfony/DependencyInjection/BuildConsumptionExtensionsPassTest.php b/pkg/enqueue/Tests/Symfony/DependencyInjection/BuildConsumptionExtensionsPassTest.php
@@ -25,28 +25,29 @@ public function testShouldBeFinal()
         $this->assertClassFinal(BuildConsumptionExtensionsPass::class);
     }
 
-    public function testCouldBeConstructedWithName()
+    public function testCouldBeConstructedWithoutArguments()
     {
-        $pass = new BuildConsumptionExtensionsPass('aName');
-
-        $this->assertAttributeSame('aName', 'name', $pass);
+        new BuildConsumptionExtensionsPass();
     }
 
-    public function testThrowIfNameEmptyOnConstruct()
+    public function testThrowIfEnqueueTransportsParameterNotSet()
     {
-        $this->expectException(\InvalidArgumentException::class);
-        $this->expectExceptionMessage('The name could not be empty.');
-        new BuildConsumptionExtensionsPass('');
+        $pass = new BuildConsumptionExtensionsPass();
+
+        $this->expectException(\LogicException::class);
+        $this->expectExceptionMessage('The "enqueue.transports" parameter must be set.');
+        $pass->process(new ContainerBuilder());
     }
 
-    public function testShouldDoNothingIfExtensionsServiceIsNotRegistered()
+    public function testThrowsIfNoConsumptionExtensionsServiceFoundForConfiguredTransport()
     {
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['foo', 'bar']);
 
-        //guard
-        $this->assertFalse($container->hasDefinition('enqueue.transport.aName.consumption_extensions'));
+        $pass = new BuildConsumptionExtensionsPass();
 
-        $pass = new BuildConsumptionExtensionsPass('aName');
+        $this->expectException(\LogicException::class);
+        $this->expectExceptionMessage('Service "enqueue.transport.foo.consumption_extensions" not found');
         $pass->process($container);
     }
 
@@ -56,6 +57,7 @@ public function testShouldRegisterTransportExtension()
         $extensions->addArgument([]);
 
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['aName']);
         $container->setDefinition('enqueue.transport.aName.consumption_extensions', $extensions);
 
         $container->register('aFooExtension', ExtensionInterface::class)
@@ -65,7 +67,7 @@ public function testShouldRegisterTransportExtension()
             ->addTag('enqueue.transport.consumption_extension', ['transport' => 'aName'])
         ;
 
-        $pass = new BuildConsumptionExtensionsPass('aName');
+        $pass = new BuildConsumptionExtensionsPass();
         $pass->process($container);
 
         $this->assertInternalType('array', $extensions->getArgument(0));
@@ -81,6 +83,7 @@ public function testShouldIgnoreOtherTransportExtensions()
         $extensions->addArgument([]);
 
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['aName']);
         $container->setDefinition('enqueue.transport.aName.consumption_extensions', $extensions);
 
         $container->register('aFooExtension', ExtensionInterface::class)
@@ -90,7 +93,7 @@ public function testShouldIgnoreOtherTransportExtensions()
             ->addTag('enqueue.transport.consumption_extension', ['transport' => 'anotherName'])
         ;
 
-        $pass = new BuildConsumptionExtensionsPass('aName');
+        $pass = new BuildConsumptionExtensionsPass();
         $pass->process($container);
 
         $this->assertInternalType('array', $extensions->getArgument(0));
@@ -105,6 +108,7 @@ public function testShouldAddExtensionIfTransportAll()
         $extensions->addArgument([]);
 
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['aName']);
         $container->setDefinition('enqueue.transport.aName.consumption_extensions', $extensions);
 
         $container->register('aFooExtension', ExtensionInterface::class)
@@ -114,7 +118,7 @@ public function testShouldAddExtensionIfTransportAll()
             ->addTag('enqueue.transport.consumption_extension', ['transport' => 'anotherName'])
         ;
 
-        $pass = new BuildConsumptionExtensionsPass('aName');
+        $pass = new BuildConsumptionExtensionsPass();
         $pass->process($container);
 
         $this->assertInternalType('array', $extensions->getArgument(0));
@@ -129,6 +133,7 @@ public function testShouldTreatTagsWithoutTransportAsDefaultTransport()
         $extensions->addArgument([]);
 
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['default']);
         $container->setDefinition('enqueue.transport.default.consumption_extensions', $extensions);
 
         $container->register('aFooExtension', ExtensionInterface::class)
@@ -138,7 +143,7 @@ public function testShouldTreatTagsWithoutTransportAsDefaultTransport()
             ->addTag('enqueue.transport.consumption_extension')
         ;
 
-        $pass = new BuildConsumptionExtensionsPass('default');
+        $pass = new BuildConsumptionExtensionsPass();
         $pass->process($container);
 
         $this->assertInternalType('array', $extensions->getArgument(0));
@@ -151,6 +156,7 @@ public function testShouldTreatTagsWithoutTransportAsDefaultTransport()
     public function testShouldOrderExtensionsByPriority()
     {
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['default']);
 
         $extensions = new Definition();
         $extensions->addArgument([]);
@@ -168,7 +174,7 @@ public function testShouldOrderExtensionsByPriority()
         $extension->addTag('enqueue.transport.consumption_extension', ['priority' => 2]);
         $container->setDefinition('baz_extension', $extension);
 
-        $pass = new BuildConsumptionExtensionsPass('default');
+        $pass = new BuildConsumptionExtensionsPass();
         $pass->process($container);
 
         $orderedExtensions = $extensions->getArgument(0);
@@ -182,6 +188,7 @@ public function testShouldOrderExtensionsByPriority()
     public function testShouldAssumePriorityZeroIfPriorityIsNotSet()
     {
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['default']);
 
         $extensions = new Definition();
         $extensions->addArgument([]);
@@ -199,7 +206,7 @@ public function testShouldAssumePriorityZeroIfPriorityIsNotSet()
         $extension->addTag('enqueue.transport.consumption_extension', ['priority' => -1]);
         $container->setDefinition('baz_extension', $extension);
 
-        $pass = new BuildConsumptionExtensionsPass('default');
+        $pass = new BuildConsumptionExtensionsPass();
         $pass->process($container);
 
         $orderedExtensions = $extensions->getArgument(0);
@@ -219,6 +226,7 @@ public function testShouldMergeWithAddedPreviously()
         ]);
 
         $container = new ContainerBuilder();
+        $container->setParameter('enqueue.transports', ['aName']);
         $container->setDefinition('enqueue.transport.aName.consumption_extensions', $extensions);
 
         $container->register('aFooExtension', ExtensionInterface::class)
@@ -228,7 +236,7 @@ public function testShouldMergeWithAddedPreviously()
             ->addTag('enqueue.transport.consumption_extension')
         ;
 
-        $pass = new BuildConsumptionExtensionsPass('aName');
+        $pass = new BuildConsumptionExtensionsPass();
         $pass->process($container);
 
         $this->assertInternalType('array', $extensions->getArgument(0));
diff --git a/pkg/enqueue/Tests/Symfony/DependencyInjection/BuildProcessorRegistryPassTest.php b/pkg/enqueue/Tests/Symfony/DependencyInjection/BuildProcessorRegistryPassTest.php
