blocking messenger consumer

symfony · melihovv · Aug 30, 2022 · Aug 10, 2022 · Aug 11, 2022 · Aug 30, 2022
commit f871f19b9897718085d3f9845ce48f9719e18bac
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php
@@ -63,7 +63,7 @@ public function testItReturnsTheDecodedMessageToTheHandlerInBlockingMode()
 
         $connection->method('getQueueNames')->willReturn(['queueName']);
         $connection->method('pull')->willReturnCallback(function (string $queueName, callable $callback) use ($amqpQueue, $amqpEnvelope) {
-            call_user_func($callback, $amqpEnvelope, $amqpQueue);
+            \call_user_func($callback, $amqpEnvelope, $amqpQueue);
         });
 
         $receiver = new AmqpReceiver($connection, $serializer);

diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpTransportTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpTransportTest.php
@@ -74,7 +74,7 @@ public function testReceivesMessagesInBlockingMode()
         $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);
+            \call_user_func($callback, $amqpEnvelope, $amqpQueue);
         });
 
         $transport->pull(function (Envelope $envelope) use ($decodedMessage) {

diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php
@@ -55,7 +55,7 @@ public function pull(callable $callback): void
      */
     public function pullFromQueues(array $queueNames, callable $callback): void
     {
-        if (0 === count($queueNames)) {
+        if (0 === \count($queueNames)) {
             return;
         }
 
@@ -71,7 +71,7 @@ public function pullFromQueues(array $queueNames, callable $callback): void
 
     private function pullEnvelope(string $queueName, ?callable $callback): void
     {
-        if ($callback !== null) {
+        if (null !== $callback) {
             $callback = function (\AMQPEnvelope $amqpEnvelope, \AMQPQueue $queue) use ($callback) {
                 $queueName = $queue->getName();
                 $body = $amqpEnvelope->getBody();

diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
@@ -437,9 +437,9 @@ public function get(string $queueName): ?\AMQPEnvelope
      * 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.
+     *                                                             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
      */

diff --git a/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php b/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php
@@ -33,7 +33,6 @@
 use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnTimeLimitListener;
 use Symfony\Component\Messenger\RoutableMessageBus;
-use Symfony\Component\Messenger\Transport\Receiver\BlockingReceiverInterface;
 use Symfony\Component\Messenger\Worker;
 
 /**

diff --git a/src/Symfony/Component/Messenger/Tests/Command/ConsumeMessagesCommandTest.php b/src/Symfony/Component/Messenger/Tests/Command/ConsumeMessagesCommandTest.php
@@ -79,7 +79,7 @@ public function testRunWithBlockingModeOption()
 
         $receiver = $this->createMock(QueueBlockingReceiverInterface::class);
         $receiver->expects($this->once())->method('pullFromQueues')->willReturnCallback(function (array $queueNames, callable $callback) use ($envelope) {
-            call_user_func($callback, $envelope);
+            \call_user_func($callback, $envelope);
         });
 
         $receiverLocator = $this->createMock(ContainerInterface::class);

diff --git a/src/Symfony/Component/Messenger/Tests/WorkerTest.php b/src/Symfony/Component/Messenger/Tests/WorkerTest.php
@@ -28,7 +28,6 @@
 use Symfony\Component\Messenger\EventListener\ResetServicesListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
 use Symfony\Component\Messenger\Exception\RuntimeException;
-use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Handler\Acknowledger;
 use Symfony\Component\Messenger\Handler\BatchHandlerInterface;
 use Symfony\Component\Messenger\Handler\BatchHandlerTrait;
@@ -694,7 +693,7 @@ public function dispatch(object $event): object
         $worker = new Worker(['transport' => $receiver], $bus, $dispatcher);
         $worker->run([
             'blocking-mode' => true,
-            'queues' => ['foo']
+            'queues' => ['foo'],
         ]);
 
         $this->assertSame($apiMessage, $envelopes[0]->getMessage());
@@ -782,7 +781,7 @@ public function pull(callable $callback): void
         foreach ($envelopes as $envelope) {
             $shouldContinue = $callback($envelope);
 
-    
23D3
        if ($shouldContinue === false) {
+            if (false === $shouldContinue) {
                 return;
             }
         }

diff --git a/src/Symfony/Component/Messenger/Transport/Receiver/BlockingReceiverInterface.php b/src/Symfony/Component/Messenger/Transport/Receiver/BlockingReceiverInterface.php
@@ -19,8 +19,8 @@
 interface BlockingReceiverInterface extends ReceiverInterface
 {
     /**
-     * @param callable(\AMQPEnvelope):?false $callback If callback return false, then processing thread will be
-     * returned to PHP script.
+     * @param callable(\AMQPEnvelope):?false $callback if callback return false, then processing thread will be
+     *                                                 returned to PHP script
      *
      * @throws TransportException If there is an issue communicating with the transport
      */

diff --git a/src/Symfony/Component/Messenger/Transport/Receiver/QueueBlockingReceiverInterface.php b/src/Symfony/Component/Messenger/Transport/Receiver/QueueBlockingReceiverInterface.php
@@ -21,9 +21,9 @@ interface QueueBlockingReceiverInterface extends BlockingReceiverInterface
     /**
      * Pull messages from the specified queue names instead of consuming from all queues.
      *
-     * @param string[] $queueNames
-     * @param callable(\AMQPEnvelope):?false $callback If callback return false, then processing thread will be
-     * returned to PHP script.
+     * @param string[]                       $queueNames
+     * @param callable(\AMQPEnvelope):?false $callback   if callback return false, then processing thread will be
+     *                                                   returned to PHP script
      */
     public function pullFromQueues(array $queueNames, callable $callback): void;
 }
diff --git a/src/Symfony/Component/Messenger/Worker.php b/src/Symfony/Component/Messenger/Worker.php
@@ -84,8 +84,8 @@ public function run(array $options = []): void
         $this->metadata->set(['queueNames' => $queueNames]);
 
         if ($blockingMode) {
-            if (count($this->receivers) > 1) {
-                throw new RuntimeException('In blocking mode only one receiver is supported');
+            if (\count($this->receivers) > 1) {
+                throw new RuntimeException('In blocking mode only one receiver is supported.');
             }
 
             foreach ($this->receivers as $transportName => $receiver) {
@@ -130,10 +130,10 @@ public function run(array $options = []): void
                     };
 
                     if ($queueNames) {
-                        /** @var QueueBlockingReceiverInterface $receiver */
+                        /* @var QueueBlockingReceiverInterface $receiver */
                         $receiver->pullFromQueues($queueNames, $callback);
                     } else {
-                        /** @var BlockingReceiverInterface $receiver */
+                        /* @var BlockingReceiverInterface $receiver */
                         $receiver->pull($callback);
                     }
                 } else {