eMAGTechLabs / RabbitMqBundle

EventListener/PcntlSignalDispatchSubscriber.php

Spacing +1 added lines, -1 removed lines

@@ -21,7 +21,7 @@
 
     public function __construct(Consumer $consumer)
     {
-        $stopConsumer = function () use ($consumer) {
+        $stopConsumer = function() use ($consumer) {
             // Process current message, then halt consumer
             $consumer->forceStopConsumer();
             // Halt consumer if waiting for a new message from the queue

RabbitMq/AMQPConnectionFactory.php

Spacing +5 added lines, -5 removed lines

@@ -31,7 +31,7 @@ 
     {
         $parameters = $this->parseUrl(array_merge($this->defaultParameters, $parameters));
         if (is_array($parameters['ssl_context'])) {
-            $parameters['ssl_context'] = ! empty($parameters['ssl_context'])
+            $parameters['ssl_context'] = !empty($parameters['ssl_context'])
                 ? stream_context_create(array('ssl' => $parameters['ssl_context']))
                 : null;
         }
@@ -46,10 +46,10 @@ 
             $parameters['user'],
             $parameters['password'],
             $parameters['vhost'],
-            false,      // insist
+            false, // insist
             'AMQPLAIN', // login_method
-            null,       // login_response
-            'en_US',    // locale
+            null, // login_response
+            'en_US', // locale
         ];
 
         $isSocketConnection = $class == \PhpAmqpLib\Connection\AMQPSocketConnection::class || is_subclass_of($class, \PhpAmqpLib\Connection\AMQPSocketConnection::class);
@@ -93,7 +93,7 @@ 
             $parameters['host'] = urldecode($url['host']);
         }
         if (isset($url['port'])) {
-            $parameters['port'] = (int)$url['port'];
+            $parameters['port'] = (int) $url['port'];
         }
         if (isset($url['user'])) {
             $parameters['user'] = urldecode($url['user']);

RabbitMq/RpcClient.php

Spacing +1 added lines, -1 removed lines

@@ -124,7 +124,7 @@
         }
 
         $replices = [];
-        foreach($this->messages as $message) {
+        foreach ($this->messages as $message) {
             /** @var AMQPMessage $message */
             if (!$message->has('correlation_id')) {
                 $this->logger->error('unexpected message. rpc replies have no correlation_id ');

RabbitMq/Consumer.php

Spacing +12 added lines, -13 removed lines

@@ -95,19 +95,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);
                 });
 
@@ -122,7 +121,7 @@ 
     {
         $this->queueConsumings[] = $queueConsuming;
         $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;
@@ -165,11 +164,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);
             }
         }
 
@@ -229,7 +228,7 @@ 
                     break;
                 }
             } catch (AMQPTimeoutException $e) {
-                foreach($this->executeCallbackStrategies as $executeCallbackStrategy) {
+                foreach ($this->executeCallbackStrategies as $executeCallbackStrategy) {
                     $executeCallbackStrategy->onCatchTimeout($e);
                 }
                 $now = new \DateTime();
@@ -295,7 +294,7 @@ 
 
         if (!$queueConsuming->noAck) {
             $messages = array_combine(
-                array_map(fn ($message) => $message->getDeliveryTag(), $messages),
+                array_map(fn($message) => $message->getDeliveryTag(), $messages),
                 $messages
             );
 
@@ -314,9 +313,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');
             }
@@ -326,7 +325,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 {
@@ -368,7 +367,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]

Serializer/JsonMessageBodySerializer.php

Spacing +1 added lines, -1 removed lines

@@ -33,7 +33,7 @@
                 'message' => $body->getMessage(),
             ]);
         }
-        return json_encode($body);// $this->serializer->serialize($body, 'json');
+        return json_encode($body); // $this->serializer->serialize($body, 'json');
     }
 
     public function deserialize(string $body)

Command/ConsumerCommand.php

Spacing +1 added lines, -1 removed lines

@@ -113,7 +113,7 @@
             new ConsoleLogger($output)
         );
         $declarationRegistry = $this->container->get('old_sound_rabbit_mq.declaration_registry');
-        foreach($consumer->getQueueConsumings() as $queueConsuming) {
+        foreach ($consumer->getQueueConsumings() as $queueConsuming) {
             $declarator->declareForQueueDeclaration($queueConsuming->queueName, $declarationRegistry);
         }
     }

Command/DeclareCommand.php

Spacing +1 added lines, -1 removed lines

@@ -42,7 +42,7 @@
 
         $connection = $input->getArgument('connection');
         $channelAlias = sprintf('old_sound_rabbit_mq.channel.%s', $connection);
-        if(!$this->container->has($channelAlias)) {
+        if (!$this->container->has($channelAlias)) {
             throw new InvalidOptionException('Connection is not exist');
         };
 

Declarations/DeclarationsRegistry.php

Spacing +1 added lines, -1 removed lines

@@ -35,7 +35,7 @@
      */
     public function getBindingsByExchange(ExchangeDeclaration $exchange): array
     {
-        return array_filter($this->bindings, function ($binding) use ($exchange) {
+        return array_filter($this->bindings, function($binding) use ($exchange) {
             return $binding->exchange === $exchange->name || ($binding->destinationIsExchange && $binding->destination === $exchange->name);
         });
     }

Declarations/QueueDeclaration.php

Spacing +1 added lines, -1 removed lines

@@ -38,7 +38,7 @@
      */
     public function declure() {
         foreach ($this->queues as $name => $options) {
-            list($queueName, ,) = $this->getChannel()->queue_declare($name, $options['passive'],
+            list($queueName,,) = $this->getChannel()->queue_declare($name, $options['passive'],
                 $options['durable'], $options['exclusive'],
                 $options['auto_delete'], $options['nowait'],
                 $options['arguments'], $options['ticket']);