@@ -24,6 +24,10 @@ discard block |
||
| 24 | 24 | private $correlationId; |
| 25 | 25 | private $callbackQueue; |
| 26 | 26 | |
| 27 | + /** |
|
| 28 | + * @param string $queueName |
|
| 29 | + * @param string $fromName |
|
| 30 | + */ |
|
| 27 | 31 | public function __construct($queueName, ConnectionManager $manager, $fromName, $connectionName = 'default') |
| 28 | 32 | { |
| 29 | 33 | $this->connectionName = $connectionName; |
@@ -74,7 +78,7 @@ discard block |
||
| 74 | 78 | } |
| 75 | 79 | |
| 76 | 80 | /** |
| 77 | - * @param $data |
|
| 81 | + * @param string $data |
|
| 78 | 82 | * @param int $expire |
| 79 | 83 | * @param int $priority |
| 80 | 84 | * @throws QueueNotFoundException |
@@ -151,7 +155,7 @@ discard block |
||
| 151 | 155 | /** |
| 152 | 156 | * @todo unecessary method set, its only exists to run tests whitout stay jailed in infinite while waiting response. |
| 153 | 157 | * |
| 154 | - * @param $content |
|
| 158 | + * @param string $content |
|
| 155 | 159 | */ |
| 156 | 160 | public function setResponse($content) |
| 157 | 161 | { |
@@ -181,7 +185,7 @@ discard block |
||
| 181 | 185 | } |
| 182 | 186 | |
| 183 | 187 | /** |
| 184 | - * @return string |
|
| 188 | + * @return boolean |
|
| 185 | 189 | */ |
| 186 | 190 | public function getExchange() |
| 187 | 191 | { |
@@ -189,7 +193,7 @@ discard block |
||
| 189 | 193 | } |
| 190 | 194 | |
| 191 | 195 | /** |
| 192 | - * @return string |
|
| 196 | + * @return boolean |
|
| 193 | 197 | */ |
| 194 | 198 | public function getExchangeType() |
| 195 | 199 | { |
@@ -206,7 +210,7 @@ discard block |
||
| 206 | 210 | |
| 207 | 211 | /** |
| 208 | 212 | * @param CmobiAMQPConnectionInterface $connection |
| 209 | - * @param $expire |
|
| 213 | + * @param integer $expire |
|
| 210 | 214 | * @param $corralationId |
| 211 | 215 | * @return CmobiAMQPConnectionInterface |
| 212 | 216 | */ |
@@ -3,7 +3,6 @@ |
||
| 3 | 3 | namespace Cmobi\RabbitmqBundle\Transport\Rpc; |
| 4 | 4 | |
| 5 | 5 | use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel; |
| 6 | -use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnection; |
|
| 7 | 6 | use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnectionInterface; |
| 8 | 7 | use Cmobi\RabbitmqBundle\Connection\ConnectionManager; |
| 9 | 8 | use Cmobi\RabbitmqBundle\Queue\CmobiAMQPMessage; |
@@ -86,11 +86,11 @@ discard block |
||
| 86 | 86 | $connection = $this->connectionManager->getConnection($this->connectionName); |
| 87 | 87 | $channel = $connection->channel(); |
| 88 | 88 | |
| 89 | - if (! $this->queueHasExists($channel)) { |
|
| 89 | + if (!$this->queueHasExists($channel)) { |
|
| 90 | 90 | throw new QueueNotFoundException("Queue $this->queueName not declared."); |
| 91 | 91 | } |
| 92 | 92 | $msg = new CmobiAMQPMessage( |
| 93 | - (string) $data, |
|
| 93 | + (string)$data, |
|
| 94 | 94 | [ |
| 95 | 95 | 'correlation_id' => $this->correlationId, |
| 96 | 96 | 'reply_to' => $this->callbackQueue, |
@@ -99,7 +99,7 @@ discard block |
||
| 99 | 99 | ); |
| 100 | 100 | $channel->basic_publish($msg, '', $this->getQueueName()); |
| 101 | 101 | |
| 102 | - while (! $this->response) { |
|
| 102 | + while (!$this->response) { |
|
| 103 | 103 | try { |
| 104 | 104 | $channel->wait(null, 0, ($expire / 1000)); |
| 105 | 105 | } catch (\Exception $e) { |
@@ -169,7 +169,7 @@ discard block |
||
| 169 | 169 | /** @return string */ |
| 170 | 170 | public function generateCorrelationId() |
| 171 | 171 | { |
| 172 | - return uniqid($this->getQueueName()) . Uuid::uuid4()->toString() . microtime(); |
|
| 172 | + return uniqid($this->getQueueName()).Uuid::uuid4()->toString().microtime(); |
|
| 173 | 173 | } |
| 174 | 174 | |
| 175 | 175 | /** |
@@ -216,17 +216,17 @@ discard block |
||
| 216 | 216 | try { |
| 217 | 217 | $connection->close(); |
| 218 | 218 | $failed = false; |
| 219 | - fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...' . PHP_EOL); |
|
| 219 | + fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...'.PHP_EOL); |
|
| 220 | 220 | $connection = $this->getConnectionManager()->getConnection($this->connectionName); |
| 221 | 221 | $channel = $connection->channel(); |
| 222 | 222 | $this->createCallbackQueue($channel, $expire, $corralationId); |
| 223 | 223 | } catch (\Exception $e) { |
| 224 | 224 | $failed = true; |
| 225 | 225 | sleep(3); |
| 226 | - fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - ' . $e->getMessage() . PHP_EOL); |
|
| 226 | + fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - '.$e->getMessage().PHP_EOL); |
|
| 227 | 227 | } |
| 228 | 228 | } while ($failed); |
| 229 | - fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!' . PHP_EOL); |
|
| 229 | + fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!'.PHP_EOL); |
|
| 230 | 230 | |
| 231 | 231 | return $connection; |
| 232 | 232 | } |
@@ -2,7 +2,6 @@ |
||
| 2 | 2 | |
| 3 | 3 | namespace Cmobi\RabbitmqBundle\Transport\Subscriber; |
| 4 | 4 | |
| 5 | -use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel; |
|
| 6 | 5 | use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnectionInterface; |
| 7 | 6 | use Cmobi\RabbitmqBundle\Connection\ConnectionManager; |
| 8 | 7 | use Cmobi\RabbitmqBundle\Queue\CmobiAMQPMessage; |
@@ -43,7 +43,7 @@ |
||
| 43 | 43 | $channel = $connection->channel(); |
| 44 | 44 | $queueBag = new SubscriberQueueBag($this->getExchange(), $this->getExchangeType(), $this->getQueueName()); |
| 45 | 45 | $channel->exchangeDeclare($queueBag->getExchangeDeclare()); |
| 46 | - $msg = new CmobiAMQPMessage((string) $data); |
|
| 46 | + $msg = new CmobiAMQPMessage((string)$data); |
|
| 47 | 47 | $channel->basic_publish($msg, $queueBag->getExchange()); |
| 48 | 48 | |
| 49 | 49 | $channel->close(); |
@@ -35,13 +35,13 @@ |
||
| 35 | 35 | $connection = $this->connectionManager->getConnection(); |
| 36 | 36 | $channel = $connection->channel(); |
| 37 | 37 | |
| 38 | - if (! $this->queueHasExists($channel)) { |
|
| 38 | + if (!$this->queueHasExists($channel)) { |
|
| 39 | 39 | throw new QueueNotFoundException("Queue $this->queueName not declared."); |
| 40 | 40 | } |
| 41 | 41 | $queueBag = new WorkerQueueBag($this->getQueueName()); |
| 42 | 42 | $channel->queueDeclare($queueBag->getQueueDeclare()); |
| 43 | 43 | $msg = new CmobiAMQPMessage( |
| 44 | - (string) $data, |
|
| 44 | + (string)$data, |
|
| 45 | 45 | [ |
| 46 | 46 | 'delivery_mode' => 2, // make message persistent |
| 47 | 47 | 'priority' => $priority, |