eMAGTechLabs / RabbitMqBundle

RabbitMq/Consumer.php

Spacing +13 added lines, -15 removed lines

@@ -97,19 +97,18 @@ 
 
     protected function setup(): Consumer
     {
-        foreach($this->queueConsumings as $index => $queueConsuming) {
+        foreach ($this->queueConsumings as $index => $queueConsuming) {
             $this->channel->basic_qos($queueConsuming->qosPrefetchSize, $queueConsuming->qosPrefetchCount, false);
 
             $consumerTag = $this->channel->basic_consume(
                 $queueConsuming->queueName,
                 $queueConsuming->consumerTag ?
-                    $queueConsuming->consumerTag :
-                    sprintf("PHPPROCESS_%s_%s_%s", gethostname(), getmypid(), $index),
+                    $queueConsuming->consumerTag : sprintf("PHPPROCESS_%s_%s_%s", gethostname(), getmypid(), $index),
                 $queueConsuming->noLocal,
                 $queueConsuming->noAck,
                 $queueConsuming->exclusive,
                 $queueConsuming->nowait,
-                function (AMQPMessage $message) use ($queueConsuming) {
+                function(AMQPMessage $message) use ($queueConsuming) {
                     $this->getExecuteCallbackStrategy($queueConsuming)->consumeCallback($message);
                 });
 
@@ -135,12 +134,11 @@ 
         $this->queueConsumings[] = $queueConsuming;
         if (null === $executeCallbackStrategy) {
             $executeCallbackStrategy = null === $queueConsuming->batchCount ?
-                new SimpleExecuteCallbackStrategy() :
-                new BatchExecuteCallbackStrategy($queueConsuming->batchCount);
+                new SimpleExecuteCallbackStrategy() : new BatchExecuteCallbackStrategy($queueConsuming->batchCount);
         }
 
         $executeCallbackStrategy->setMessagesProccessor(new FnMessagesProcessor(
-            (function (array $messages) use ($queueConsuming) {
+            (function(array $messages) use ($queueConsuming) {
                 $logAmqpContext = ['queue' => $queueConsuming->queueName];
                 if ($this->getExecuteCallbackStrategy($queueConsuming)->canPrecessMultiMessages()) {
                     $logAmqpContext['messages'] = $messages;
@@ -183,11 +181,11 @@ 
         $canPrecessMultiMessages = $executeCallbackStrategy->canPrecessMultiMessages();
         if ($canPrecessMultiMessages) {
             if (!$queueConsuming->callback instanceof BatchConsumerInterface) {
-                throw new \InvalidArgumentException('TODO '. $queueConsuming->queueName);
+                throw new \InvalidArgumentException('TODO '.$queueConsuming->queueName);
             }
         } else {
             if (!$queueConsuming->callback instanceof ConsumerInterface) {
-                throw new \InvalidArgumentException('TODO '. $queueConsuming->queueName);
+                throw new \InvalidArgumentException('TODO '.$queueConsuming->queueName);
             }
         }
 
@@ -247,7 +245,7 @@ 
                     break;
                 }
             } catch (AMQPTimeoutException $e) {
-                foreach($this->executeCallbackStrategies as $executeCallbackStrategy) {
+                foreach ($this->executeCallbackStrategies as $executeCallbackStrategy) {
                     $executeCallbackStrategy->onCatchTimeout($e);
                 }
                 $now = new \DateTime();
@@ -313,7 +311,7 @@ 
 
         if (!$queueConsuming->noAck) {
             $messages = array_combine(
-                array_map(fn ($message) => $message->getDeliveryTag(), $messages),
+                array_map(fn($message) => $message->getDeliveryTag(), $messages),
                 $messages
             );
 
@@ -332,9 +330,9 @@ 
     {
         $executeCallbackStrategy = $this->getExecuteCallbackStrategy($queueConsuming);
 
-        $ack = !array_search(fn ($reply) => $reply !== null && $reply !== ConsumerInterface::MSG_ACK, $replies, true);
+        $ack = !array_search(fn($reply) => $reply !== null && $reply !== ConsumerInterface::MSG_ACK, $replies, true);
         if ($this->multiAck && count($messages) > 1 && $ack) {
-            $channels = array_map(fn ($message) => $message->getChannel(), $messages);
+            $channels = array_map(fn($message) => $message->getChannel(), $messages);
             if (count($channels) !== array_unique($channels)) { // all messages have same channel
                 throw new InvalidArgumentException('Messages can not be processed as multi ack with different channels');
             }
@@ -344,7 +342,7 @@ 
             $executeCallbackStrategy->onMessageProcessed($message);
 
             return array_combine(
-                array_map(fn ($message) => $message->getDeliveryTag(), $messages),
+                array_map(fn($message) => $message->getDeliveryTag(), $messages),
                 array_fill(0, count($messages), ConsumerInterface::MSG_ACK)
             );
         } else {
@@ -386,7 +384,7 @@ 
                 'content_type' => 'text/plain',
                 'correlation_id' => $message->get('correlation_id'),
             ]);
-            $message->getChannel()->basic_publish($replayMessage , '', $message->get('reply_to'));
+            $message->getChannel()->basic_publish($replayMessage, '', $message->get('reply_to'));
         } else {
             $this->logger->error('Rpc call send msg to queue which have not rpc reponse', [
                 'amqp' => ['message' => $message]