@@ -28,7 +28,7 @@ |
||
28 | 28 | */ |
29 | 29 | public function toClosure() |
30 | 30 | { |
31 | - return function (AMQPMessage $message) { |
|
31 | + return function(AMQPMessage $message) { |
|
32 | 32 | $this->getQueueService()->handle($message); |
33 | 33 | }; |
34 | 34 | } |
@@ -129,7 +129,7 @@ |
||
129 | 129 | |
130 | 130 | /** |
131 | 131 | * @param string $msg |
132 | - * @param null $correlationId |
|
132 | + * @param string $correlationId |
|
133 | 133 | * |
134 | 134 | * @return CmobiAMQPMessage |
135 | 135 | */ |
@@ -20,6 +20,10 @@ discard block |
||
20 | 20 | private $correlationId; |
21 | 21 | private $callbackQueue; |
22 | 22 | |
23 | + /** |
|
24 | + * @param string $queueName |
|
25 | + * @param string $fromName |
|
26 | + */ |
|
23 | 27 | public function __construct($queueName, ConnectionManager $manager, $fromName) |
24 | 28 | { |
25 | 29 | $this->queueName = $queueName; |
@@ -55,7 +59,7 @@ discard block |
||
55 | 59 | } |
56 | 60 | |
57 | 61 | /** |
58 | - * @param $data |
|
62 | + * @param string $data |
|
59 | 63 | * @param int $expire |
60 | 64 | * @param int $priority |
61 | 65 | * @throws QueueNotFoundException |
@@ -148,7 +152,7 @@ discard block |
||
148 | 152 | /** |
149 | 153 | * @todo unecessary method set, its only exists to run tests whitout stay jailed in infinite while waiting response. |
150 | 154 | * |
151 | - * @param $content |
|
155 | + * @param string $content |
|
152 | 156 | */ |
153 | 157 | public function setResponse($content) |
154 | 158 | { |
@@ -178,7 +182,7 @@ discard block |
||
178 | 182 | } |
179 | 183 | |
180 | 184 | /** |
181 | - * @return string |
|
185 | + * @return boolean |
|
182 | 186 | */ |
183 | 187 | public function getExchange() |
184 | 188 | { |
@@ -186,7 +190,7 @@ discard block |
||
186 | 190 | } |
187 | 191 | |
188 | 192 | /** |
189 | - * @return string |
|
193 | + * @return boolean |
|
190 | 194 | */ |
191 | 195 | public function getExchangeType() |
192 | 196 | { |
@@ -46,7 +46,7 @@ discard block |
||
46 | 46 | */ |
47 | 47 | public function refreshChannel() |
48 | 48 | { |
49 | - if (! $this->connection->isConnected()) { |
|
49 | + if (!$this->connection->isConnected()) { |
|
50 | 50 | $this->connection->reconnect(); |
51 | 51 | } |
52 | 52 | $this->channel = $this->connection->channel(); |
@@ -66,7 +66,7 @@ discard block |
||
66 | 66 | $this->response = null; |
67 | 67 | $this->refreshChannel(); |
68 | 68 | |
69 | - if (! $this->queueHasExists()) { |
|
69 | + if (!$this->queueHasExists()) { |
|
70 | 70 | throw new QueueNotFoundException("Queue $this->queueName not declared."); |
71 | 71 | } |
72 | 72 | $this->correlationId = $this->generateCorrelationId(); |
@@ -91,7 +91,7 @@ discard block |
||
91 | 91 | [$this, 'onResponse'] |
92 | 92 | ); |
93 | 93 | $msg = new CmobiAMQPMessage( |
94 | - (string) $data, |
|
94 | + (string)$data, |
|
95 | 95 | [ |
96 | 96 | 'correlation_id' => $this->correlationId, |
97 | 97 | 'reply_to' => $this->callbackQueue, |
@@ -100,7 +100,7 @@ discard block |
||
100 | 100 | ); |
101 | 101 | $this->getChannel()->basic_publish($msg, '', $this->getQueueName()); |
102 | 102 | |
103 | - while (! $this->response) { |
|
103 | + while (!$this->response) { |
|
104 | 104 | $this->getChannel()->wait(null, 0, ($expire / 1000)); |
105 | 105 | } |
106 | 106 | $this->getChannel()->close(); |
@@ -166,7 +166,7 @@ discard block |
||
166 | 166 | /** @return string */ |
167 | 167 | public function generateCorrelationId() |
168 | 168 | { |
169 | - return uniqid($this->getQueueName()) . Uuid::uuid4()->toString() . microtime(); |
|
169 | + return uniqid($this->getQueueName()).Uuid::uuid4()->toString().microtime(); |
|
170 | 170 | } |
171 | 171 | |
172 | 172 | /** |
@@ -25,7 +25,7 @@ |
||
25 | 25 | } |
26 | 26 | |
27 | 27 | /** |
28 | - * @param $queueName |
|
28 | + * @param string $queueName |
|
29 | 29 | * @param QueueServiceInterface $queueService |
30 | 30 | * @param QueueBagInterface $queueBag |
31 | 31 | * @return Queue |
@@ -30,7 +30,7 @@ |
||
30 | 30 | */ |
31 | 31 | public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag) |
32 | 32 | { |
33 | - if (! $queueBag instanceof SubscriberQueueBag) { |
|
33 | + if (!$queueBag instanceof SubscriberQueueBag) { |
|
34 | 34 | throw new \Exception('Unsupported QueueBag'); |
35 | 35 | } |
36 | 36 | $queue = new Queue($this->getConnectionManager(), $queueBag, $this->connectionName); |
@@ -25,7 +25,7 @@ |
||
25 | 25 | } |
26 | 26 | |
27 | 27 | /** |
28 | - * @param $queueName |
|
28 | + * @param string $queueName |
|
29 | 29 | * @param QueueServiceInterface $queueService |
30 | 30 | * @param QueueBagInterface $queueBag |
31 | 31 | * @return Queue |
@@ -30,7 +30,7 @@ |
||
30 | 30 | */ |
31 | 31 | public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag) |
32 | 32 | { |
33 | - if (! $queueBag instanceof WorkerQueueBag) { |
|
33 | + if (!$queueBag instanceof WorkerQueueBag) { |
|
34 | 34 | throw new \Exception('Unsupported QueueBag'); |
35 | 35 | } |
36 | 36 | $queue = new Queue($this->getConnectionManager(), $queueBag, $this->connectionName); |
@@ -6,7 +6,6 @@ |
||
6 | 6 | use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnection; |
7 | 7 | use Cmobi\RabbitmqBundle\Connection\ConnectionManager; |
8 | 8 | use Cmobi\RabbitmqBundle\Connection\Exception\InvalidAMQPChannelException; |
9 | -use Psr\Log\LoggerInterface; |
|
10 | 9 | |
11 | 10 | class Queue implements QueueInterface |
12 | 11 | { |