Port the solution in 5.4

Oleg Namaka · Oleg Namaka · commit e11e2d7be3af · 2024-10-25T19:37:00.000-05:00
diff --git a/composer.json b/composer.json
@@ -167,7 +167,8 @@
         "allow-plugins": {
             "php-http/discovery": false,
             "symfony/runtime": true
-        }
+        },
+        "use-parent-dir": true
     },
     "autoload": {
         "psr-4": {
@@ -204,10 +205,6 @@
                     "symfony/contracts": "2.5.x-dev"
                 }
             }
-        },
-        {
-            "type": "path",
-            "url": "src/Symfony/Component/Runtime"
         }
     ],
     "minimum-stability": "dev"
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php
@@ -46,6 +46,32 @@ public function testItReturnsTheDecodedMessageToTheHandler()
         $this->assertEquals(new DummyMessage('Hi'), $actualEnvelopes[0]->getMessage());
     }
 
+    public function testItReturnsTheDecodedMessageToTheHandlerInBlockingMode()
+    {
+        $connection = $this->getMockBuilder(Connection::class)
+            ->disableOriginalConstructor()
+            ->onlyMethods(['getQueueNames', 'pull'])
+            ->getMock();
+        $serializer = new Serializer(
+            new SerializerComponent\Serializer([new ObjectNormalizer()], ['json' => new JsonEncoder()])
+        );
+
+        $amqpEnvelope = $this->createAMQPEnvelope();
+
+        $amqpQueue = $this->createMock(\AMQPQueue::class);
+        $amqpQueue->method('getName')->willReturn('queueName');
+
+        $connection->method('getQueueNames')->willReturn(['queueName']);
+        $connection->method('pull')->willReturnCallback(function (string $queueName, callable $callback) use ($amqpQueue, $amqpEnvelope) {
+            \call_user_func($callback, $amqpEnvelope, $amqpQueue);
+        });
+
+        $receiver = new AmqpReceiver($connection, $serializer);
+        $receiver->pull(function (Envelope $envelope) {
+            $this->assertEquals(new DummyMessage('Hi'), $envelope->getMessage());
+        });
+    }
+
     public function testItThrowsATransportExceptionIfItCannotAcknowledgeMessage()
     {
         $this->expectException(TransportException::class);
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpTransportTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpTransportTest.php
@@ -52,6 +52,36 @@ public function testReceivesMessages()
         $this->assertSame($decodedMessage, $envelopes[0]->getMessage());
     }
 
