@@ -48,7 +48,6 @@ |
||
| 48 | 48 | } |
| 49 | 49 | |
| 50 | 50 | /** |
| 51 | - * @param string $queueName NB: complete queue name as used by SQS |
|
| 52 | 51 | * @return Producer |
| 53 | 52 | * @todo test that we can successfully send messages to 2 queues using the same SqsClient |
| 54 | 53 | */ |
@@ -98,7 +98,7 @@ |
||
| 98 | 98 | for ($i = 0; $i < count($messages); $i += 10) { |
| 99 | 99 | $entries = array(); |
| 100 | 100 | $toSend = array_slice($messages, $i, 10); |
| 101 | - foreach($toSend as $message) { |
|
| 101 | + foreach ($toSend as $message) { |
|
| 102 | 102 | $entries[] = array_merge( |
| 103 | 103 | array( |
| 104 | 104 | 'MessageBody' => $message['msgBody'], |
@@ -43,7 +43,7 @@ discard block |
||
| 43 | 43 | return array('list-available', 'list-configured', 'create', 'info', 'purge', 'delete'); |
| 44 | 44 | } |
| 45 | 45 | |
| 46 | - public function executeAction($action, array $arguments=array()) |
|
| 46 | + public function executeAction($action, array $arguments = array()) |
|
| 47 | 47 | { |
| 48 | 48 | switch ($action) { |
| 49 | 49 | case 'list-available': |
@@ -145,11 +145,11 @@ discard block |
||
| 145 | 145 | |
| 146 | 146 | protected function getProducerService() |
| 147 | 147 | { |
| 148 | - return $this->container->get('kaliop_queueing.sqs.'. $this->queueName . '_producer'); |
|
| 148 | + return $this->container->get('kaliop_queueing.sqs.'.$this->queueName.'_producer'); |
|
| 149 | 149 | } |
| 150 | 150 | |
| 151 | 151 | public function registerQueue($queueName) |
| 152 | 152 | { |
| 153 | - $this->registeredQueues[]=$queueName; |
|
| 153 | + $this->registeredQueues[] = $queueName; |
|
| 154 | 154 | } |
| 155 | 155 | } |
@@ -122,7 +122,7 @@ |
||
| 122 | 122 | * @param string $routingKey |
| 123 | 123 | * @return Consumer |
| 124 | 124 | */ |
| 125 | - public function createConsumer($queueName, $queueUrl, $connectionId, $callback=null, $routingKey=null) |
|
| 125 | + public function createConsumer($queueName, $queueUrl, $connectionId, $callback = null, $routingKey = null) |
|
| 126 | 126 | { |
| 127 | 127 | $class = $this->container->getParameter('kaliop_queueing.sqs.consumer.class'); |
| 128 | 128 | $consumer = new $class($this->getConnectionConfig($connectionId)); |
@@ -103,7 +103,7 @@ discard block |
||
| 103 | 103 | */ |
| 104 | 104 | public function setCallback($callback) |
| 105 | 105 | { |
| 106 | - if (! $callback instanceof \Kaliop\QueueingBundle\Queue\MessageConsumerInterface) { |
|
| 106 | + if (!$callback instanceof \Kaliop\QueueingBundle\Queue\MessageConsumerInterface) { |
|
| 107 | 107 | throw new \RuntimeException('Can not set callback to SQS Consumer, as it is not a MessageConsumerInterface'); |
| 108 | 108 | } |
| 109 | 109 | $this->callback = $callback; |
@@ -174,7 +174,7 @@ discard block |
||
| 174 | 174 | 'MessageAttributeNames' => array('All') |
| 175 | 175 | ); |
| 176 | 176 | |
| 177 | - while(true) { |
|
| 177 | + while (true) { |
|
| 178 | 178 | $reqTime = microtime(true); |
| 179 | 179 | |
| 180 | 180 | if ($timeout > 0) { |
@@ -211,13 +211,13 @@ discard block |
||
| 211 | 211 | $messages = $result->get('Messages'); |
| 212 | 212 | |
| 213 | 213 | if (is_array($messages)) { |
| 214 | - foreach($messages as $message) { |
|
| 214 | + foreach ($messages as $message) { |
|
| 215 | 215 | |
| 216 | 216 | // How we implement routing keys with SQS: since it is not supported natively, we check if the route |
| 217 | 217 | // matches after having downloaded the message. If it does not match, we just skip processing it. |
| 218 | 218 | // Since we will not call deleteMessage, SQS will requeue the message in a short time. |
| 219 | 219 | // This is far from optimal, but it might be better than nothing |
| 220 | - if (! $this->matchRoutingKey($message)) { |
|
| 220 | + if (!$this->matchRoutingKey($message)) { |
|
| 221 | 221 | continue; |
| 222 | 222 | } |
| 223 | 223 | |
@@ -232,7 +232,7 @@ discard block |
||
| 232 | 232 | $data = $message['Body']; |
| 233 | 233 | unset($message['Body']); |
| 234 | 234 | |
| 235 | - $contentType = isset( $message['MessageAttributes'][$this->contentTypeAttribute]['StringValue'] ) ? |
|
| 235 | + $contentType = isset($message['MessageAttributes'][$this->contentTypeAttribute]['StringValue']) ? |
|
| 236 | 236 | $message['MessageAttributes'][$this->contentTypeAttribute]['StringValue'] : ''; |
| 237 | 237 | |
| 238 | 238 | if ($contentType != '') { |
@@ -344,7 +344,7 @@ discard block |
||
| 344 | 344 | gc_collect_cycles(); |
| 345 | 345 | } |
| 346 | 346 | |
| 347 | - if ($this->memoryLimit > 0 && !$this->forceStop && memory_get_usage(true) >= ($this->memoryLimit * 1024 * 1024) ) { |
|
| 347 | + if ($this->memoryLimit > 0 && !$this->forceStop && memory_get_usage(true) >= ($this->memoryLimit * 1024 * 1024)) { |
|
| 348 | 348 | $this->forceStop("Memory limit of {$this->memoryLimit} MB reached while consuming messages"); |
| 349 | 349 | } |
| 350 | 350 | |