sportarchive / CloudProcessingEngine

pollers/src/InputPoller.php

Spacing +12 added lines, -12 removed lines

@@ -34,7 +34,7 @@ 
  * It opens the JSON input and starts a execute a callback corresponding to the command
  */
 
-require_once __DIR__ . "/../vendor/autoload.php";
+require_once __DIR__."/../vendor/autoload.php";
 
 use Aws\Swf\Exception;
 use SA\CpeSdk;
@@ -95,7 +95,7 @@ 
         foreach ($this->config->{'clients'} as $client)
         {
             $msg = null;
-            $this->cpeLogger->log_out("DEBUG", __DIR__, "Polling from client: " . print_r($client, true));
+            $this->cpeLogger->log_out("DEBUG", __DIR__, "Polling from client: ".print_r($client, true));
             
             // Long Polling messages from client input queue
             $queue = $client->{'queues'}->{'input'};
@@ -139,7 +139,7 @@ 
             $this->cpeLogger->log_out(
                 "ERROR", 
                 basename(__FILE__), 
-                "Command '" . $message->{"type"} . "' is unknown! Ignoring ..."
+                "Command '".$message->{"type"}."' is unknown! Ignoring ..."
             );
             return;
         }
@@ -147,13 +147,13 @@ 
         $this->cpeLogger->log_out(
             "INFO", 
             basename(__FILE__), 
-            "Received message '" . $message->{"type"}  . "'"
+            "Received message '".$message->{"type"}."'"
         );
         if ($this->debug)
             $this->cpeLogger->log_out(
                 "DEBUG", 
                 basename(__FILE__), 
-                "Details:\n" . json_encode($message, JSON_PRETTY_PRINT)
+                "Details:\n".json_encode($message, JSON_PRETTY_PRINT)
             );
 
         // We call the callback function that handles this message  
@@ -226,10 +226,10 @@ 
     private function validate_message($message)
     {
         if (!isset($message) || 
-            !isset($message->{"time"})   || $message->{"time"} == "" || 
-            !isset($message->{"jobId"})  || $message->{"jobId"} == "" || 
-            !isset($message->{"type"})   || $message->{"type"} == "" || 
-            !isset($message->{"data"})   || $message->{"data"} == "")
+            !isset($message->{"time"}) || $message->{"time"} == "" || 
+            !isset($message->{"jobId"}) || $message->{"jobId"} == "" || 
+            !isset($message->{"type"}) || $message->{"type"} == "" || 
+            !isset($message->{"data"}) || $message->{"data"} == "")
             throw new CpeSdk\CpeException("'time', 'type', 'jobId' or 'data' fields missing in JSON message file!",
                 self::INVALID_JSON);
         
