[consumption] Add ability to consume from multiple transports.

Author: makasim

Client/ArrayProcessorRegistry.php renamed to ArrayProcessorRegistry.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Enqueue\Client;
+namespace Enqueue;
 
 use Interop\Queue\Processor;
 
@@ -16,7 +16,10 @@ class ArrayProcessorRegistry implements ProcessorRegistryInterface
      */
     public function __construct(array $processors = [])
     {
-        $this->processors = $processors;
+        $this->processors = [];
+        array_walk($processors, function (Processor $processor, string $key) {
+            $this->processors[$key] = $processor;
+        });
     }
 
     public function add(string $name, Processor $processor): void

Client/DelegateProcessor.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Client;
 
+use Enqueue\ProcessorRegistryInterface;
 use Interop\Queue\Context;
 use Interop\Queue\Message as InteropMessage;
 use Interop\Queue\Processor;

Consumption/ArrayQueueConsumerRegistry.php

@@ -0,0 +1,36 @@
+<?php
+
+namespace Enqueue\Consumption;
+
+class ArrayQueueConsumerRegistry implements QueueConsumerRegistryInterface
+{
+    /**
+     * @var QueueConsumerInterface[]
+     */
+    private $consumers;
+
+    /**
+     * @param QueueConsumerInterface[] $queueConsumers
+     */
+    public function __construct(array $queueConsumers = [])
+    {
+        $this->consumers = [];
+        array_walk($queueConsumers, function (QueueConsumerInterface $consumer, string $key) {
+            $this->consumers[$key] = $consumer;
+        });
+    }
+
+    public function add(string $name, QueueConsumerInterface $consumer): void
+    {
+        $this->consumers[$name] = $consumer;
+    }
+
+    public function get(string $name): QueueConsumerInterface
+    {
+        if (false == isset($this->consumers[$name])) {
+            throw new \LogicException(sprintf('QueueConsumer was not found, name: "%s".', $name));
+        }
+
+        return $this->consumers[$name];
+    }
+}

Consumption/QueueConsumerRegistryInterface.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Consumption;
+
+interface QueueConsumerRegistryInterface
+{
+    public function get(string $name): QueueConsumerInterface;
+}

Client/ProcessorRegistryInterface.php renamed to ProcessorRegistryInterface.php

@@ -1,6 +1,8 @@
 <?php
 
-namespace Enqueue\Client;
+declare(strict_types=1);
+
+namespace Enqueue;
 
 use Interop\Queue\Processor;
 

Symfony/Consumption/ConfigurableConsumeCommand.php

@@ -0,0 +1,90 @@
+<?php
+
+namespace Enqueue\Symfony\Consumption;
+
+use Enqueue\Consumption\ChainExtension;
+use Enqueue\Consumption\Extension\LoggerExtension;
+use Enqueue\Consumption\QueueConsumerRegistryInterface;
+use Enqueue\Consumption\QueueSubscriberInterface;
+use Enqueue\ProcessorRegistryInterface;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+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 ConfigurableConsumeCommand extends Command
+{
+    use LimitsExtensionsCommandTrait;
+    use QueueConsumerOptionsCommandTrait;
+
+    protected static $defaultName = 'enqueue:transport:consume';
+
+    /**
+     * @var QueueConsumerRegistryInterface
+     */
+    protected $consumerRegistry;
+
+    /**
+     * @var ProcessorRegistryInterface
+     */
+    private $processorRegistry;
+
+    public function __construct(QueueConsumerRegistryInterface $consumerRegistry, ProcessorRegistryInterface $processorRegistry)
+    {
+        parent::__construct(static::$defaultName);
+
+        $this->consumerRegistry = $consumerRegistry;
+        $this->processorRegistry = $processorRegistry;
+    }
+
+    protected function configure(): void
+    {
+        $this->configureLimitsExtensions();
+        $this->configureQueueConsumerOptions();
+
+        $this
+            ->setDescription('A worker that consumes message from a broker. '.
+                'To use this broker you have to explicitly set a queue to consume from '.
+                'and a message processor service')
+            ->addArgument('processor', InputArgument::REQUIRED, 'A message processor.')
+            ->addArgument('queues', InputArgument::OPTIONAL | InputArgument::IS_ARRAY, 'A queue to consume from', [])
+            ->addOption('transport', 't', InputOption::VALUE_OPTIONAL, 'The transport to consume messages from.', 'default')
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): ?int
+    {
+        $consumer = $this->consumerRegistry->get($input->getOption('transport'));
+
+        $this->setQueueConsumerOptions($consumer, $input);
+
+        $processor = $this->processorRegistry->get($input->getArgument('processor'));
+
+        $queues = $input->getArgument('queues');
+        if (empty($queues) && $processor instanceof QueueSubscriberInterface) {
+            $queues = $processor::getSubscribedQueues();
+        }
+
+        if (empty($queues)) {
+            throw new \LogicException(sprintf(
+                'The queue is not provided. The processor must implement "%s" interface and it must return not empty array of queues or a queue set using as a second argument.',
+                QueueSubscriberInterface::class
+            ));
+        }
+
+        $extensions = $this->getLimitsExtensions($input, $output);
+        array_unshift($extensions, new LoggerExtension(new ConsoleLogger($output)));
+
+        $runtimeExtensions = new ChainExtension($extensions);
+
+        foreach ($queues as $queue) {
+            $consumer->bind($queue, $processor);
+        }
+
+        $consumer->consume($runtimeExtensions);
+
+        return null;
+    }
+}

