sportarchive / CloudProcessingEngine

pollers/src/ActivityPoller.php

Spacing +6 added lines, -6 removed lines

@@ -100,7 +100,7 @@ 
         // Poll from all the taskList registered for each activities 
         if ($this->debug)
             $this->cpeLogger->log_out("DEBUG", basename(__FILE__),
-                "Polling activity taskList '" . $this->taskList  . "' ... ");
+                "Polling activity taskList '" . $this->taskList . "' ... ");
             
         try {
             // Call SWF and poll for incoming tasks
@@ -201,9 +201,9 @@ 
                 if (!file_exists($activityToHandle->{"file"}))
                 {
                     $this->cpeLogger->log_out("ERROR", basename(__FILE__),
-                        "The code file '".$activityToHandle->{"file"}."' for activity: name=" 
+                        "The code file '" . $activityToHandle->{"file"} . "' for activity: name=" 
                         . $activityToHandle->{"name"} . ",version=" 
-                        . $activityToHandle->{"version"}." doesn't exists! Check if the file is accessible and if the path is correct in your config file.");
+                        . $activityToHandle->{"version"} . " doesn't exists! Check if the file is accessible and if the path is correct in your config file.");
                     return false;
                 }
                 
@@ -242,7 +242,7 @@ 
         $this->cpeLogger->log_out("ERROR", basename(__FILE__),
             "No Activity handler was found for: name=" 
             . $this->activityName . ",version=" 
-            . $this->activityVersion.". Check your config file and ensure your 'activity' name AND 'version' is there.");    
+            . $this->activityVersion . ". Check your config file and ensure your 'activity' name AND 'version' is there.");    
         return false;
     }
 }
