CMProductions / queues

src/Cmp/Queue/Infrastructure/RabbitMQ/AMQPLazyConnectionSingleton.php

Doc Comments +6 added lines

@@ -9,6 +9,12 @@
 
     private static $amqpLazyConnection;
 
+    /**
+     * @param string $host
+     * @param string $port
+     * @param string $user
+     * @param string $password
+     */
     public static function getInstance($host, $port, $user, $password, $vhost = '/')
     {
         if(null === static::$amqpLazyConnection) {

Spacing +1 added lines, -1 removed lines

@@ -11,7 +11,7 @@
 
     public static function getInstance($host, $port, $user, $password, $vhost = '/')
     {
-        if(null === static::$amqpLazyConnection) {
+        if (null === static::$amqpLazyConnection) {
             static::$amqpLazyConnection = new AMQPLazyConnection($host, $port, $user, $password, $vhost);
         }
 

src/Cmp/Queue/Infrastructure/RabbitMQ/RabbitMQReader.php

Doc Comments +3 added lines

@@ -51,6 +51,9 @@
         $this->channel->wait();
     }
 
+    /**
+     * @param callable $callback
+     */
     private function initialize($callback)
     {
         $this->rabbitMQMessageHandler->setEventCallback($callback);

Unused Use Statements -1 removed lines

@@ -3,7 +3,6 @@
 namespace Cmp\Queue\Infrastructure\RabbitMQ;
 
 use Cmp\DomainEvent\Domain\Event\DomainEvent;
-use Cmp\DomainEvent\Domain\Subscriber\AbstractSubscriber;
 use Cmp\Queue\Domain\Reader\QueueReader;
 use Psr\Log\LoggerInterface;
 

src/Cmp/DomainEvent/Infrastructure/Subscriber/RabbitMQ/RabbitMQSubscriberInitializer.php

Spacing +2 added lines, -2 removed lines

@@ -54,9 +54,9 @@
 
             $channel->exchange_declare($this->exchange, 'topic', false, true, false);
 
-            list($queueName, ,) = $channel->queue_declare($this->queue, false, false, true, true);
+            list($queueName,,) = $channel->queue_declare($this->queue, false, false, true, true);
 
-            foreach($this->domainTopics as $domainTopic) {
+            foreach ($this->domainTopics as $domainTopic) {
                 $this->logger->info('Binding Topic:' . $domainTopic);
                 $channel->queue_bind($queueName, $this->exchange, $domainTopic);
             }

src/Cmp/DomainEvent/Domain/Subscriber/Subscriber.php

Spacing +2 added lines, -2 removed lines

@@ -45,7 +45,7 @@ 
     public function notify(DomainEvent $domainEvent)
     {
         $this->logger->debug('Domain Event received, notifying subscribers');
-        foreach($this->subscriptors as $subscriptor) {
+        foreach ($this->subscriptors as $subscriptor) {
             if ($subscriptor->isSubscribed($domainEvent)) {
                 $subscriptor->notify($domainEvent);
             }
@@ -54,7 +54,7 @@ 
 
     public function start()
     {
-        while(true) {
+        while (true) {
             $this->queueReader->process(array($this, 'notify'));
         }
     }

src/Cmp/Queue/Infrastructure/RabbitMQ/RabbitMQWriter.php

Spacing +1 added lines, -1 removed lines

@@ -56,7 +56,7 @@
             $this->channel = $this->rabbitMQInitializer->initialize();
         }
 
-        foreach($messages as $message) {
+        foreach ($messages as $message) {
             $encodedMessage = json_encode($message);
             $this->logger->debug('Writing:' . $encodedMessage);
             $msg = new AMQPMessage($encodedMessage, array('delivery_mode' => 2));

src/Cmp/Queue/Infrastructure/RabbitMQ/RabbitMQConfig.php

Spacing +1 added lines, -1 removed lines

@@ -48,7 +48,7 @@
      * @param string $queue
      * @param string $vhost
      */
-    public function __construct($host, $port, $user, $password, $exchange, $queue='', $vhost='/')
+    public function __construct($host, $port, $user, $password, $exchange, $queue = '', $vhost = '/')
     {
         $this->host = $host;
         $this->port = $port;

src/Cmp/Queue/Infrastructure/RabbitMQ/RabbitMQWriterInitializer.php

Spacing +2 added lines, -2 removed lines

@@ -72,8 +72,8 @@
     {
         $channel = $this->connection->channel();
 
-        $exchangeDelayed = self::DELAY_QUEUE_PREFIX.$delay.$this->exchangeName;
-        $queueDelayed = self::DELAY_QUEUE_PREFIX.$delay.'Queue';
+        $exchangeDelayed = self::DELAY_QUEUE_PREFIX . $delay . $this->exchangeName;
+        $queueDelayed = self::DELAY_QUEUE_PREFIX . $delay . 'Queue';
 
         // Delay Queue
         $channel->exchange_declare($exchangeDelayed, 'fanout', false, true, true);

src/Cmp/Queue/Domain/Writer/AbstractWriter.php

Spacing +1 added lines, -1 removed lines

@@ -41,7 +41,7 @@
 
         if ($numOfDomainObjects === 1) {
             $this->writeOne($this->messages[0]);
-        } else if($numOfDomainObjects > 1) {
+        } else if ($numOfDomainObjects > 1) {
             $this->writeSome($this->messages);
         }
         $this->messages = [];

src/Cmp/Task/Infrastructure/Consumer/RabbitMQ/RabbitMQConsumerInitializer.php

Spacing +1 added lines, -1 removed lines

@@ -54,7 +54,7 @@
 
             $channel->exchange_declare($this->exchange, 'fanout', false, true, false);
 
-            list($queueName, ,) = $channel->queue_declare($this->queue, false, true, false, false);
+            list($queueName,,) = $channel->queue_declare($this->queue, false, true, false, false);
 
             $channel->queue_bind($queueName, $this->exchange);