symfony · weaverryan · Mar 26, 2019 · Mar 26, 2019 · Guikingone · Mar 26, 2019
diff --git a/src/Symfony/Component/Messenger/CHANGELOG.md b/src/Symfony/Component/Messenger/CHANGELOG.md
@@ -4,6 +4,7 @@ CHANGELOG
 4.3.0
 -----
 
+ * [BC BREAK]: The `Connection::exchange()` method for Amqp was made private.
  * New classes: `RoutableMessageBus`, `AddBusNameStampMiddleware`
    and `BusNameStamp` were added, which allow you to add a bus identifier
    to the `Envelope` then find the correct bus when receiving from

diff --git a/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php b/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php
@@ -75,6 +75,7 @@ protected function configure(): void
                 new InputOption('memory-limit', 'm', InputOption::VALUE_REQUIRED, 'The memory limit the worker can consume'),
                 new InputOption('time-limit', 't', InputOption::VALUE_REQUIRED, 'The time limit in seconds the worker can run'),
                 new InputOption('bus', 'b', InputOption::VALUE_REQUIRED, 'Name of the bus to which received messages should be dispatched (if not passed, bus is determined automatically.'),
+                new InputOption('queues', null, InputOption::VALUE_REQUIRED, 'comma-separated list of queue names in order of priority'),
             ])
             ->setDescription('Consumes messages')
             ->setHelp(<<<'EOF'
@@ -183,7 +184,15 @@ protected function execute(InputInterface $input, OutputInterface $output): void
             $io->comment('Re-run the command with a -vv option to see logs about consumed messages.');
         }
 
-        $worker = new Worker($receiver, $bus, $receiverName, $retryStrategy, $this->eventDispatcher, $this->logger);
+        // TODO - make "default" equal to null?
+        $queues = [];
+        if (null !== $input->getOption('queues')) {
+            $queues = array_map(function ($queue) {
+                return trim($queue);
+            }, explode(',', $input->getOption('queues')));
+        }
+
+        $worker = new Worker($receiver, $bus, $queues, $receiverName, $retryStrategy, $this->eventDispatcher, $this->logger);
         $worker->run();
     }
 

diff --git a/src/Symfony/Component/Messenger/Stamp/QueueNameStamp.php b/src/Symfony/Component/Messenger/Stamp/QueueNameStamp.php
@@ -0,0 +1,30 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Stamp;
+
+/**
+ * @author Ryan Weaver <ryan@symfonycasts.com>
+ */
+class QueueNameStamp implements StampInterface
+{
+    private $queueName;
+
+    public function __construct(string $queueName)
+    {
+        $this->queueName = $queueName;
+    }
+
+    public function getQueueName(): string
+    {
+        return $this->queueName;
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpReceiver.php b/src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpReceiver.php
@@ -41,11 +41,22 @@ public function __construct(Connection $connection, SerializerInterface $seriali
     /**
      * {@inheritdoc}
      */
-    public function receive(callable $handler): void
+    public function receive(callable $handler, array $queues = []): void
     {
+        if (0 === \count($queues)) {
+            $queues[] = null;
+        }
+
         while (!$this->shouldStop) {
             try {
-                $amqpEnvelope = $this->connection->get();
+                foreach ($queues as $queue) {
+                    $amqpEnvelope = $this->connection->get($queue);
+
+                    // stop when you find one
+                    if (null !== $amqpEnvelope) {
+                        break;
+                    }
+                }
             } catch (\AMQPException $exception) {
                 throw new TransportException($exception->getMessage(), 0, $exception);
             }

diff --git a/src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpSender.php b/src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpSender.php
@@ -14,6 +14,7 @@
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Stamp\DelayStamp;
+use Symfony\Component\Messenger\Stamp\QueueNameStamp;
 use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -43,15 +44,16 @@ public function send(Envelope $envelope): Envelope
     {
         $encodedMessage = $this->serializer->encode($envelope);
 
+        /** @var QueueNameStamp|null $queueNameStamp */
+        $queueNameStamp = $envelope->last(QueueNameStamp::class);
+        $queueName = null !== $queueNameStamp ? $queueNameStamp->getQueueName() : null;
+
         /** @var DelayStamp|null $delayStamp */
         $delayStamp = $envelope->last(DelayStamp::class);
-        $delay = 0;
-        if (null !== $delayStamp) {
-            $delay = $delayStamp->getDelay();
-        }
+        $delay = null !== $delayStamp ? $delayStamp->getDelay() : 0;
 
         try {
-            $this->connection->publish($encodedMessage['body'], $encodedMessage['headers'] ?? [], $delay);
+            $this->connection->publish($encodedMessage['body'], $encodedMessage['headers'] ?? [], $queueName, $delay);
         } catch (\AMQPException $e) {
             throw new TransportException($e->getMessage(), 0, $e);
         }

diff --git a/src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpTransport.php b/src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpTransport.php
@@ -37,9 +37,9 @@ public function __construct(Connection $connection, SerializerInterface $seriali
     /**
      * {@inheritdoc}
      */
-    public function receive(callable $handler): void
+    public function receive(callable $handler, array $queues = null): void
     {
-        ($this->receiver ?? $this->getReceiver())->receive($handler);
+        ($this->receiver ?? $this->getReceiver())->receive($handler, $queues);
     }
 
     /**