+    public function testReceivesMessagesInBlockingMode()
+    {
+        $transport = $this->getTransport(
+            $serializer = $this->createMock(SerializerInterface::class),
+            $connection = $this->getMockBuilder(Connection::class)
+                ->disableOriginalConstructor()
+                ->onlyMethods(['getQueueNames', 'pull'])
+                ->getMock(),
+        );
+
+        $decodedMessage = new DummyMessage('Decoded.');
+
+        $amqpEnvelope = $this->createMock(\AMQPEnvelope::class);
+        $amqpEnvelope->method('getBody')->willReturn('body');
+        $amqpEnvelope->method('getHeaders')->willReturn(['my' => 'header']);
+
+        $amqpQueue = $this->createMock(\AMQPQueue::class);
+        $amqpQueue->method('getName')->willReturn('queueName');
+
+        $serializer->method('decode')->with(['body' => 'body', 'headers' => ['my' => 'header']])->willReturn(new Envelope($decodedMessage));
+        $connection->method('getQueueNames')->willReturn(['queueName']);
+        $connection->method('pull')->willReturnCallback(function (string $queueName, callable $callback) use ($amqpQueue, $amqpEnvelope) {
+            \call_user_func($callback, $amqpEnvelope, $amqpQueue);
+        });
+
+        $transport->pull(function (Envelope $envelope) use ($decodedMessage) {
+            $this->assertSame($decodedMessage, $envelope->getMessage());
+        });
+    }
+
     private function getTransport(?SerializerInterface $serializer = null, ?Connection $connection = null): AmqpTransport
     {
         $serializer = $serializer ?? $this->createMock(SerializerInterface::class);
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php
@@ -16,6 +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\QueueBlockingReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -25,7 +26,7 @@
  *
  * @author Samuel Roze <samuel.roze@gmail.com>
  */
-class AmqpReceiver implements QueueReceiverInterface, MessageCountAwareInterface
+class AmqpReceiver implements QueueReceiverInterface, QueueBlockingReceiverInterface, MessageCountAwareInterface
 {
     private $serializer;
     private $connection;
@@ -44,9 +45,46 @@ public function get(): iterable
         yield from $this->getFromQueues($this->connection->getQueueNames());
     }
 
-    /**
-     * {@inheritdoc}
-     */
+    public function pull(callable $callback): void
+    {
+        $this->pullFromQueues($this->connection->getQueueNames(), $callback);
+    }
+
+    public function pullFromQueues(array $queueNames, callable $callback): void
+    {
+        if (0 === \count($queueNames)) {
+            return;
+        }
+
+        // Pop last queue to send callback
+        $firstQueue = array_pop($queueNames);
+
+        foreach ($queueNames as $queueName) {
+            $this->pullEnvelope($queueName, null);
+        }
+
+        $this->pullEnvelope($firstQueue, $callback);
+    }
+
+    private function pullEnvelope(string $queueName, ?callable $callback): void
+    {
+        if (null !== $callback) {
+            $callback = function (\AMQPEnvelope $amqpEnvelope, \AMQPQueue $queue) use ($callback) {
+                $queueName = $queue->getName();
+                $body = $amqpEnvelope->getBody();
+                $envelope = $this->decodeAmqpEnvelope($amqpEnvelope, $body, $queueName);
+
+                return $callback($envelope->with(new AmqpReceivedStamp($amqpEnvelope, $queueName)));
+            };
+        }
+
+        try {
+            $this->connection->pull($queueName, $callback);
+        } catch (\AMQPException $exception) {
+            throw new TransportException($exception->getMessage(), 0, $exception);
+        }
+    }
+
     public function getFromQueues(array $queueNames): iterable
     {
         foreach ($queueNames as $queueName) {
@@ -67,9 +105,15 @@ private function getEnvelope(string $queueName): iterable
         }
 
         $body = $amqpEnvelope->getBody();
+        $envelope = $this->decodeAmqpEnvelope($amqpEnvelope, $body, $queueName);
+
+        yield $envelope->with(new AmqpReceivedStamp($amqpEnvelope, $queueName));
+    }
 
+    private function decodeAmqpEnvelope(\AMQPEnvelope $amqpEnvelope, $body, string $queueName): Envelope
+    {
         try {
-            $envelope = $this->serializer->decode([
+            return $this->serializer->decode([
                 'body' => false === $body ? '' : $body, // workaround https://github.com/pdezwart/php-amqp/issues/351
                 'headers' => $amqpEnvelope->getHeaders(),
             ]);
@@ -79,8 +123,6 @@ private function getEnvelope(string $queueName): iterable
 
             throw $exception;
         }
-
-        yield $envelope->with(new AmqpReceivedStamp($amqpEnvelope, $queueName));
     }
 
     /**
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpTransport.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpTransport.php
@@ -13,6 +13,7 @@
 
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
+use Symfony\Component\Messenger\Transport\Receiver\QueueBlockingReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -22,7 +23,7 @@
 /**
  * @author Nicolas Grekas <p@tchwork.com>
  */
-class AmqpTransport implements QueueReceiverInterface, TransportInterface, SetupableTransportInterface, MessageCountAwareInterface
+class AmqpTransport implements QueueReceiverInterface, QueueBlockingReceiverInterface, TransportInterface, SetupableTransportInterface, MessageCountAwareInterface
 {
     private $serializer;
     private $connection;
@@ -43,6 +44,22 @@ public function get(): iterable
         return ($this->receiver ?? $this->getReceiver())->get();
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function pull(callable $callback): void
+    {
+        $this->getReceiver()->pull($callback);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function pullFromQueues(array $queueNames, callable $callback): void
+    {
+        $this->getReceiver()->pullFromQueues($queueNames, $callback);
+    }
+
     /**
      * {@inheritdoc}
      */
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
@@ -281,7 +281,7 @@ private static function normalizeQueueArguments(array $arguments): array
             }
 
             if (!is_numeric($arguments[$key])) {
-                throw new InvalidArgumentException(sprintf('Integer expected for queue argument "%s", "%s" given.', $key, get_debug_type($arguments[$key])));
+                throw new InvalidArgumentException(sprintf('Integer expected for queue argument "%s", "%s" given.', $key, gettype($arguments[$key])));
             }
 
             $arguments[$key] = (int) $arguments[$key];
@@ -457,6 +457,27 @@ public function get(string $queueName): ?\AMQPEnvelope
         return null;
     }
 
+    /**
+     * Consume a message from the specified queue in blocking mode.
+     *
+     * @param ?callable(\AMQPEnvelope,\AMQPQueue):?false $callback If callback return false, then processing thread will be
+     *                                                             returned from AMQPQueue::consume() to PHP script. If null is passed, then the messages delivered to this client
+     *                                                             will be made available to the first real callback registered. That allows one to have a single callback
+     *                                                             consuming from multiple queues.
+     *
+     * @throws \AMQPException
+     */
+    public function pull(string $queueName, ?callable $callback): void
+    {
+        $this->clearWhenDisconnected();
+
+        if ($this->autoSetupExchange) {
+            $this->setupExchangeAndQueues();
+        }
+
+        $this->queue($queueName)->consume($callback);
+    }
+
     public function ack(\AMQPEnvelope $message, string $queueName): bool
     {
         return $this->queue($queueName)->ack($message->getDeliveryTag()) ?? true;
diff --git a/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php b/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php
@@ -81,6 +81,7 @@ protected function configure(): void
                 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 | InputOption::VALUE_IS_ARRAY, 'Limit receivers to only consume from the specified queues'),
                 new InputOption('no-reset', null, InputOption::VALUE_NONE, 'Do not reset container services after each message'),
+                new InputOption('blocking-mode', null, InputOption::VALUE_NONE, 'Consume messages in blocking mode. If option is specified only one receiver is supported'),
             ])
             ->setDescription(self::$defaultDescription)
             ->setHelp(<<<'EOF'
@@ -122,6 +123,12 @@ protected function configure(): void
 Use the --no-reset option to prevent services resetting after each message (may lead to leaking services' state between messages):
 
     <info>php %command.full_name% <receiver-name> --no-reset</info>
+
+Use the --blocking-mode option to force receiver to work in blocking mode
+("consume" method will be used instead of "get" in RabbitMQ for example).
+Only supported by some receivers, and you should pass only one receiver:
+
+    <info>php %command.full_name% <receiver-name> --blocking-mode</info>
 EOF
             )
         ;
@@ -227,6 +234,7 @@ protected function execute(InputInterface $input, OutputInterface $output)
         $worker = new Worker($receivers, $bus, $this->eventDispatcher, $this->logger);
         $options = [
             'sleep' => $input->getOption('sleep') * 1000000,
+            'blocking-mode' => (bool) $input->getOption('blocking-mode'),
         ];
         if ($queues = $input->getOption('queues')) {
             $options['queues'] = $queues;
diff --git a/src/Symfony/Component/Messenger/Tests/Command/ConsumeMessagesCommandTest.php b/src/Symfony/Component/Messenger/Tests/Command/ConsumeMessagesCommandTest.php
@@ -28,6 +28,7 @@
 use Symfony\Component\Messenger\RoutableMessageBus;
 use Symfony\Component\Messenger\Stamp\BusNameStamp;
 use Symfony\Component\Messenger\Tests\Fixtures\ResettableDummyReceiver;
+use Symfony\Component\Messenger\Transport\Receiver\QueueBlockingReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 
 class ConsumeMessagesCommandTest extends TestCase
@@ -72,6 +73,42 @@ public function testBasicRun()
         $this->assertStringContainsString('[OK] Consuming messages from transport "dummy-receiver"', $tester->getDisplay());
     }
 
+    public function testRunWithBlockingModeOption()
+    {
+        $envelope = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
+
+        $receiver = $this->createMock(QueueBlockingReceiverInterface::class);
+        $receiver->expects($this->once())->method('pullFromQueues')->willReturnCallback(function (array $queueNames, callable $callback) use ($envelope) {
+            \call_user_func($callback, $envelope);
+        });
+
+        $receiverLocator = $this->createMock(ContainerInterface::class);
+        $receiverLocator->expects($this->once())->method('has')->with('dummy-receiver')->willReturn(true);
+        $receiverLocator->expects($this->once())->method('get')->with('dummy-receiver')->willReturn($receiver);
+
+        $bus = $this->createMock(MessageBusInterface::class);
+        $bus->expects($this->once())->method('dispatch');
+
+        $busLocator = $this->createMock(ContainerInterface::class);
+        $busLocator->expects($this->once())->method('has')->with('dummy-bus')->willReturn(true);
+        $busLocator->expects($this->once())->method('get')->with('dummy-bus')->willReturn($bus);
+
+        $command = new ConsumeMessagesCommand(new RoutableMessageBus($busLocator), $receiverLocator, new EventDispatcher());
+
+        $application = new Application();
+        $application->add($command);
+        $tester = new CommandTester($application->get('messenger:consume'));
+        $tester->execute([
+            'receivers' => ['dummy-receiver'],
+            '--limit' => 1,
+            '--blocking-mode' => true,
+            '--queues' => ['foo'],
+        ]);
+
+        $tester->assertCommandIsSuccessful();
+        $this->assertStringContainsString('[OK] Consuming messages from transport "dummy-receiver"', $tester->getDisplay());
+    }
+
     public function testRunWithBusOption()
     {
         $envelope = new Envelope(new \stdClass());
diff --git a/src/Symfony/Component/Messenger/Tests/WorkerTest.php b/src/Symfony/Component/Messenger/Tests/WorkerTest.php
diff --git a/src/Symfony/Component/Messenger/Transport/Receiver/BlockingReceiverInterface.php b/src/Symfony/Component/Messenger/Transport/Receiver/BlockingReceiverInterface.php
diff --git a/src/Symfony/Component/Messenger/Transport/Receiver/QueueBlockingReceiverInterface.php b/src/Symfony/Component/Messenger/Transport/Receiver/QueueBlockingReceiverInterface.php
diff --git a/src/Symfony/Component/Messenger/Worker.php b/src/Symfony/Component/Messenger/Worker.php
