[ 'type' => 'CC', // Handler Category (Possible values: HC, CC, HF) 'handler' => 'App\Controllers\PolicyController', ], ]; public function __construct() { } public static function streamOutput($data) { ob_implicit_flush(true); // try { ob_end_flush(); } catch(Exception $e) { echo $e->getMessage(); } echo $data; flush(); } public static function processJobs(array $jobdata = []) { $db = \Config\Database::connect(); $runningJob = $db->query( "SELECT id FROM partner_jobs WHERE status=? LIMIT 1", [self::STATUS_RUNNING] )->getRow(); if ($runningJob !== null) { SELF::streamOutput("Another job is already running (id: {$runningJob->id}). Skipping.\n"); return false; } $query = " SELECT id, name, payload, uuid FROM partner_jobs WHERE status=? ORDER BY created_dt ASC LIMIT 1"; $jobs = $db->query($query, [self::STATUS_QUEUED])->getResult(); if (count($jobs)) { $job = $jobs[0]; return SELF::processJob(['id' => $job->id, 'uuid' => $job->uuid]); } return false; } /** * process jobs */ public static function processJob(array $jobdata = []) { // print_r($jobdata); // echo 'listen'; // die(); $db = \Config\Database::connect(); $db->transStart(); // Do not start a new job while another job is already running. $runningJob = $db->query( "SELECT id FROM partner_jobs WHERE status=? LIMIT 1", [self::STATUS_RUNNING] )->getRow(); if ($runningJob !== null) { $db->transComplete(); SELF::streamOutput("Another job is already running (id: {$runningJob->id}). Skipping.\n"); return false; } if (isset($jobdata) && count($jobdata)) { $query = " SELECT id, name, payload, uuid FROM partner_jobs WHERE status=? AND id=? AND uuid=? ORDER BY created_dt ASC LIMIT 1 FOR UPDATE"; $where_condition = [self::STATUS_QUEUED, $jobdata['id'], $jobdata['uuid']]; } else { $query = " SELECT id, name, payload, uuid FROM partner_jobs WHERE status=? ORDER BY created_dt ASC LIMIT 1 FOR UPDATE"; $where_condition = [self::STATUS_QUEUED]; } $job = $db->query($query, $where_condition)->getResult(); $db->transComplete(); // print_r($job);die(); if ($job !== []) { $job = $job[0]; // echo "\nProcessing job id - " . $job->id . "\n"; SELF::streamOutput("\nProcessing job id - " . $job->id . "\n"); // sleep(5); // echo "Job name - " . $job->name . "\n"; SELF::streamOutput("Job name - " . $job->name . "\n"); // sleep(5); // print_r(SELF::$event_class_mapping); // echo array_key_exists($job->name,SELF::$event_class_mapping) ? 'mapped' : 'notmapped'; // die(); try { $start = microtime(true); $runtime = null; $job_status = self::STATUS_RUNNING; $db->query("UPDATE partner_jobs SET status=? WHERE id=? AND uuid=?", [$job_status, $job->id,$job->uuid]); if (!array_key_exists($job->name,SELF::$event_class_mapping)) { throw new \RuntimeException('Job ' . $job->name . ' handler not registered'); } $handler = SELF::$event_class_mapping[$job->name]; $handleInstance = null; if($handler['type'] == 'CC' || $handler['type'] == 'HC') { // echo 'CLASS - ' . $handler['type'].' - ' . $handler['handler']; $handlerClass = $handler['handler']; if(class_exists($handlerClass)) { $handleInstance = new $handlerClass(); } else { throw new \RuntimeException('Job ' . $job->name . ' or handler class not found'); // echo $e->getMessage(); } if (method_exists($handleInstance, $job->name)) { $jobHandler = [$handleInstance, $job->name]; //throw new \RuntimeException('Job ' . $job->name . ' not found'); } else if(method_exists($handleInstance, 'handle')) { $jobHandler = [$handleInstance, 'handle']; } else { throw new \RuntimeException('Job ' . $job->name . ' or handler not found'); } } else if($handler['type'] == 'HF') { // echo 'HF - ' . $handler['type']; $jobHandler = $handleInstance = $handler['handler']; // echo $jobHandler; } else{ throw new \RuntimeException('Job ' . $job->name . ' Invalid job type'); } $payload = json_decode($job->payload, true); if (!is_array($payload)) { throw new \InvalidArgumentException('Invalid payload format here'); } $payload = is_array($payload) ? $payload : []; try { $response = $jobHandler($payload,$job->id); $job_status = self::STATUS_DONE; } catch(\Exception $e) { $job_status = self::STATUS_FAILED; $runtime = $runtime === null ? microtime(true) - $start : $runtime; $response = ['file_name' => $e->getFile(),'error' => $e->getMessage(),'line_no' => $e->getLine(),'info' => $e->getTraceAsString(),'scope' => 'task failed']; } $runtime = microtime(true) - $start; } catch (\Exception $e) { //die(); $job_status = self::STATUS_FAILED; $runtime = $runtime === null ? microtime(true) - $start : $runtime; $response = ['file_name' => $e->getFile(),'error' => $e->getMessage(),'line_no' => $e->getLine(),'info' => $e->getTraceAsString(),'scope' => 'worker failed']; } $db->query("UPDATE partner_jobs SET status=?, run_time=?, response=? WHERE id=? AND uuid=?", [ $job_status, $runtime, json_encode($response), $job->id,$job->uuid ]); //update file status if this job is directly linked with a file id if($job_status == self::STATUS_FAILED) { //get file from job payload // if(array_key_exists('file_id', $payload) && $payload['file_id'] != NULL && $payload['file_id'] != "" && is_numeric($payload['file_id']) && count($payload) == 1) // { // $fileModel = new FileModel(); // $fileModel->where('id', $payload['file_id']) // ->set(['status' => 'failed','reason' => json_encode(['error_type' => 0,'error_summary' => [0],'error_data' => 'System Error, please contact Admin/Support team']) ]) // ->update(); // } } // echo "Job $job->id $job_status. Response - $response \n"; // echo "Job $job->id $job_status \n"; SELF::streamOutput("Job $job->id $job_status \n"); return true; } else { echo 'no job found in queue'; } } }