CMProductions / queues

src/Cmp/Queues/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/Cmp/Queues/Infrastructure/AmqpLib/v26/RabbitMQ/Queue/DelayedQueueWriter.php

Spacing +10 added lines, -10 removed lines

@@ -68,7 +68,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;
     }
@@ -82,15 +82,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 delayed messages: '.$exception->getMessage());
+        } catch (\Exception $exception) {
+            $this->logger->error('Error writing delayed messages: ' . $exception->getMessage());
             throw new WriterException($exception->getMessage(), $exception->getCode());
         }
     }
@@ -100,13 +100,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,
@@ -121,8 +121,8 @@ 
                 ]
             );
             $this->channel->queue_bind($delayedQueue, $this->delayedExchangeName);
-        } catch(\Exception $exception) {
-            $this->logger->error('Error configuring delayed queues: '.$exception->getMessage());
+        } catch (\Exception $exception) {
+            $this->logger->error('Error configuring delayed queues: ' . $exception->getMessage());
             throw new WriterException($exception->getMessage(), $exception->getCode());
         }
     }

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

Spacing +7 added lines, -7 removed lines

@@ -64,8 +64,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;
                 }
@@ -75,8 +75,8 @@ 
                 $this->channel->batch_basic_publish($msg, $this->exchangeConfig->getName(), $message->getName());
             }
             $this->channel->publish_batch();
-            foreach($messagesWithDelay as $delay => $delayedMessages) {
-                if(!isset($this->delayedQueueWriterRegistry[$delay])) {
+            foreach ($messagesWithDelay as $delay => $delayedMessages) {
+                if (!isset($this->delayedQueueWriterRegistry[$delay])) {
                     $this->delayedQueueWriterRegistry[$delay] = new DelayedQueueWriter(
                         $this->exchangeConfig->getName(),
                         $delay,
@@ -86,8 +86,8 @@ 
                 }
                 $this->delayedQueueWriterRegistry[$delay]->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());
         }
     }
@@ -97,7 +97,7 @@ 
      */
     protected function initialize()
     {
-        if($this->channel) {
+        if ($this->channel) {
             return;
         }
         $this->logger->info('Connecting to RabbitMQ');

src/Cmp/Queues/Domain/Event/DomainEvent.php

Indentation +2 added lines, -2 removed lines

@@ -36,8 +36,8 @@
     public function __construct($origin, $name, $occurredOn, array $body = [])
     {
         $this->setOrigin($origin)
-             ->setName($name)
-             ->setOccurredOn($occurredOn)
+                ->setName($name)
+                ->setOccurredOn($occurredOn)
         ;
         $this->body = $body;
     }

Spacing +3 added lines, -3 removed lines

@@ -91,7 +91,7 @@ 
      */
     protected function setOrigin($origin)
     {
-        if(empty($origin)) {
+        if (empty($origin)) {
             throw new DomainEventException('DomainEvent origin cannot be empty');
         }
         $this->origin = $origin;
@@ -105,7 +105,7 @@ 
      */
     protected function setName($name)
     {
-        if(empty($name)) {
+        if (empty($name)) {
             throw new DomainEventException('DomainEvent name cannot be empty');
         }
         $this->name = $name;
@@ -119,7 +119,7 @@ 
      */
     protected function setOccurredOn($occurredOn)
     {
-        if(!is_null($occurredOn) && !preg_match('/^\d+(\.\d{1,4})?$/', $occurredOn)) { // accepts also microseconds
+        if (!is_null($occurredOn) && !preg_match('/^\d+(\.\d{1,4})?$/', $occurredOn)) { // accepts also microseconds
             throw new DomainEventException("$occurredOn is not a valid unix timestamp.");
         }
         $this->occurredOn = $occurredOn;

src/Cmp/Queues/Domain/Event/Publisher.php

Spacing +1 added lines, -1 removed lines

@@ -41,7 +41,7 @@
      */
     public function publish()
     {
-        if(!isset($this->events[0])) {
+        if (!isset($this->events[0])) {
             throw new DomainEventException('You must add at least 1 DomainEvent in order to publish to queue.');
         }
         $this->queueWriter->write($this->events);

src/Cmp/Queues/Domain/Task/Task.php

Indentation +1 added lines, -1 removed lines

@@ -31,7 +31,7 @@
     public function __construct($name, array $body, $delay=0)
     {
         $this->setName($name)
-             ->setDelay($delay)
+                ->setDelay($delay)
         ;
         $this->body = $body;
     }

Spacing +3 added lines, -3 removed lines

@@ -28,7 +28,7 @@ 
      * @param array $body
      * @param int $delay
      */
-    public function __construct($name, array $body, $delay=0)
+    public function __construct($name, array $body, $delay = 0)
     {
         $this->setName($name)
              ->setDelay($delay)
@@ -67,7 +67,7 @@ 
      */
     protected function setName($name)
     {
-        if(empty($name)) {
+        if (empty($name)) {
             throw new TaskException('Task name cannot be empty');
         }
         $this->name = $name;
@@ -81,7 +81,7 @@ 
      */
     protected function setDelay($delay)
     {
-        if(!is_null($delay) && !preg_match('/^\d+$/', $delay)) {
+        if (!is_null($delay) && !preg_match('/^\d+$/', $delay)) {
             throw new TaskException("Task delay $delay is not a valid delay.");
         }
         $this->delay = $delay;

src/Cmp/Queues/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);

src/Cmp/Queues/Domain/Queue/QueueReader.php

Spacing +1 added lines, -1 removed lines

@@ -17,7 +17,7 @@
      * @throws ReaderException
      * @return void
      */
-    public function read(callable $callback, $timeout=0);
+    public function read(callable $callback, $timeout = 0);
 
     /**
      * Deletes all messages from the queue

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

Unused Use Statements -1 removed lines

@@ -10,7 +10,6 @@
 use Cmp\Queues\Infrastructure\AmqpLib\v26\RabbitMQ\Queue\Config\QueueConfig;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AMQPLazyConnection;
-use PhpAmqpLib\Exception\AMQPRuntimeException;
 use PhpAmqpLib\Exception\AMQPTimeoutException;
 use Psr\Log\LoggerInterface;
 

Spacing +5 added lines, -5 removed lines

@@ -96,17 +96,17 @@ 
      * @throws ReaderException
      * @throws TimeoutReaderException
      */
-    public function read(callable $callback, $timeout=0)
+    public function read(callable $callback, $timeout = 0)
     {
         $this->initialize();
         $this->messageHandler->setCallback($callback);
 
         try {
             $this->consume($timeout);
-        } catch(AMQPTimeoutException $e) {
+        } catch (AMQPTimeoutException $e) {
             $this->stopConsuming();
             throw new TimeoutReaderException("Timed out at $timeout seconds while reading.", 0, $e);
-        } catch(\Exception $e) {
+        } catch (\Exception $e) {
             $this->stopConsuming();
             throw new ReaderException("Error occurred while reading", 0, $e);
         }
@@ -165,7 +165,7 @@ 
     protected function consume($timeout)
     {
         if ($this->consumerTag === '') {
-            $this->logger->debug('Waiting for messages on queue:'.$this->queueConfig->getName());
+            $this->logger->debug('Waiting for messages on queue:' . $this->queueConfig->getName());
             $this->consumerTag = $this->channel->basic_consume(
                 $this->queueConfig->getName(),
                 '',
@@ -206,7 +206,7 @@ 
         if ($this->consumerTag) {
             try {
                 $this->channel->basic_cancel($this->consumerTag);
-            } catch(\Exception $e) {
+            } catch (\Exception $e) {
             }
 
             $this->consumerTag = '';