Add standalone setup broker feature.

makasim · makasim · commit a10725b8d368 · 2018-12-20T15:02:43.000+02:00
diff --git a/pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php b/pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php
@@ -59,6 +59,7 @@ public function load(array $configs, ContainerBuilder $container): void
             $transportFactory->buildContext($container, []);
             $transportFactory->buildQueueConsumer($container, $modules['consumption']);
             $transportFactory->buildRpcClient($container, []);
+            $transportFactory->buildSetupBroker($container, []);
 
             // client
             if (isset($modules['client'])) {
diff --git a/pkg/enqueue-bundle/EnqueueBundle.php b/pkg/enqueue-bundle/EnqueueBundle.php
@@ -14,6 +14,7 @@
 use Enqueue\Symfony\Client\DependencyInjection\BuildTopicSubscriberRoutesPass as BuildClientTopicSubscriberRoutesPass;
 use Enqueue\Symfony\DependencyInjection\BuildConsumptionExtensionsPass;
 use Enqueue\Symfony\DependencyInjection\BuildProcessorRegistryPass;
+use Enqueue\Symfony\DependencyInjection\BuildSetupBrokerPass;
 use Symfony\Component\DependencyInjection\Compiler\PassConfig;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\HttpKernel\Bundle\Bundle;
@@ -25,6 +26,7 @@ public function build(ContainerBuilder $container): void
         //transport passes
         $container->addCompilerPass(new BuildConsumptionExtensionsPass());
         $container->addCompilerPass(new BuildProcessorRegistryPass());
+        $container->addCompilerPass(new BuildSetupBrokerPass());
 
         //client passes
         $container->addCompilerPass(new BuildClientConsumptionExtensionsPass());
diff --git a/pkg/enqueue-bundle/Resources/config/services.yml b/pkg/enqueue-bundle/Resources/config/services.yml
@@ -15,6 +15,15 @@ services:
         tags:
             - { name: 'console.command' }
 
+    enqueue.transport.setup_broker:
+        class: 'Enqueue\Symfony\SetupBrokerCommand'
+        arguments:
+            - '@enqueue.locator'
+            - '%enqueue.default_transport%'
+            - 'enqueue.transport.%s.setup_broker'
+        tags:
+            - { name: 'console.command' }
+
     enqueue.client.consume_command:
         class: 'Enqueue\Symfony\Client\ConsumeCommand'
         arguments:
diff --git a/pkg/enqueue/SetupBroker/ChainSetupBroker.php b/pkg/enqueue/SetupBroker/ChainSetupBroker.php
@@ -0,0 +1,33 @@
+<?php
+namespace Enqueue\SetupBroker;
+
+use Psr\Log\LoggerInterface;
+
+class ChainSetupBroker implements SetupBrokerInterface
+{
+    /**
+     * @var SetupBrokerInterface[]
+     */
+    private $setupBrokers;
+
+    /**
+     * @param SetupBrokerInterface[] $setupBrokers
+     */
+    public function __construct(array $setupBrokers)
+    {
+        $this->setupBrokers = [];
+
+        array_walk($setupBrokers, function (SetupBrokerInterface $setupBroker) {
+            $this->setupBrokers[] = $setupBroker;
+        });
+    }
+
+    public function setupBroker(LoggerInterface $logger = null): void
+    {
+        foreach ($this->setupBrokers as $setupBroker) {
+            $logger->info(get_class($setupBroker));
+
+            $setupBroker->setupBroker($logger);
+        }
+    }
+}
diff --git a/pkg/enqueue/SetupBroker/SetupBrokerInterface.php b/pkg/enqueue/SetupBroker/SetupBrokerInterface.php
@@ -0,0 +1,9 @@
+<?php
+namespace Enqueue\SetupBroker;
+
+use Psr\Log\LoggerInterface;
+
+interface SetupBrokerInterface
+{
+    public function setupBroker(LoggerInterface $logger = null): void;
+}
diff --git a/pkg/enqueue/Symfony/DependencyInjection/BuildSetupBrokerPass.php b/pkg/enqueue/Symfony/DependencyInjection/BuildSetupBrokerPass.php
@@ -0,0 +1,47 @@
+<?php
+
+namespace Enqueue\Symfony\DependencyInjection;
+
+use Enqueue\Symfony\DiUtils;
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Reference;
+
+final class BuildSetupBrokerPass implements CompilerPassInterface
+{
+    public function process(ContainerBuilder $container): void
+    {
+        if (false == $container->hasParameter('enqueue.transports')) {
+            throw new \LogicException('The "enqueue.transports" parameter must be set.');
+        }
+
+        $names = $container->getParameter('enqueue.transports');
+        $defaultName = $container->getParameter('enqueue.default_transport');
+
+        foreach ($names as $name) {
+            $diUtils = DiUtils::create(TransportFactory::MODULE, $name);
+
+            $setupBrokerId = $diUtils->format('setup_broker');
+            if (false == $container->hasDefinition($setupBrokerId)) {
+                throw new \LogicException(sprintf('Service "%s" not found', $setupBrokerId));
+            }
+
+            $tag = 'enqueue.transport.setup_broker';
+            $map = [];
+            foreach ($container->findTaggedServiceIds($tag) as $serviceId => $tagAttributes) {
+                foreach ($tagAttributes as $tagAttribute) {
+                    $transport = $tagAttribute['transport'] ?? $defaultName;
+
+                    if ($transport !== $name && 'all' !== $transport) {
+                        continue;
+                    }
+
+                    $map[] = new Reference($serviceId);
+                }
+            }
+
+            $setupBroker = $container->getDefinition($setupBrokerId);
+            $setupBroker->setArgument(0, $map);
+        }
+    }
+}
diff --git a/pkg/enqueue/Symfony/DependencyInjection/TransportFactory.php b/pkg/enqueue/Symfony/DependencyInjection/TransportFactory.php
@@ -11,6 +11,7 @@
 use Enqueue\Resources;
 use Enqueue\Rpc\RpcClient;
 use Enqueue\Rpc\RpcFactory;
+use Enqueue\SetupBroker\ChainSetupBroker;
 use Enqueue\Symfony\ContainerProcessorRegistry;
 use Enqueue\Symfony\DiUtils;
 use Interop\Queue\ConnectionFactory;
@@ -243,6 +244,15 @@ public function buildRpcClient(ContainerBuilder $container, array $config): void
         }
     }
 
