[Messenger] Allow to limit consumer to specific queues

Author: dbu

Transport/AmqpReceiver.php

@@ -16,7 +16,7 @@
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
 use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
-use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
+use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 
@@ -25,7 +25,7 @@
  *
  * @author Samuel Roze <[email protected]>
  */
-class AmqpReceiver implements ReceiverInterface, MessageCountAwareInterface
+class AmqpReceiver implements QueueReceiverInterface, MessageCountAwareInterface
 {
     private $serializer;
     private $connection;
@@ -41,7 +41,15 @@ public function __construct(Connection $connection, SerializerInterface $seriali
      */
     public function get(): iterable
     {
-        foreach ($this->connection->getQueueNames() as $queueName) {
+        yield from $this->getFromQueues($this->connection->getQueueNames());
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getFromQueues(array $queueNames): iterable
+    {
+        foreach ($queueNames as $queueName) {
             yield from $this->getEnvelope($queueName);
         }
     }

composer.json

@@ -18,7 +18,7 @@
     "require": {
         "php": ">=7.2.5",
         "symfony/deprecation-contracts": "^2.1",
-        "symfony/messenger": "^5.1"
+        "symfony/messenger": "^5.3"
     },
     "require-dev": {
         "symfony/event-dispatcher": "^4.4|^5.0",