@@ -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 | } |
@@ -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,12 +86,12 @@ 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 | $this->createCallbackQueue($channel, $expire); |
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 | $channel->basic_publish($msg, '', $this->getQueueName()); |
102 | 102 | |
103 | - while (! $this->response) { |
|
103 | + while (!$this->response) { |
|
104 | 104 | try { |
105 | 105 | $channel->wait(null, 0, ($expire / 1000)); |
106 | 106 | } catch (\Exception $e) { |
@@ -170,7 +170,7 @@ discard block |
||
170 | 170 | /** @return string */ |
171 | 171 | public function generateCorrelationId() |
172 | 172 | { |
173 | - return uniqid($this->getQueueName()) . Uuid::uuid4()->toString() . microtime(); |
|
173 | + return uniqid($this->getQueueName()).Uuid::uuid4()->toString().microtime(); |
|
174 | 174 | } |
175 | 175 | |
176 | 176 | /** |
@@ -217,17 +217,17 @@ discard block |
||
217 | 217 | try { |
218 | 218 | $connection->close(); |
219 | 219 | $failed = false; |
220 | - fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...' . PHP_EOL); |
|
220 | + fwrite($this->logOutput, 'start RpcClient::forceReconnect() - trying connect...'.PHP_EOL); |
|
221 | 221 | $connection = $this->getConnectionManager()->getConnection($this->connectionName); |
222 | 222 | $channel = $connection->channel(); |
223 | 223 | $this->createCallbackQueue($channel, $expire, $corralationId); |
224 | 224 | } catch (\Exception $e) { |
225 | 225 | $failed = true; |
226 | 226 | sleep(3); |
227 | - fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - ' . $e->getMessage() . PHP_EOL); |
|
227 | + fwrite($this->errOutput, 'failed RpcClient::forceReconnect() - '.$e->getMessage().PHP_EOL); |
|
228 | 228 | } |
229 | 229 | } while ($failed); |
230 | - fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!' . PHP_EOL); |
|
230 | + fwrite($this->logOutput, 'RpcClient::forceReconnect() - connected!'.PHP_EOL); |
|
231 | 231 | |
232 | 232 | return $connection; |
233 | 233 | } |