@@ -274,7 +274,7 @@ discard block |
||
274 | 274 | } |
275 | 275 | |
276 | 276 | /** |
277 | - * @return string |
|
277 | + * @return boolean |
|
278 | 278 | */ |
279 | 279 | public function getExchange() |
280 | 280 | { |
@@ -282,7 +282,7 @@ discard block |
||
282 | 282 | } |
283 | 283 | |
284 | 284 | /** |
285 | - * @return string |
|
285 | + * @return boolean |
|
286 | 286 | */ |
287 | 287 | public function getType() |
288 | 288 | { |
@@ -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 | $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']); |
@@ -62,10 +62,10 @@ |
||
62 | 62 | $this->parameters['user'], |
63 | 63 | $this->parameters['password'], |
64 | 64 | $this->parameters['vhost'], |
65 | - false, // insist |
|
65 | + false, // insist |
|
66 | 66 | 'AMQPLAIN', // login_method |
67 | - null, // login_response |
|
68 | - 'en_US', // locale |
|
67 | + null, // login_response |
|
68 | + 'en_US', // locale |
|
69 | 69 | $this->parameters['connection_timeout'], |
70 | 70 | $this->parameters['read_write_timeout'], |
71 | 71 | $this->parameters['ssl_context'], |
@@ -29,11 +29,11 @@ |
||
29 | 29 | */ |
30 | 30 | public function toClosure() |
31 | 31 | { |
32 | - return function (AMQPMessage $message) { |
|
32 | + return function(AMQPMessage $message) { |
|
33 | 33 | $response = $this->getQueueService()->handle($message); |
34 | 34 | |
35 | 35 | $amqpResponse = new CmobiAMQPMessage( |
36 | - (string) $response, |
|
36 | + (string)$response, |
|
37 | 37 | ['correlation_id' => $message->get('correlation_id')] |
38 | 38 | ); |
39 | 39 |
@@ -76,7 +76,7 @@ |
||
76 | 76 | ->disableOriginalConstructor() |
77 | 77 | ->getMock(); |
78 | 78 | $callbackMock->method('toClosure') |
79 | - ->willReturn(function () {}); |
|
79 | + ->willReturn(function() {}); |
|
80 | 80 | |
81 | 81 | return $callbackMock; |
82 | 82 | } |
@@ -3,7 +3,6 @@ |
||
3 | 3 | namespace Cmobi\RabbitmqBundle\Transport\Worker; |
4 | 4 | |
5 | 5 | use Cmobi\RabbitmqBundle\Connection\ConnectionManager; |
6 | -use Cmobi\RabbitmqBundle\Connection\Exception\InvalidAMQPChannelException; |
|
7 | 6 | use Cmobi\RabbitmqBundle\Queue\Queue; |
8 | 7 | use Cmobi\RabbitmqBundle\Queue\QueueBagInterface; |
9 | 8 | use Cmobi\RabbitmqBundle\Queue\QueueBuilderInterface; |
@@ -34,7 +34,7 @@ |
||
34 | 34 | */ |
35 | 35 | public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag) |
36 | 36 | { |
37 | - if (! $queueBag instanceof WorkerQueueBag) { |
|
37 | + if (!$queueBag instanceof WorkerQueueBag) { |
|
38 | 38 | throw new \Exception('Unsupported QueueBag'); |
39 | 39 | } |
40 | 40 | $queue = new Queue($this->getConnectionManager(), $queueBag, $this->logger, $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 |
@@ -41,7 +41,7 @@ |
||
41 | 41 | $this->refreshChannel(); |
42 | 42 | $queueBag = new SubscriberQueueBag($this->getExchange(), $this->getExchangeType(), $this->getQueueName()); |
43 | 43 | $this->getChannel()->exchangeDeclare($queueBag->getExchangeDeclare()); |
44 | - $msg = new CmobiAMQPMessage((string) $data); |
|
44 | + $msg = new CmobiAMQPMessage((string)$data); |
|
45 | 45 | $this->getChannel()->basic_publish($msg, $queueBag->getExchange()); |
46 | 46 | |
47 | 47 | $this->getChannel()->close(); |
@@ -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, |
@@ -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 |
@@ -33,7 +33,7 @@ |
||
33 | 33 | */ |
34 | 34 | public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag) |
35 | 35 | { |
36 | - if (! $queueBag instanceof RpcQueueBag) { |
|
36 | + if (!$queueBag instanceof RpcQueueBag) { |
|
37 | 37 | throw new \Exception('Unsupported QueueBag'); |
38 | 38 | } |
39 | 39 | $queue = new Queue($this->getConnectionManager(), $queueBag, $this->logger, $this->connectionName); |
@@ -10,6 +10,9 @@ discard block |
||
10 | 10 | private $resolver; |
11 | 11 | private $options; |
12 | 12 | |
13 | + /** |
|
14 | + * @param string $queueName |
|
15 | + */ |
|
13 | 16 | public function __construct( |
14 | 17 | $queueName, |
15 | 18 | $basicQos = 1, |
@@ -187,7 +190,7 @@ discard block |
||
187 | 190 | } |
188 | 191 | |
189 | 192 | /** |
190 | - * @return string |
|
193 | + * @return boolean |
|
191 | 194 | */ |
192 | 195 | public function getExchange() |
193 | 196 | { |
@@ -195,7 +198,7 @@ discard block |
||
195 | 198 | } |
196 | 199 | |
197 | 200 | /** |
198 | - * @return string |
|
201 | + * @return boolean |
|
199 | 202 | */ |
200 | 203 | public function getType() |
201 | 204 | { |