@@ -17,6 +17,10 @@ |
||
17 | 17 | private $exchange; |
18 | 18 | private $exchangeType; |
19 | 19 | |
20 | + /** |
|
21 | + * @param string $exchange |
|
22 | + * @param string $fromName |
|
23 | + */ |
|
20 | 24 | public function __construct( |
21 | 25 | $exchange, |
22 | 26 | $exchangeType = ExchangeType::FANOUT, |
@@ -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(); |
@@ -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); |
@@ -16,6 +16,10 @@ discard block |
||
16 | 16 | private $fromName; |
17 | 17 | private $queueName; |
18 | 18 | |
19 | + /** |
|
20 | + * @param string $queueName |
|
21 | + * @param string $fromName |
|
22 | + */ |
|
19 | 23 | public function __construct($queueName, ConnectionManager $manager, $fromName) |
20 | 24 | { |
21 | 25 | $this->queueName = $queueName; |
@@ -107,7 +111,7 @@ discard block |
||
107 | 111 | } |
108 | 112 | |
109 | 113 | /** |
110 | - * @return string |
|
114 | + * @return boolean |
|
111 | 115 | */ |
112 | 116 | public function getExchange() |
113 | 117 | { |
@@ -115,7 +119,7 @@ discard block |
||
115 | 119 | } |
116 | 120 | |
117 | 121 | /** |
118 | - * @return string |
|
122 | + * @return boolean |
|
119 | 123 | */ |
120 | 124 | public function getExchangeType() |
121 | 125 | { |
@@ -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, |
@@ -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 | { |
@@ -145,17 +145,17 @@ |
||
145 | 145 | do { |
146 | 146 | try { |
147 | 147 | $failed = false; |
148 | - fwrite($this->logOutput, 'start Queue::forceReconnect() - trying connect...' . PHP_EOL); |
|
148 | + fwrite($this->logOutput, 'start Queue::forceReconnect() - trying connect...'.PHP_EOL); |
|
149 | 149 | $this->connection = $this->getConnectionManager()->getConnection($this->connectionName); |
150 | 150 | $this->channel = $this->getConnection()->channel(); |
151 | 151 | $this->createQueue(); |
152 | 152 | } catch (\Exception $e) { |
153 | 153 | $failed = true; |
154 | 154 | sleep(3); |
155 | - fwrite($this->errOutput, 'failed Queue::forceReconnect() - ' . $e->getMessage() . PHP_EOL); |
|
155 | + fwrite($this->errOutput, 'failed Queue::forceReconnect() - '.$e->getMessage().PHP_EOL); |
|
156 | 156 | } |
157 | 157 | } while ($failed); |
158 | - fwrite($this->logOutput, 'Queue::forceReconnect() - connected!' . PHP_EOL); |
|
158 | + fwrite($this->logOutput, 'Queue::forceReconnect() - connected!'.PHP_EOL); |
|
159 | 159 | |
160 | 160 | return $this->channel; |
161 | 161 | } |
@@ -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; |
@@ -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; |
@@ -71,7 +75,7 @@ discard block |
||
71 | 75 | } |
72 | 76 | |
73 | 77 | /** |
74 | - * @param $data |
|
78 | + * @param string $data |
|
75 | 79 | * @param int $expire |
76 | 80 | * @param int $priority |
77 | 81 | * @throws QueueNotFoundException |
@@ -150,7 +154,7 @@ discard block |
||
150 | 154 | /** |
151 | 155 | * @todo unecessary method set, its only exists to run tests whitout stay jailed in infinite while waiting response. |
152 | 156 | * |
153 | - * @param $content |
|
157 | + * @param string $content |
|
154 | 158 | */ |
155 | 159 | public function setResponse($content) |
156 | 160 | { |
@@ -180,7 +184,7 @@ discard block |
||
180 | 184 | } |
181 | 185 | |
182 | 186 | /** |
183 | - * @return string |
|
187 | + * @return boolean |
|
184 | 188 | */ |
185 | 189 | public function getExchange() |
186 | 190 | { |
@@ -188,7 +192,7 @@ discard block |
||
188 | 192 | } |
189 | 193 | |
190 | 194 | /** |
191 | - * @return string |
|
195 | + * @return boolean |
|
192 | 196 | */ |
193 | 197 | public function getExchangeType() |
194 | 198 | { |
@@ -205,8 +209,8 @@ discard block |
||
205 | 209 | |
206 | 210 | /** |
207 | 211 | * @param CmobiAMQPConnectionInterface $connection |
208 | - * @param $expire |
|
209 | - * @param $sufix |
|
212 | + * @param integer $expire |
|
213 | + * @param string $sufix |
|
210 | 214 | * @param $corralationId |
211 | 215 | * @return CmobiAMQPConnectionInterface |
212 | 216 | */ |
@@ -83,13 +83,13 @@ discard block |
||
83 | 83 | $connection = $this->connectionManager->getConnection($this->connectionName); |
84 | 84 | $channel = $connection->channel(); |
85 | 85 | |
86 | - if (! $this->queueHasExists($channel)) { |
|
86 | + if (!$this->queueHasExists($channel)) { |
|
87 | 87 | throw new QueueNotFoundException("Queue $this->queueName not declared."); |
88 | 88 | } |
89 | - $sufix = Uuid::uuid4()->toString() . microtime(); |
|
89 | + $sufix = Uuid::uuid4()->toString().microtime(); |
|
90 | 90 | $this->createCallbackQueue($channel, $expire, $sufix); |
91 | 91 | $msg = new CmobiAMQPMessage( |
92 | - (string) $data, |
|
92 | + (string)$data, |
|
93 | 93 | [ |
94 | 94 | 'correlation_id' => $this->correlationId, |
95 | 95 | 'reply_to' => $this->callbackQueue, |
@@ -98,7 +98,7 @@ discard block |
||
98 | 98 | ); |
99 | 99 | $channel->basic_publish($msg, '', $this->getQueueName()); |
100 | 100 | |
101 | - while (! $this->response) { |
|
101 | + while (!$this->response) { |
|
102 | 102 | try { |
103 | 103 | $channel->wait(null, 0, ($expire / 1000)); |
104 | 104 | } catch (\Exception $e) { |
@@ -168,7 +168,7 @@ discard block |
||
168 | 168 | /** @return string */ |
169 | 169 | public function generateCorrelationId() |
170 | 170 | { |
171 | - return uniqid($this->getQueueName()) . Uuid::uuid4()->toString() . microtime(); |
|
171 | + return uniqid($this->getQueueName()).Uuid::uuid4()->toString().microtime(); |
|
172 | 172 | } |
173 | 173 | |
174 | 174 | /** |
@@ -215,16 +215,16 @@ discard block |
||
215 | 215 | try { |
216 | 216 | $connection->close(); |
217 | 217 | sleep(2); |
218 | - fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...' . PHP_EOL); |
|
218 | + fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...'.PHP_EOL); |
|
219 | 219 | $connection = $this->getConnectionManager()->getConnection($this->connectionName); |
220 | 220 | $channel = $connection->channel(); |
221 | 221 | $this->createCallbackQueue($channel, $expire, $sufix, $corralationId); |
222 | 222 | } catch (\Exception $e) { |
223 | - fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - ' . $e->getMessage() . PHP_EOL); |
|
223 | + fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - '.$e->getMessage().PHP_EOL); |
|
224 | 224 | |
225 | 225 | return $connection; |
226 | 226 | } |
227 | - fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!' . PHP_EOL); |
|
227 | + fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!'.PHP_EOL); |
|
228 | 228 | |
229 | 229 | return $connection; |
230 | 230 | } |