@@ -262,7 +262,7 @@ 
 // Usage
 function usage($defaultConfigFile)
 {
-    echo("Usage: php ". basename(__FILE__) . " -D <domain> -A <activity_name> -V <activity_version> [-T <task_list>] [-h] [-d] [-c <config_file path>] [-l <log path>]\n");
+    echo("Usage: php " . basename(__FILE__) . " -D <domain> -A <activity_name> -V <activity_version> [-T <task_list>] [-h] [-d] [-c <config_file path>] [-l <log path>]\n");
     echo("-h: Print this help\n");
     echo("-d: Debug mode\n");
     echo("-c <config_file path>: Optional parameter to override the default configuration file: '$defaultConfigFile'.\n");
@@ -331,7 +331,7 @@ 
     // Tasklist
     if (!isset($options['T']))
     {
-        $taskList = $options['A'].'-'.$options['V'];
+        $taskList = $options['A'] . '-' . $options['V'];
     } else {
         $taskList = $options['T'];
     }

Braces +28 added lines, -21 removed lines

@@ -98,9 +98,10 @@ 
     public function poll_for_activities()
     {
         // Poll from all the taskList registered for each activities 
-        if ($this->debug)
-            $this->cpeLogger->log_out("DEBUG", basename(__FILE__),
+        if ($this->debug) {
+                    $this->cpeLogger->log_out("DEBUG", basename(__FILE__),
                 "Polling activity taskList '" . $this->taskList  . "' ... ");
+        }
             
         try {
             // Call SWF and poll for incoming tasks
@@ -124,8 +125,9 @@ 
     {
         // Get activityType and WorkflowExecution info
         if (!($activityType      = $activityTask->get("activityType")) ||
-            !($workflowExecution = $activityTask->get("workflowExecution")))
-            return false;
+            !($workflowExecution = $activityTask->get("workflowExecution"))) {
+                    return false;
+        }
         
         $this->cpeLogger->log_out("INFO",
             basename(__FILE__),
@@ -152,16 +154,18 @@ 
             // Perform input validation
             $this->activityHandler->do_input_validation();
 
-            if ($this->debug)
-                $this->cpeLogger->log_out("DEBUG", basename(__FILE__), 
+            if ($this->debug) {
+                            $this->cpeLogger->log_out("DEBUG", basename(__FILE__), 
                     "Activity input:\n" . print_r($this->activityHandler->input, true));
+            }
         
             // Run activity task
             $result = $this->activityHandler->do_activity($activityTask);
 
-            if ($this->debug && $result)
-                $this->cpeLogger->log_out("DEBUG", basename(__FILE__), 
+            if ($this->debug && $result) {
+                            $this->cpeLogger->log_out("DEBUG", basename(__FILE__), 
                     "Activity output:\n" . print_r($result, true));
+            }
         
         } catch (CpeSdk\CpeException $e) {
             $reason  = $e->ref;
@@ -216,8 +220,9 @@ 
                 // Instantiate the Activity class that will process Tasks
 
                 if (!isset($this->cpeLogger) ||
-                    !$this->cpeLogger)
-                    print "EMPTY !!!\n";
+                    !$this->cpeLogger) {
+                                    print "EMPTY !!!\n";
+                }
                 
                 $this->activityHandler = 
                     new $activityToHandle->{"class"}(
@@ -287,15 +292,18 @@ 
     global $activityVersion;
     
     // Handle input parameters
-    if (!($options = getopt("D:T:A:V:c:l:hd")))
-        usage($defaultConfigFile);
+    if (!($options = getopt("D:T:A:V:c:l:hd"))) {
+            usage($defaultConfigFile);
+    }
     
-    if (!count($options) || isset($options['h']))
-        usage($defaultConfigFile);
+    if (!count($options) || isset($options['h'])) {
+            usage($defaultConfigFile);
+    }
 
     // Debug
-    if (isset($options['d']))
-        $debug = true;
+    if (isset($options['d'])) {
+            $debug = true;
+    }
 
     // Domain
     if (!isset($options['D']))
@@ -353,8 +361,7 @@ 
             print "\n[CONFIG ISSUE]\nConfiguration file '$defaultConfigFile' invalid or non-existant.\n\n[EASY FIX]\nGo to the directory mentioned in the error above and rename the template file 'cpeConfigTemplate.json' to 'cpeConfig.json'. Configure your Activities. As example you have Activities for CloudTranscode already setup in the template. You can declare your Activities and start executing tasks in an SWF workflow.\n";
             exit(1);
         }
-    }
-    catch (Exception $e) {
+    } catch (Exception $e) {
         print $e;
     }
             
@@ -374,8 +381,7 @@ 
 // Instantiate ActivityPoller
 try {
     $activityPoller = new ActivityPoller($config);
-} 
-catch (CpeSdk\CpeException $e) {
+} catch (CpeSdk\CpeException $e) {
     $cpeLogger->log_out("FATAL",
         basename(__FILE__), $e->getMessage());
     exit(1);
@@ -384,6 +390,7 @@ 
 $cpeLogger->log_out("INFO", basename(__FILE__), "Starting activity tasks polling");
 
 // Start polling loop
-while (42)
+while (42) {
     $activityPoller->poll_for_activities();
+}
 

pollers/src/InputPoller.php

Spacing +11 added lines, -11 removed lines

@@ -114,12 +114,12 @@ 
                 $this->cpeLogger->log_out(
                     "ERROR", 
                     basename(__FILE__), 
-                    $e->getMessage().print_r($msg, true));
+                    $e->getMessage() . print_r($msg, true));
             } catch (\Exception $e) {
                 $this->cpeLogger->log_out(
                     "ERROR", 
                     basename(__FILE__), 
-                    $e->getMessage().print_r($msg, true));
+                    $e->getMessage() . print_r($msg, true));
             }
             
             // Message polled. Valid or not, we delete it from SQS
@@ -147,7 +147,7 @@ 
         $this->cpeLogger->log_out(
             "INFO", 
             basename(__FILE__), 
-            "Received message '" . $message->{"type"}  . "'"
+            "Received message '" . $message->{"type"} . "'"
         );
         if ($this->debug)
             $this->cpeLogger->log_out(
@@ -196,7 +196,7 @@ 
             $this->cpeLogger->log_out(
                 "INFO",
                 basename(__FILE__),
-                "New workflow submitted to SWF: ".$workflowRunId->get('runId'));
+                "New workflow submitted to SWF: " . $workflowRunId->get('runId'));
 
             // Send WORKFLOW_SCHEDULED message back to client
             $this->cpeSqsWriter->workflow_scheduled($workflowType, $workflowRunId->get('runId'), $workflowId, $message);
@@ -222,10 +222,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);
         
@@ -246,7 +246,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");
@@ -283,9 +283,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 +32 added lines, -23 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(
@@ -225,13 +230,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);
+        }
     }
 }
 
@@ -262,11 +269,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']))
@@ -304,8 +313,7 @@ 
 // Create InputPoller object
 try {
     $inputPoller = new InputPoller($config);
-}
-catch (CpeSdk\CpeException $e) {
+} catch (CpeSdk\CpeException $e) {
     echo $e->getMessage();
     $cpeLogger->log_out(
         "FATAL", 
@@ -319,5 +327,6 @@ 
 print "Start polling ...\n";
 
 // Start polling loop to get incoming commands from SQS input queues
-while (42)
+while (42) {
     $inputPoller->poll_SQS_queues();
+}