@@ -250,7 +250,7 @@ 
 function usage($defaultConfigFile)
 {
     echo("# Description\nThe InputPoller connects your client applications to your stack. It handles communication over the two SQS channels you created for your client application.\n\n");
-    echo("Usage: php ". basename(__FILE__) . " [-h] [-d] -n <client_name> [-l <log path>]\n");
+    echo("Usage: php ".basename(__FILE__)." [-h] [-d] -n <client_name> [-l <log path>]\n");
     echo("-h: Print this help\n");
     echo("-d: Debug mode\n");
     echo("-l <log_path>: Location where logs will be dumped in (folder).\n");
@@ -287,9 +287,9 @@ 
     $cpeLogger = new CpeSdk\CpeLogger($logPath, $options['n'], $debug);
         
     $config = new \stdClass;
-    $config->clients = [(object)[
+    $config->clients = [(object) [
             'name' => $options['n'],
-            'queues' => (object)[
+            'queues' => (object) [
                 'input' => getenv('INPUT_QUEUE'),
                 'output' => getenv('OUTPUT_QUEUE'),
             ],

Braces +35 added lines, -25 removed lines

@@ -86,9 +86,10 @@ 
     // If a msg is received, we pass it to 'handle_input' for processing
     public function poll_SQS_queues()
     {
-        if (!isset($this->config->{'clients'}))
-            throw new CpeSdk\CpeException("Clients configuration invalid. Check the config file or your parameters.",
+        if (!isset($this->config->{'clients'})) {
+                    throw new CpeSdk\CpeException("Clients configuration invalid. Check the config file or your parameters.",
                 self::INVALID_CONFIG);
+        }
         
         // For all clients in config files
         // We poll from queues
@@ -102,13 +103,14 @@ 
             try {
                 if ($msg = $this->cpeSqsListener->receive_message($queue, 10))
                 {
-                    if (!($decoded = json_decode($msg['Body'])))
-                        $this->cpeLogger->log_out(
+                    if (!($decoded = json_decode($msg['Body']))) {
+                                            $this->cpeLogger->log_out(
                             "ERROR", 
                             basename(__FILE__), 
                             "JSON data invalid in queue: '$queue'");
-                    else                    
-                        $this->handle_message($decoded);
+                    } else {
+                                            $this->handle_message($decoded);
+                    }
                 }
             } catch (CpeSdk\CpeException $e) {
                 $this->cpeLogger->log_out(
@@ -123,8 +125,9 @@ 
             }
             
             // Message polled. Valid or not, we delete it from SQS
-            if ($msg)
-                $this->cpeSqsListener->delete_message($queue, $msg);
+            if ($msg) {
+                            $this->cpeSqsListener->delete_message($queue, $msg);
+            }
         }
     }
 
@@ -149,12 +152,13 @@ 
             basename(__FILE__), 
             "Received message '" . $message->{"type"}  . "'"
         );
-        if ($this->debug)
-            $this->cpeLogger->log_out(
+        if ($this->debug) {
+                    $this->cpeLogger->log_out(
                 "DEBUG", 
                 basename(__FILE__), 
                 "Details:\n" . json_encode($message, JSON_PRETTY_PRINT)
             );
+        }
 
         // We call the callback function that handles this message  
         $this->{$this->typesMap[$message->{"type"}]}($message);
@@ -168,12 +172,13 @@ 
     // Start a new workflow in SWF to initiate new transcoding job
     private function start_job($message)
     {
-        if ($this->debug)
-            $this->cpeLogger->log_out(
+        if ($this->debug) {
+                    $this->cpeLogger->log_out(
                 "DEBUG",
                 basename(__FILE__),
                 "Starting new workflow!"
             );
+        }
         
         // Workflow info
         $workflowType = array(
@@ -192,9 +197,10 @@ 
                 "taskList"     => array("name" => $message->{'data'}->{'workflow'}->{'taskList'}),
                 "input"        => json_encode($message->{'data'})
             );
-            if (isset($message->{'data'}->{'workflow'}->{'executionStartToCloseTimeout'})) 
-                $payload["executionStartToCloseTimeout"] =
+            if (isset($message->{'data'}->{'workflow'}->{'executionStartToCloseTimeout'})) {
+                            $payload["executionStartToCloseTimeout"] =
                     $message->{'data'}->{'workflow'}->{'executionStartToCloseTimeout'};
+            }
             $workflowRunId = $this->cpeSwfHandler->swf->startWorkflowExecution($payload);
 
             $this->cpeLogger->log_out(
@@ -229,13 +235,15 @@ 
             !isset($message->{"time"})   || $message->{"time"} == "" || 
             !isset($message->{"jobId"})  || $message->{"jobId"} == "" || 
             !isset($message->{"type"})   || $message->{"type"} == "" || 
-            !isset($message->{"data"})   || $message->{"data"} == "")
-            throw new CpeSdk\CpeException("'time', 'type', 'jobId' or 'data' fields missing in JSON message file!",
+            !isset($message->{"data"})   || $message->{"data"} == "") {
+                    throw new CpeSdk\CpeException("'time', 'type', 'jobId' or 'data' fields missing in JSON message file!",
                 self::INVALID_JSON);
+        }
         
-        if (!isset($message->{'data'}->{'workflow'}))
-            throw new CpeSdk\CpeException("Input doesn't contain any workflow information. You must provide the workflow you want to send this job to!",
+        if (!isset($message->{'data'}->{'workflow'})) {
+                    throw new CpeSdk\CpeException("Input doesn't contain any workflow information. You must provide the workflow you want to send this job to!",
                 self::INVALID_JSON);
+        }
     }
 }
 
@@ -266,11 +274,13 @@ 
     // Handle input parameters
     $options = getopt("l:hdn:");
 
-    if (isset($options['h']))
-        usage();
+    if (isset($options['h'])) {
+            usage();
+    }
     
-    if (isset($options['d'])) 
-        $debug = true;
+    if (isset($options['d'])) {
+            $debug = true;
+    }
         
     $logPath = null;
     if (isset($options['l']))
@@ -308,8 +318,7 @@ 
 // Create InputPoller object
 try {
     $inputPoller = new InputPoller($config);
-}
-catch (CpeSdk\CpeException $e) {
+} catch (CpeSdk\CpeException $e) {
     echo $e->getMessage();
     $cpeLogger->log_out(
         "FATAL", 
@@ -323,5 +332,6 @@ 
 print "Start polling ...\n";
 
 // Start polling loop to get incoming commands from SQS input queues
-while (42)
+while (42) {
     $inputPoller->poll_SQS_queues();
+}