+    public function buildSetupBroker(ContainerBuilder $container, array $config): void
+    {
+        $container->register($this->diUtils->format('setup_broker'), ChainSetupBroker::class)
+            ->addArgument([])
+        ;
+
+        $this->addServiceToLocator($container, 'setup_broker');
+    }
+
     private function assertServiceExists(ContainerBuilder $container, string $serviceId): void
     {
         if (false == $container->hasDefinition($serviceId)) {
diff --git a/pkg/enqueue/Symfony/SetupBrokerCommand.php b/pkg/enqueue/Symfony/SetupBrokerCommand.php
@@ -0,0 +1,76 @@
+<?php
+
+namespace Enqueue\Symfony;
+
+use Enqueue\Client\DriverInterface;
+use Enqueue\SetupBroker\SetupBrokerInterface;
+use Psr\Container\ContainerInterface;
+use Psr\Container\NotFoundExceptionInterface;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Logger\ConsoleLogger;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class SetupBrokerCommand extends Command
+{
+    protected static $defaultName = 'enqueue:transport:setup-broker';
+
+    /**
+     * @var ContainerInterface
+     */
+    private $container;
+
+    /**
+     * @var string
+     */
+    private $defaultTransport;
+
+    /**
+     * @var string
+     */
+    private $setupBrokerIdPattern;
+
+    public function __construct(ContainerInterface $container, string $defaultTransport, string $setupBrokerIdPattern = 'enqueue.transport.%s.setup_broker')
+    {
+        $this->container = $container;
+        $this->defaultTransport = $defaultTransport;
+        $this->setupBrokerIdPattern = $setupBrokerIdPattern;
+
+        parent::__construct(static::$defaultName);
+    }
+
+    protected function configure(): void
+    {
+        $this
+            ->setDescription('Setup broker. Configure the broker, creates queues, topics and so on.')
+            ->addOption('transport', 't', InputOption::VALUE_OPTIONAL, 'The transport to consume messages from.', $this->defaultTransport)
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): ?int
+    {
+        $transport = $input->getOption('transport');
+
+        $logger = new ConsoleLogger($output);
+
+        try {
+            $setupBroker = $this->getSetupBroker($transport);
+
+            $logger->info(get_class($setupBroker));
+
+            $setupBroker->setupBroker();
+        } catch (NotFoundExceptionInterface $e) {
+            throw new \LogicException(sprintf('Transport "%s" is not supported.', $transport), null, $e);
+        }
+
+        $output->writeln('Broker set up');
+
+        return null;
+    }
+
+    private function getSetupBroker(string $transport): SetupBrokerInterface
+    {
+        return $this->container->get(sprintf($this->setupBrokerIdPattern, $transport));
+    }
+}
