CMProductions / queues

src/Infrastructure/AmqpLib/v26/RabbitMQ/Queue/MessageHandler.php

Doc Comments -1 removed lines

@@ -26,7 +26,6 @@
     /**
      * MessageHandler constructor.
      * @param JSONMessageFactory $jsonMessageFactory
-     * @param callable $callback
      */
     public function __construct(JSONMessageFactory $jsonMessageFactory)
     {

src/Infrastructure/AmqpLib/v26/RabbitMQ/Queue/QueueReader.php

Spacing +1 added lines, -1 removed lines

@@ -117,7 +117,7 @@
      */
     protected function initialize()
     {
-        if($this->channel) {
+        if ($this->channel) {
             return;
         }
         $this->logger->info('Connecting to RabbitMQ');

src/Infrastructure/AmqpLib/v26/RabbitMQ/Queue/Config/ConnectionConfig.php

Spacing +1 added lines, -1 removed lines

@@ -16,7 +16,7 @@
     protected $password;
     protected $vHost;
 
-    public function __construct($host, $port, $user, $password, $vHost='/')
+    public function __construct($host, $port, $user, $password, $vHost = '/')
     {
         $this->host = $host;
         $this->port = $port;

src/Infrastructure/AmqpLib/v26/RabbitMQ/Queue/DelayedQueueWriter.php

Spacing +10 added lines, -10 removed lines

@@ -67,7 +67,7 @@ 
     {
         $this->delay = $delay;
         $this->exchangeName = $exchangeName;
-        $this->delayedExchangeName = self::DELAY_QUEUE_PREFIX.$this->delay.$this->exchangeName;
+        $this->delayedExchangeName = self::DELAY_QUEUE_PREFIX . $this->delay . $this->exchangeName;
         $this->logger = $logger;
         $this->channel = $channel;
     }
@@ -81,15 +81,15 @@ 
     {
         $this->initialize();
         try {
-            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));
                 $this->channel->batch_basic_publish($msg, $this->delayedExchangeName, $message->getName());
             }
             $this->channel->publish_batch();
-        } catch(\Exception $exception) {
-            $this->logger->error('Error writing messages: '.$exception->getMessage());
+        } catch (\Exception $exception) {
+            $this->logger->error('Error writing messages: ' . $exception->getMessage());
             throw new WriterException($exception->getMessage(), $exception->getCode());
         }
     }
@@ -99,13 +99,13 @@ 
      */
     protected function initialize()
     {
-        try{
-            $delayedQueue = self::DELAY_QUEUE_PREFIX.$this->delay.'Queue';
+        try {
+            $delayedQueue = self::DELAY_QUEUE_PREFIX . $this->delay . 'Queue';
 
-            $this->logger->info('Creating delayed exchange '.$this->delayedExchangeName);
+            $this->logger->info('Creating delayed exchange ' . $this->delayedExchangeName);
             // Delay Queue
             $this->channel->exchange_declare($this->delayedExchangeName, 'fanout', false, true, true);
-            $this->logger->info('Creating delayed queue '.$delayedQueue);
+            $this->logger->info('Creating delayed queue ' . $delayedQueue);
             $this->channel->queue_declare(
                 $delayedQueue,
                 false,
@@ -120,8 +120,8 @@ 
                 ]
             );
             $this->channel->queue_bind($delayedQueue, $this->delayedExchangeName);
-        } catch(\Exception $exception) {
-            $this->logger->error('Error writing delayed messages: '.$exception->getMessage());
+        } catch (\Exception $exception) {
+            $this->logger->error('Error writing delayed messages: ' . $exception->getMessage());
             throw new WriterException($exception->getMessage(), $exception->getCode());
         }
     }

src/Infrastructure/AmqpLib/v26/RabbitMQ/Queue/QueueWriter.php

Spacing +6 added lines, -6 removed lines

@@ -59,8 +59,8 @@ 
         $this->initialize();
         try {
             $messagesWithDelay = [];
-            foreach($messages as $message) {
-                if($message->getDelay() > 0) {
+            foreach ($messages as $message) {
+                if ($message->getDelay() > 0) {
                     $messagesWithDelay[$message->getDelay()][] = $message;
                     continue;
                 }
@@ -70,7 +70,7 @@ 
                 $this->channel->batch_basic_publish($msg, $this->exchangeConfig->getName(), $message->getName());
             }
             $this->channel->publish_batch();
-            foreach($messagesWithDelay as $delay => $delayedMessages) {
+            foreach ($messagesWithDelay as $delay => $delayedMessages) {
                 $delayedQueueWriter = new DelayedQueueWriter(
                     $this->exchangeConfig->getName(),
                     $delay,
@@ -79,8 +79,8 @@ 
                 );
                 $delayedQueueWriter->write($delayedMessages);
             }
-        } catch(\Exception $exception) {
-            $this->logger->error('Error writing messages: '.$exception->getMessage());
+        } catch (\Exception $exception) {
+            $this->logger->error('Error writing messages: ' . $exception->getMessage());
             throw new WriterException($exception->getMessage(), $exception->getCode());
         }
     }
@@ -90,7 +90,7 @@ 
      */
     protected function initialize()
     {
-        if($this->channel) {
+        if ($this->channel) {
             return;
         }
         $this->logger->info('Connecting to RabbitMQ');

src/Infrastructure/AmqpLib/v26/RabbitMQ/DomainEvent/Subscriber.php

Spacing +1 added lines, -1 removed lines

@@ -49,7 +49,7 @@
         $messageHandler->setCallback(array($this, 'notify'));
         $queueReader = new QueueReader(
             new AMQPLazyConnection($host, $port, $user, $password, $vHost),
-            new QueueConfig(uniqid($queueName.'_', true), false, false, true, true),
+            new QueueConfig(uniqid($queueName . '_', true), false, false, true, true),
             new ExchangeConfig($exchangeName, 'topic', false, true, false),
             $bindConfig,
             new ConsumeConfig(false, false, true, false),

src/Domain/Task/Consumer.php

Spacing +1 added lines, -1 removed lines

@@ -21,7 +21,7 @@
 
     public function consume(callable $callback)
     {
-        while(true) {
+        while (true) {
             $this->consumeOnce($callback);
         }
     }

src/Domain/Task/Task.php

Spacing +1 added lines, -1 removed lines

@@ -27,7 +27,7 @@
      * @param array $body
      * @param int $delay
      */
-    public function __construct($name, array $body, $delay=0)
+    public function __construct($name, array $body, $delay = 0)
     {
         $this->name = $name;
         $this->body = $body;

src/Domain/Task/Producer.php

Spacing +1 added lines, -1 removed lines

@@ -40,7 +40,7 @@
      */
     public function produce()
     {
-        if(!isset($this->tasks[0])) {
+        if (!isset($this->tasks[0])) {
             throw new TaskException('You must add at least 1 Task before producing.');
         }
         $this->queueWriter->write($this->tasks);