kaliop-uk / kueueingbundle-kinesis

Adapter/Kinesis/Driver.php

Doc Comments +3 added lines

@@ -18,6 +18,9 @@
         return $consumer;
     }
 
+    /**
+     * @param string $queueName
+     */
     protected function getQueueCallback($queueName)
     {
         $callbacks = $this->container->getParameter('kaliop_queueing_kinesis.default.consumers');

Service/DoctrineCacheSequenceNumberStore.php

Spacing +1 added lines, -1 removed lines

@@ -14,7 +14,7 @@
 
     public function __construct(Cache $cacheService)
     {
-        $this->cache =  $cacheService;
+        $this->cache = $cacheService;
     }
 
     /**

Adapter/Kinesis/Message.php

Spacing +1 added lines, -1 removed lines

@@ -10,7 +10,7 @@
     protected $properties = array();
     protected $contentType;
 
-    public function __construct($body, array $properties = array(), $contentType='application/json')
+    public function __construct($body, array $properties = array(), $contentType = 'application/json')
     {
         $this->body = $body;
         $this->properties = $properties;

Adapter/Kinesis/QueueManager.php

Spacing +1 added lines, -1 removed lines

@@ -44,7 +44,7 @@
         return array('info', 'list', 'delete');
     }
 
-    public function executeAction($action, array $arguments=array())
+    public function executeAction($action, array $arguments = array())
     {
         switch ($action) {
             case 'delete':

Adapter/Kinesis/Producer.php

Spacing +2 added lines, -2 removed lines

@@ -70,7 +70,7 @@ 
     public function batchPublish(array $messages)
     {
         $records = aray();
-        foreach($messages as $message) {
+        foreach ($messages as $message) {
             $records[] = array(
                 'Data' => $message['msgBody'],
                 'PartitionKey' => $message['routingKey']
@@ -121,7 +121,7 @@ 
      */
     public function setContentType($contentType)
     {
-        if($contentType != 'application/json') {
+        if ($contentType != 'application/json') {
             throw new \Exception("Unsupported content-type for message serialization: $contentType. Only 'application/json' is supported");
         }
 

Adapter/Kinesis/Consumer.php

Spacing +4 added lines, -4 removed lines

@@ -53,7 +53,7 @@ 
      */
     public function setCallback($callback)
     {
-        if (! $callback instanceof \Kaliop\QueueingBundle\Queue\MessageConsumerInterface) {
+        if (!$callback instanceof \Kaliop\QueueingBundle\Queue\MessageConsumerInterface) {
             throw new \RuntimeException('Can not set callback to SQS Consumer, as it is not a MessageConsumerInterface');
         }
         $this->callback = $callback;
@@ -107,7 +107,7 @@ 
      * @param int $timeout
      * @return nothing
      */
-    public function consume($amount, $timeout=0)
+    public function consume($amount, $timeout = 0)
     {
         $iterator = $this->getInitialMessageIterator();
 
@@ -117,7 +117,7 @@ 
             $remaining = $timeout;
         }
 
-        while(true) {
+        while (true) {
             $reqTime = microtime(true);
             $result = $this->client->getRecords(array(
                 'ShardIterator' => $iterator,
@@ -131,7 +131,7 @@ 
                 $this->sequenceNumberStore->save($this->streamName, $this->shardId, $last['SequenceNumber']);
             }
 
-            foreach($records as $record) {
+            foreach ($records as $record) {
                 $data = $record['Data'];
                 unset($record['Data']);
                 $this->callback->receive(new Message($data, $record));