[consumption] add support of QueueSubscriberInterface to consume command.

Author: makasim

docs/bundle/cli_commands.md

@@ -138,18 +138,18 @@ Help:
 ## enqueue:transport:consume
 
 ```
-./bin/console enqueue:transport:consume   --help
+./bin/console enqueue:transport:consume --help
 Usage:
-  enqueue:transport:consume [options] [--] <queue> <processor-service>
+  enqueue:transport:consume [options] [--] <processor-service>
 
 Arguments:
-  queue                              Queues to consume from
   processor-service                  A message processor service
 
 Options:
       --message-limit=MESSAGE-LIMIT  Consume n messages and exit
       --time-limit=TIME-LIMIT        Consume messages during this time
       --memory-limit=MEMORY-LIMIT    Consume messages until process reaches this memory limit in MB
+      --queue[=QUEUE]                Queues to consume from (multiple values allowed)
   -h, --help                         Display this help message
   -q, --quiet                        Do not output any message
   -V, --version                      Display this application version

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

@@ -148,7 +148,7 @@ public function testTransportConsumeMessagesCommandShouldConsumeMessage(array $e
         $tester->execute([
             '--message-limit' => 1,
             '--time-limit' => '+10sec',
-            'queue' => 'enqueue.test',
+            '--queue' => ['enqueue.test'],
             'processor-service' => 'test.message.processor',
         ]);
 

pkg/enqueue/Symfony/Consumption/ContainerAwareConsumeMessagesCommand.php

@@ -5,10 +5,12 @@
 use Enqueue\Consumption\ChainExtension;
 use Enqueue\Consumption\Extension\LoggerExtension;
 use Enqueue\Consumption\QueueConsumer;
+use Enqueue\Consumption\QueueSubscriberInterface;
 use Enqueue\Psr\PsrProcessor;
 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;
 use Symfony\Component\DependencyInjection\ContainerAwareInterface;
@@ -48,8 +50,8 @@ protected function configure()
             ->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('queue', InputArgument::REQUIRED, 'Queues to consume from')
             ->addArgument('processor-service', InputArgument::REQUIRED, 'A message processor service')
+            ->addOption('queue', null, InputOption::VALUE_IS_ARRAY | InputOption::VALUE_OPTIONAL, 'Queues to consume from', [])
         ;
     }
 
@@ -58,28 +60,38 @@ protected function configure()
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        $queueName = $input->getArgument('queue');
-
         /** @var PsrProcessor $processor */
         $processor = $this->container->get($input->getArgument('processor-service'));
-        if (!$processor instanceof  PsrProcessor) {
+        if (false == $processor instanceof  PsrProcessor) {
             throw new \LogicException(sprintf(
                 'Invalid message processor service given. It must be an instance of %s but %s',
                 PsrProcessor::class,
                 get_class($processor)
             ));
         }
 
+        $queues = $input->getOption('queue');
+        if (empty($queues) && $processor instanceof QueueSubscriberInterface) {
+            $queues = $processor::getSubscribedQueues();
+        }
+
+        if (empty($queues)) {
+            throw new \LogicException(sprintf(
+                'The queues are not provided. The processor must implement "%s" interface and it must return not empty array of queues or queues set using --queue option.',
+                QueueSubscriberInterface::class
+            ));
+        }
+
         $extensions = $this->getLimitsExtensions($input, $output);
         array_unshift($extensions, new LoggerExtension(new ConsoleLogger($output)));
 
         $runtimeExtensions = new ChainExtension($extensions);
 
         try {
-            $queue = $this->consumer->getPsrContext()->createQueue($queueName);
-            // @todo set additional queue options
+            foreach($queues as $queue) {
+                $this->consumer->bind($queue, $processor);
+            }
 
-            $this->consumer->bind($queue, $processor);
             $this->consumer->consume($runtimeExtensions);
         } finally {
             $this->consumer->getPsrContext()->close();

pkg/enqueue/Tests/Symfony/Consumption/ContainerAwareConsumeMessagesCommandTest.php

@@ -8,6 +8,7 @@
 use Enqueue\Psr\PsrProcessor;
 use Enqueue\Psr\PsrQueue;
 use Enqueue\Symfony\Consumption\ContainerAwareConsumeMessagesCommand;
+use Enqueue\Tests\Symfony\Consumption\Mock\QueueSubscriberProcessor;
 use Symfony\Component\Console\Tester\CommandTester;
 use Symfony\Component\DependencyInjection\Container;
 use PHPUnit\Framework\TestCase;
@@ -32,10 +33,11 @@ public function testShouldHaveExpectedOptions()
 
         $options = $command->getDefinition()->getOptions();
 
-        $this->assertCount(3, $options);
+        $this->assertCount(4, $options);
         $this->assertArrayHasKey('memory-limit', $options);
         $this->assertArrayHasKey('message-limit', $options);
         $this->assertArrayHasKey('time-limit', $options);
+        $this->assertArrayHasKey('queue', $options);
     }
 
     public function testShouldHaveExpectedAttributes()
@@ -44,41 +46,63 @@ public function testShouldHaveExpectedAttributes()
 
         $arguments = $command->getDefinition()->getArguments();
 
-        $this->assertCount(2, $arguments);
+        $this->assertCount(1, $arguments);
         $this->assertArrayHasKey('processor-service', $arguments);
-        $this->assertArrayHasKey('queue', $arguments);
     }
 
     public function testShouldThrowExceptionIfProcessorInstanceHasWrongClass()
     {
-        $this->setExpectedException(\LogicException::class, 'Invalid message processor service given.'.
-            ' It must be an instance of Enqueue\Psr\PsrProcessor but stdClass');
-
         $container = new Container();
         $container->set('processor-service', new \stdClass());
 
         $command = new ContainerAwareConsumeMessagesCommand($this->createQueueConsumerMock());
         $command->setContainer($container);
 
         $tester = new CommandTester($command);
+
+        $this->expectException(\LogicException::class);
+        $this->expectExceptionMessage('Invalid message processor service given. It must be an instance of Enqueue\Psr\PsrProcessor but stdClass');
         $tester->execute([
-            'queue' => 'queue-name',
             'processor-service' => 'processor-service',
+            '--queue' => ['queue-name'],
         ]);
     }
 
-    public function testShouldExecuteConsumption()
+    public function testThrowIfNeitherQueueOptionNorProcessorImplementsQueueSubscriberInterface()
     {
         $processor = $this->createProcessor();
 
-        $queue = $this->createQueueMock();
+        $consumer = $this->createQueueConsumerMock();
+        $consumer
+            ->expects($this->never())
+            ->method('bind')
+        ;
+        $consumer
+            ->expects($this->never())
+            ->method('consume')
+        ;
+
+        $container = new Container();
+        $container->set('processor-service', $processor);
+
+        $command = new ContainerAwareConsumeMessagesCommand($consumer);
+        $command->setContainer($container);
+
+        $tester = new CommandTester($command);
+
+
+        $this->expectException(\LogicException::class);
+        $this->expectExceptionMessage('The queues are not provided. The processor must implement "Enqueue\Consumption\QueueSubscriberInterface" interface and it must return not empty array of queues or queues set using --queue option.');
+        $tester->execute([
+            'processor-service' => 'processor-service',
+        ]);
+    }
+
+    public function testShouldExecuteConsumptionWithExplisitlySetQueueViaQueueOption()
+    {
+        $processor = $this->createProcessor();
 
         $context = $this->createContextMock();
-        $context
-            ->expects($this->once())
-            ->method('createQueue')
-            ->willReturn($queue)
-        ;
         $context
             ->expects($this->once())
             ->method('close')
@@ -88,15 +112,15 @@ public function testShouldExecuteConsumption()
         $consumer
             ->expects($this->once())
             ->method('bind')
-            ->with($this->identicalTo($queue), $this->identicalTo($processor))
+            ->with('queue-name', $this->identicalTo($processor))
         ;
         $consumer
             ->expects($this->once())
             ->method('consume')
             ->with($this->isInstanceOf(ChainExtension::class))
         ;
         $consumer
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(1))
             ->method('getPsrContext')
             ->will($this->returnValue($context))
         ;
@@ -109,8 +133,52 @@ public function testShouldExecuteConsumption()
 
         $tester = new CommandTester($command);
         $tester->execute([
-            'queue' => 'queue-name',
             'processor-service' => 'processor-service',
+            '--queue' => ['queue-name'],
+        ]);
+    }
+
+    public function testShouldExecuteConsumptionWhenProcessorImplementsQueueSubscriberInterface()
+    {
+        $processor = new QueueSubscriberProcessor();
+
+        $context = $this->createContextMock();
+        $context
+            ->expects($this->once())
+            ->method('close')
+        ;
+
+        $consumer = $this->createQueueConsumerMock();
+        $consumer
+            ->expects($this->at(0))
+            ->method('bind')
+            ->with('fooSubscribedQueues', $this->identicalTo($processor))
+        ;
+        $consumer
+            ->expects($this->at(1))
+            ->method('bind')
+            ->with('barSubscribedQueues', $this->identicalTo($processor))
+        ;
+        $consumer
+            ->expects($this->at(2))
+            ->method('consume')
+            ->with($this->isInstanceOf(ChainExtension::class))
+        ;
+        $consumer
+            ->expects($this->at(3))
+            ->method('getPsrContext')
+            ->will($this->returnValue($context))
+        ;
+
+        $container = new Container();
+        $container->set('processor-service', $processor);
+
+        $command = new ContainerAwareConsumeMessagesCommand($consumer);
+        $command->setContainer($container);
+
+        $tester = new CommandTester($command);
+        $tester->execute([
+            'processor-service' => 'processor-service'
         ]);
     }
 

pkg/enqueue/Tests/Symfony/Consumption/Mock/QueueSubscriberProcessor.php

@@ -0,0 +1,19 @@
+<?php
+namespace Enqueue\Tests\Symfony\Consumption\Mock;
+
+use Enqueue\Consumption\QueueSubscriberInterface;
+use Enqueue\Psr\PsrContext;
+use Enqueue\Psr\PsrMessage;
+use Enqueue\Psr\PsrProcessor;
+
+class QueueSubscriberProcessor implements PsrProcessor, QueueSubscriberInterface
+{
+    public function process(PsrMessage $message, PsrContext $context)
+    {
+    }
+
+    public static function getSubscribedQueues()
+    {
+        return ['fooSubscribedQueues', 'barSubscribedQueues'];
+    }
+}