contamobi / CmobiRabbitmqBundle

Transport/Subscriber/Publisher.php

Doc Comments +4 added lines

@@ -17,6 +17,10 @@
     private $exchange;
     private $exchangeType;
 
+    /**
+     * @param string $exchange
+     * @param string $fromName
+     */
     public function __construct(
         $exchange,
         $exchangeType = ExchangeType::FANOUT,

Unused Use Statements -1 removed lines

@@ -2,7 +2,6 @@
 
 namespace Cmobi\RabbitmqBundle\Transport\Subscriber;
 
-use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel;
 use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnectionInterface;
 use Cmobi\RabbitmqBundle\Connection\ConnectionManager;
 use Cmobi\RabbitmqBundle\Queue\CmobiAMQPMessage;

Spacing +1 added lines, -1 removed lines

@@ -43,7 +43,7 @@
         $channel = $connection->channel();
         $queueBag = new SubscriberQueueBag($this->getExchange(), $this->getExchangeType(), $this->getQueueName());
         $channel->exchangeDeclare($queueBag->getExchangeDeclare());
-        $msg = new CmobiAMQPMessage((string) $data);
+        $msg = new CmobiAMQPMessage((string)$data);
         $channel->basic_publish($msg, $queueBag->getExchange());
 
         $channel->close();

Transport/Subscriber/SubscriberBuilder.php

Doc Comments +1 added lines, -1 removed lines

@@ -25,7 +25,7 @@
     }
 
     /**
-     * @param $queueName
+     * @param string $queueName
      * @param QueueServiceInterface $queueService
      * @param QueueBagInterface $queueBag
      * @return Queue

Spacing +1 added lines, -1 removed lines

@@ -30,7 +30,7 @@
      */
     public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag)
     {
-        if (! $queueBag instanceof SubscriberQueueBag) {
+        if (!$queueBag instanceof SubscriberQueueBag) {
             throw new \Exception('Unsupported QueueBag');
         }
         $queue = new Queue($this->getConnectionManager(), $queueBag, $this->connectionName);

Transport/Worker/Task.php

Doc Comments +6 added lines, -2 removed lines

@@ -16,6 +16,10 @@ 
     private $fromName;
     private $queueName;
 
+    /**
+     * @param string $queueName
+     * @param string $fromName
+     */
     public function __construct($queueName, ConnectionManager $manager, $fromName)
     {
         $this->queueName = $queueName;
@@ -107,7 +111,7 @@ 
     }
 
     /**
-     * @return string
+     * @return boolean
      */
     public function getExchange()
     {
@@ -115,7 +119,7 @@ 
     }
 
     /**
-     * @return string
+     * @return boolean
      */
     public function getExchangeType()
     {

Spacing +2 added lines, -2 removed lines

@@ -35,13 +35,13 @@
         $connection = $this->connectionManager->getConnection();
         $channel = $connection->channel();
 
-        if (! $this->queueHasExists($channel)) {
+        if (!$this->queueHasExists($channel)) {
             throw new QueueNotFoundException("Queue $this->queueName not declared.");
         }
         $queueBag = new WorkerQueueBag($this->getQueueName());
         $channel->queueDeclare($queueBag->getQueueDeclare());
         $msg = new CmobiAMQPMessage(
-            (string) $data,
+            (string)$data,
             [
                 'delivery_mode' => 2, // make message persistent
                 'priority' => $priority,

Transport/Worker/WorkerBuilder.php

Doc Comments +1 added lines, -1 removed lines

@@ -25,7 +25,7 @@
     }
 
     /**
-     * @param $queueName
+     * @param string $queueName
      * @param QueueServiceInterface $queueService
      * @param QueueBagInterface $queueBag
      * @return Queue

Spacing +1 added lines, -1 removed lines

@@ -30,7 +30,7 @@
      */
     public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag)
     {
-        if (! $queueBag instanceof WorkerQueueBag) {
+        if (!$queueBag instanceof WorkerQueueBag) {
             throw new \Exception('Unsupported QueueBag');
         }
         $queue = new Queue($this->getConnectionManager(), $queueBag, $this->connectionName);

Queue/Queue.php

Unused Use Statements -1 removed lines

@@ -6,7 +6,6 @@
 use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnection;
 use Cmobi\RabbitmqBundle\Connection\ConnectionManager;
 use Cmobi\RabbitmqBundle\Connection\Exception\InvalidAMQPChannelException;
-use Psr\Log\LoggerInterface;
 
 class Queue implements QueueInterface
 {

Spacing +3 added lines, -3 removed lines

@@ -145,17 +145,17 @@
         do {
             try {
                 $failed = false;
-                fwrite($this->logOutput, 'start Queue::forceReconnect() - trying connect...' . PHP_EOL);
+                fwrite($this->logOutput, 'start Queue::forceReconnect() - trying connect...'.PHP_EOL);
                 $this->connection = $this->getConnectionManager()->getConnection($this->connectionName);
                 $this->channel = $this->getConnection()->channel();
                 $this->createQueue();
             } catch (\Exception $e) {
                 $failed = true;
                 sleep(3);
-                fwrite($this->errOutput, 'failed Queue::forceReconnect() - ' . $e->getMessage() . PHP_EOL);
+                fwrite($this->errOutput, 'failed Queue::forceReconnect() - '.$e->getMessage().PHP_EOL);
             }
         } while ($failed);
-        fwrite($this->logOutput, 'Queue::forceReconnect() - connected!' . PHP_EOL);
+        fwrite($this->logOutput, 'Queue::forceReconnect() - connected!'.PHP_EOL);
 
         return $this->channel;
     }

Transport/Rpc/RpcClient.php

Doc Comments +8 added lines, -4 removed lines

@@ -22,6 +22,10 @@ 
     private $correlationId;
     private $callbackQueue;
 
+    /**
+     * @param string $queueName
+     * @param string $fromName
+     */
     public function __construct($queueName, ConnectionManager $manager, $fromName, $connectionName = 'default')
     {
         $this->connectionName = $connectionName;
@@ -43,7 +47,7 @@ 
     }
 
     /**
-     * @param $data
+     * @param string $data
      * @param int $expire
      * @param int $priority
      * @throws QueueNotFoundException
@@ -133,7 +137,7 @@ 
     /**
      * @todo unecessary method set, its only exists to run tests whitout stay jailed in infinite while waiting response.
      *
-     * @param $content
+     * @param string $content
      */
     public function setResponse($content)
     {
@@ -163,7 +167,7 @@ 
     }
 
     /**
-     * @return string
+     * @return boolean
      */
     public function getExchange()
     {
@@ -171,7 +175,7 @@ 
     }
 
     /**
-     * @return string
+     * @return boolean
      */
     public function getExchangeType()
     {

Spacing +8 added lines, -8 removed lines

@@ -83,13 +83,13 @@ 
         $connection = $this->connectionManager->getConnection($this->connectionName);
         $channel = $connection->channel();
 
-        if (! $this->queueHasExists($channel)) {
+        if (!$this->queueHasExists($channel)) {
             throw new QueueNotFoundException("Queue $this->queueName not declared.");
         }
-        $sufix = Uuid::uuid4()->toString() . microtime();
+        $sufix = Uuid::uuid4()->toString().microtime();
         $this->createCallbackQueue($channel, $expire, $sufix);
         $msg = new CmobiAMQPMessage(
-            (string) $data,
+            (string)$data,
             [
                 'correlation_id' => $this->correlationId,
                 'reply_to' => $this->callbackQueue,
@@ -98,7 +98,7 @@ 
         );
         $channel->basic_publish($msg, '', $this->getQueueName());
 
-        while (! $this->response) {
+        while (!$this->response) {
             try {
                 $channel->wait(null, 0, ($expire / 1000));
             } catch (\Exception $e) {
@@ -168,7 +168,7 @@ 
     /** @return string */
     public function generateCorrelationId()
     {
-        return uniqid($this->getQueueName()) . Uuid::uuid4()->toString() . microtime();
+        return uniqid($this->getQueueName()).Uuid::uuid4()->toString().microtime();
     }
 
     /**
@@ -215,16 +215,16 @@ 
         try {
             $connection->close();
             sleep(2);
-            fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...' . PHP_EOL);
+            fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...'.PHP_EOL);
             $connection = $this->getConnectionManager()->getConnection($this->connectionName);
             $channel = $connection->channel();
             $this->createCallbackQueue($channel, $expire, $sufix, $corralationId);
         } catch (\Exception $e) {
-            fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - ' . $e->getMessage() . PHP_EOL);
+            fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - '.$e->getMessage().PHP_EOL);
 
             return $connection;
         }
-        fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!' . PHP_EOL);
+        fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!'.PHP_EOL);
 
         return $connection;
     }

Unused Use Statements -2 removed lines

@@ -3,13 +3,11 @@
 namespace Cmobi\RabbitmqBundle\Transport\Rpc;
 
 use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel;
-use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnection;
 use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnectionInterface;
 use Cmobi\RabbitmqBundle\Connection\ConnectionManager;
 use Cmobi\RabbitmqBundle\Queue\CmobiAMQPMessage;
 use Cmobi\RabbitmqBundle\Queue\QueueProducerInterface;
 use Cmobi\RabbitmqBundle\Transport\Exception\QueueNotFoundException;
-use PhpAmqpLib\Exception\AMQPTimeoutException;
 use PhpAmqpLib\Message\AMQPMessage;
 use Ramsey\Uuid\Uuid;