Symfony/Consumption/ConsumeMessagesCommand.php renamed to Symfony/Consumption/ConsumeCommand.php

@@ -5,14 +5,16 @@
 use Enqueue\Consumption\ChainExtension;
 use Enqueue\Consumption\Extension\LoggerExtension;
 use Enqueue\Consumption\QueueConsumerInterface;
+use Enqueue\Consumption\QueueConsumerRegistryInterface;
 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;
 use Symfony\Component\DependencyInjection\ContainerAwareInterface;
 use Symfony\Component\DependencyInjection\ContainerAwareTrait;
 
-class ConsumeMessagesCommand extends Command implements ContainerAwareInterface
+class ConsumeCommand extends Command implements ContainerAwareInterface
 {
     use ContainerAwareTrait;
     use LimitsExtensionsCommandTrait;
@@ -21,46 +23,48 @@ class ConsumeMessagesCommand extends Command implements ContainerAwareInterface
     protected static $defaultName = 'enqueue:transport:consume';
 
     /**
-     * @var QueueConsumerInterface
+     * @var QueueConsumerRegistryInterface
      */
-    protected $consumer;
+    protected $consumerRegistry;
 
     /**
-     * @param QueueConsumerInterface $consumer
+     * [name => QueueConsumerInterface].
+     *
+     * @param QueueConsumerInterface[]
      */
-    public function __construct(QueueConsumerInterface $consumer)
+    public function __construct(QueueConsumerRegistryInterface $consumerRegistry)
     {
         parent::__construct(static::$defaultName);
 
-        $this->consumer = $consumer;
+        $this->consumerRegistry = $consumerRegistry;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
     {
         $this->configureLimitsExtensions();
         $this->configureQueueConsumerOptions();
 
         $this
+            ->addOption('transport', 't', InputOption::VALUE_OPTIONAL, 'The transport to consume messages from.', 'default')
             ->setDescription('A worker that consumes message from a broker. '.
                 'To use this broker you have to configure queue consumer before adding to the command')
         ;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): ?int
     {
-        $this->setQueueConsumerOptions($this->consumer, $input);
+        // QueueConsumer must be pre configured outside of the command!
+        $consumer = $this->consumerRegistry->get($input->getOption('transport'));
+
+        $this->setQueueConsumerOptions($consumer, $input);
 
         $extensions = $this->getLimitsExtensions($input, $output);
         array_unshift($extensions, new LoggerExtension(new ConsoleLogger($output)));
 
         $runtimeExtensions = new ChainExtension($extensions);
 
-        $this->consumer->consume($runtimeExtensions);
+        $consumer->consume($runtimeExtensions);
+
+        return null;
     }
 }

Symfony/Consumption/ContainerAwareConsumeMessagesCommand.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Symfony\Consumption;
+
+use Enqueue\Consumption\QueueConsumerInterface;
+use Enqueue\Consumption\QueueConsumerRegistryInterface;
+use Psr\Container\ContainerInterface;
+
+final class ContainerQueueConsumerRegistry implements QueueConsumerRegistryInterface
+{
+    /**
+     * @var ContainerInterface
+     */
+    private $locator;
+
+    public function __construct(ContainerInterface $locator)
+    {
+        $this->locator = $locator;
+    }
+
+    public function get(string $name): QueueConsumerInterface
+    {
+        if (false == $this->locator->has($name)) {
+            throw new \LogicException(sprintf('Service locator does not have a queue consumer with name "%s".', $name));
+        }
+
+        return $this->locator->get($name);
+    }
+}