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]['name'], $taggedConsumers);
+        $consumerNames = array_map(fn($tags) => $tags[0]['name'], $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]['name'];
-            $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]['name'];
-            $originLogger = $id . '.logger';
+            $originLogger = $id.'.logger';
 
             if ($container->hasDefinition($originLogger)) {
                 $consumerDef = $container->getDefinition($id);

DependencyInjection/OldSoundRabbitMqExtension.php

Spacing +8 added lines, -9 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);
@@ -260,7 +260,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 = isset($consumer['connection']) ? $consumer['connection'] : 'default';
 
@@ -271,7 +271,7 @@ 
             $definition->addTag('old_sound_rabbit_mq.consumer', ['name' => $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'],
@@ -283,8 +283,7 @@ 
                 ]);
 
                 $executeCallbackStrategyRef = isset($consumeQueue['batch_count']) ?
-                    new Definition(BatchExecuteCallbackStrategy::class, [$consumeQueue['batch_count']]) :
-                    new Reference($simpleExecuteCallbackStrategyAlias);
+                    new Definition(BatchExecuteCallbackStrategy::class, [$consumeQueue['batch_count']]) : new Reference($simpleExecuteCallbackStrategyAlias);
 
                 $definition->addMethodCall('consumeQueue', [
                     $queueConsumingDef,
@@ -445,7 +444,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)]);
     }

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);
         }
     }