eMAGTechLabs / RabbitMqBundle

DependencyInjection/Compiler/ConsumersListCompilerPass.php

Spacing +1 added lines, -1 removed lines

@@ -13,7 +13,7 @@
     public function process(ContainerBuilder $container)
     {
         $taggedConsumers = $container->findTaggedServiceIds('old_sound_rabbit_mq.consumer');
-        $consumerNames = array_map(fn ($tags) => $tags[0]['consumer'], $taggedConsumers);
+        $consumerNames = array_map(fn($tags) => $tags[0]['consumer'], $taggedConsumers);
         $container->setParameter('old_sound_rabbit_mq.allowed_consumer_names', $consumerNames);
     }
 }
\ No newline at end of file

DependencyInjection/Compiler/ExtraContextLoggerCompilerPass.php

Spacing +2 added lines, -2 removed lines

@@ -22,7 +22,7 @@ 
 
         foreach ($taggedProducers as $id => $tags) {
             $producerName = $tags[0]['producer'];
-            $originLogger = $id . '.logger';
+            $originLogger = $id.'.logger';
 
             if ($container->hasDefinition($originLogger)) {
                 $consumerDef = $container->getDefinition($id);
@@ -34,7 +34,7 @@ 
 
         foreach ($taggedConsumers as $id => $tags) {
             $consumerName = $tags[0]['consumer'];
-            $originLogger = $id . '.logger';
+            $originLogger = $id.'.logger';
 
             if ($container->hasDefinition($originLogger)) {
                 $consumerDef = $container->getDefinition($id);

DependencyInjection/OldSoundRabbitMqExtension.php

Spacing +7 added lines, -7 removed lines

@@ -51,7 +51,7 @@ 
     {
         $this->container = $container;
 
-        $loader = new XmlFileLoader($this->container, new FileLocator(array(__DIR__ . '/../Resources/config')));
+        $loader = new XmlFileLoader($this->container, new FileLocator(array(__DIR__.'/../Resources/config')));
         $loader->load('rabbitmq.xml');
 
         $configuration = $this->getConfiguration($configs, $container);
@@ -103,11 +103,11 @@ 
      */
     protected function loadExchanges($exchanges): array
     {
-        return array_map(function ($exchange) {
+        return array_map(function($exchange) {
             $exchangeDeclaration = new Definition(ExchangeDeclaration::class);
             $exchangeDeclaration->setProperties($exchange);
 
-            foreach($this->loadBindings($exchange['bindings'], $exchange['name'], null) as $binding) {
+            foreach ($this->loadBindings($exchange['bindings'], $exchange['name'], null) as $binding) {
                 $this->container->getDefinition('old_sound_rabbit_mq.declaration_registry')->addMethodCall('addBinding', [$binding]);
             }
 
@@ -121,7 +121,7 @@ 
      */
     protected function loadQueues($queues): array
     {
-        return array_map(function ($queue, $key) use ($queues) {
+        return array_map(function($queue, $key) use ($queues) {
             $queue['name'] = $queue['name'] ?? $key;
             $queueDeclaration = new Definition(QueueDeclaration::class);
             $queueDeclaration->setProperties($queue);
@@ -255,7 +255,7 @@ 
     {
         foreach ($this->config['consumers'] as $consumerName => $consumer) {
             $alias = sprintf('old_sound_rabbit_mq.consumer.%s', $consumerName);
-            $serializerAlias = sprintf('old_sound_rabbit_mq.consumer.%s.serializer', $consumerName);// TODO
+            $serializerAlias = sprintf('old_sound_rabbit_mq.consumer.%s.serializer', $consumerName); // TODO
 
             $connectionName = $consumer['connection'] ?? 'default';
 
@@ -266,7 +266,7 @@ 
             $definition->addTag('old_sound_rabbit_mq.consumer', ['consumer' => $consumerName]);
             // TODO $this->container->setAlias($serializerAlias, SerializerInterface::class);
             // $definition->addMethodCall('setSerializer', [new Reference($serializerAlias)]);}
-            foreach($consumer['consumeQueues'] as $index => $consumeQueue) {
+            foreach ($consumer['consumeQueues'] as $index => $consumeQueue) {
                 $queueConsumingDef = new Definition(QueueConsuming::class);
                 $queueConsumingDef->setProperties([
                     'queueName' => $consumeQueue['queue'],
@@ -431,7 +431,7 @@ 
         $definition->addTag('monolog.logger', [
             'channel' => 'phpamqplib'
         ]);
-        $loggerAlias = $definitionAlias . '.loggeer';
+        $loggerAlias = $definitionAlias.'.loggeer';
         $this->container->setAlias($loggerAlias, 'logger');
         $definition->addMethodCall('setLogger', [new Reference($loggerAlias, ContainerInterface::IGNORE_ON_INVALID_REFERENCE)]);
     }

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]