tasks, $Folder->find('Queue.+\.php')); foreach ($res as &$r) { $r = basename($r, 'Task.php'); } $this->tasks = $res; } $plugins = Plugin::loaded(); foreach ($plugins as $plugin) { $pluginPaths = App::path('Shell/Task', $plugin); foreach ($pluginPaths as $pluginPath) { $Folder = new Folder($pluginPath); $res = $Folder->find('Queue.+Task\.php'); foreach ($res as &$r) { $r = $plugin . '.' . basename($r, 'Task.php'); } $this->tasks = array_merge($this->tasks, $res); } } parent::initialize(); $this->QueuedJobs->initConfig(); } /** * @return string */ public function _getDescription() { $tasks = []; foreach ($this->taskNames as $loadedTask) { $tasks[] = "\t" . '* ' . $this->_taskName($loadedTask); } $tasks = implode(PHP_EOL, $tasks); $text = <<args) < 1) { $this->out('Please call like this:'); $this->out(' bin/cake queue add '); $this->_displayAvailableTasks(); return; } $name = Inflector::camelize($this->args[0]); if (in_array($name, $this->taskNames)) { $this->{$name}->add(); } elseif (in_array('Queue' . $name . '', $this->taskNames)) { $this->{'Queue' . $name}->add(); } else { $this->out('Error: Task not found: ' . $name); $this->_displayAvailableTasks(); } } /** * Output the task without Queue or Task * example: QueueImageTask becomes Image on display * * @param string $task Task name * @return string Cleaned task name */ protected function _taskName($task) { if (strpos($task, 'Queue') === 0) { return substr($task, 5); } return $task; } /** * Run a QueueWorker loop. * Runs a Queue Worker process which will try to find unassigned jobs in the queue * which it may run and try to fetch and execute them. * * @return void */ public function runworker() { $pidFilePath = Configure::read('Queue.pidfilepath'); if ($pidFilePath) { if (!file_exists($pidFilePath)) { mkdir($pidFilePath, 0755, true); } if (function_exists('posix_getpid')) { $pid = posix_getpid(); } else { $pid = $this->QueuedJobs->key(); } # global file $fp = fopen($pidFilePath . 'queue.pid', 'w'); fwrite($fp, $pid); fclose($fp); # specific pid file if (function_exists('posix_getpid')) { $pid = posix_getpid(); } else { $pid = $this->QueuedJobs->key(); } $pidFileName = 'queue_' . $pid . '.pid'; $fp = fopen($pidFilePath . $pidFileName, 'w'); fwrite($fp, $pid); fclose($fp); } // Enable Garbage Collector (PHP >= 5.3) if (function_exists('gc_enable')) { gc_enable(); } if (function_exists('pcntl_signal')) { pcntl_signal(SIGTERM, [&$this, '_exit']); } $this->_exit = false; $starttime = time(); $group = null; if (!empty($this->params['group'])) { $group = $this->params['group']; } while (!$this->_exit) { // make sure accidental overriding isnt possible set_time_limit(0); if (!empty($pidFilePath)) { touch($pidFilePath . 'queue.pid'); } if (!empty($pidFileName)) { touch($pidFilePath . $pidFileName); } $this->_log('runworker', isset($pid) ? $pid : null); $this->out('[' . date('Y-m-d H:i:s') . '] Looking for Job ...'); $queuedTask = $this->QueuedJobs->requestJob($this->_getTaskConf(), $group); if ($queuedTask) { $this->out('Running Job of type "' . $queuedTask['job_type'] . '"'); $taskname = 'Queue' . $queuedTask['job_type']; try { $data = json_decode($queuedTask['data'], true); /* @var \Queue\Shell\Task\QueueTask $task */ $task = $this->{$taskname}; $return = $task->run((array)$data, $queuedTask['id']); $failureMessage = null; if ($task->failureMessage) { $failureMessage = $task->failureMessage; } } catch (Exception $e) { $return = false; $failureMessage = get_class($e) . ': ' . $e->getMessage(); //log the exception $this->_logError($taskname . "\n" . $failureMessage . "\n" . $e->getTraceAsString()); } if ($return) { $this->QueuedJobs->markJobDone($queuedTask); $this->out('Job Finished.'); } else { $this->QueuedJobs->markJobFailed($queuedTask, $failureMessage); $this->out('Job did not finish, requeued.'); } } elseif (Configure::read('Queue.exitwhennothingtodo')) { $this->out('nothing to do, exiting.'); $this->_exit = true; } else { $this->out('nothing to do, sleeping.'); sleep(Configure::read('Queue.sleeptime')); } // check if we are over the maximum runtime and end processing if so. if (Configure::read('Queue.workermaxruntime') && (time() - $starttime) >= Configure::read('Queue.workermaxruntime')) { $this->_exit = true; $this->out('Reached runtime of ' . (time() - $starttime) . ' Seconds (Max ' . Configure::read('Queue.workermaxruntime') . '), terminating.'); } if ($this->_exit || rand(0, 100) > (100 - Configure::read('Queue.gcprob'))) { $this->out('Performing Old job cleanup.'); $this->QueuedJobs->cleanOldJobs(); } $this->hr(); } if (!empty($pid)) { if (file_exists($pidFilePath . 'queue_' . $pid . '.pid')) { unlink($pidFilePath . 'queue_' . $pid . '.pid'); } } } /** * @param string $message * @return void */ protected function _logError($message) { if (Configure::read('Queue.log')) { Log::write('error', $message); } } /** * Manually trigger a Finished job cleanup. * * @return void */ public function clean() { $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - Configure::read('Queue.cleanuptimeout'))); $this->QueuedJobs->cleanOldJobs(); } /** * @return void */ public function kill() { $processes = $this->QueuedJobs->getProcesses(); if (!$processes) { $this->out('No processed found'); return; } $this->out(count($processes) . ' processes:'); foreach ($processes as $process => $timestamp) { $this->out(' - ' . $process . ' (last run @ ' . (new Time($timestamp)) . ')'); } $options = array_keys($processes); $options[] = 'all'; $in = $this->in('Process', $options); if ($in === 'all') { foreach ($processes as $process => $timestamp) { $this->QueuedJobs->terminateProcess((int)$process); } return; } $this->QueuedJobs->terminateProcess((int)$in); } /** * Manually reset (failed) jobs for re-run. * Careful, this should not be done while a queue task is being run. * * @return void */ public function reset() { $this->out('Resetting...'); $this->QueuedJobs->reset(); } /** * Display current settings * * @return void */ public function settings() { $this->out('Current Settings:'); $conf = (array)Configure::read('Queue'); foreach ($conf as $key => $val) { if ($val === false) { $val = 'no'; } if ($val === true) { $val = 'yes'; } $this->out('* ' . $key . ': ' . print_r($val, true)); } } /** * Display some statistics about Finished Jobs. * * @return void */ public function stats() { $this->out('Jobs currenty in the Queue:'); $types = $this->QueuedJobs->getTypes()->toArray(); foreach ($types as $type) { $this->out(' ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedJobs->getLength($type)); } $this->hr(); $this->out('Total unfinished Jobs : ' . $this->QueuedJobs->getLength()); $this->hr(); $this->out('Finished Job Statistics:'); $data = $this->QueuedJobs->getStats(); foreach ($data as $item) { $this->out(' ' . $item['job_type'] . ': '); $this->out(' Finished Jobs in Database: ' . $item['num']); $this->out(' Average Job existence : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's'); $this->out(' Average Execution delay : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's'); $this->out(' Average Execution time : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's'); } } /** * Truncates the queue table * * @return void */ public function hardReset() { $this->QueuedJobs->truncate(); $message = __d('queue', 'OK'); $this->out($message); } /** * Set up tables * * @see README * @return void */ public function install() { $this->out('Run `cake Migrations.migrate -p Queue`'); $this->out('Set up cronjob, e.g. via `crontab -e -u www-data`'); } /** * Remove table and kill workers * * @return void */ public function uninstall() { $this->out('Remove all workers and cronjobs and then delete the Queue plugin tables.'); } /** * Get option parser method to parse commandline options * * @return \Cake\Console\ConsoleOptionParser */ public function getOptionParser() { $subcommandParser = [ 'options' => [ /* 'dry-run'=> array( 'short' => 'd', 'help' => 'Dry run the update, no jobs will actually be added.', 'boolean' => true ), 'log'=> array( 'short' => 'l', 'help' => 'Log all ouput to file log.txt in TMP dir'), 'boolean' => true ), */ ], ]; $subcommandParserFull = $subcommandParser; $subcommandParserFull['options']['group'] = [ 'short' => 'g', 'help' => 'Group', 'default' => '', ]; return parent::getOptionParser() ->description($this->_getDescription()) ->addSubcommand('clean', [ 'help' => 'Remove old jobs (cleanup)', 'parser' => $subcommandParser, ]) ->addSubcommand('add', [ 'help' => 'Add Job', 'parser' => $subcommandParser, ]) ->addSubcommand('install', [ 'help' => 'Install info', 'parser' => $subcommandParser, ]) ->addSubcommand('uninstall', [ 'help' => 'Uninstall info', 'parser' => $subcommandParser, ]) ->addSubcommand('stats', [ 'help' => 'Stats', 'parser' => $subcommandParserFull, ]) ->addSubcommand('settings', [ 'help' => 'Settings', 'parser' => $subcommandParserFull, ]) ->addSubcommand('reset', [ 'help' => 'Manually reset (failed) jobs for re-run.', 'parser' => $subcommandParserFull, ]) ->addSubcommand('hard_reset', [ 'help' => 'Hard reset queue (remove all jobs)', 'parser' => $subcommandParserFull, ]) ->addSubcommand('kill', [ 'help' => 'Manually kill a worker.', 'parser' => $subcommandParserFull, ]) ->addSubcommand('runworker', [ 'help' => 'Run Worker', 'parser' => $subcommandParserFull, ]); } /** * Timestamped log. * * @param string $type Log type * @param int|null $pid PID of the process * @return void */ protected function _log($type, $pid = null) { # log? if (Configure::read('Queue.log')) { $folder = LOGS . 'queue'; if (!file_exists($folder)) { mkdir($folder, 0755, true); } $message = $type . ' ' . $pid; // skip for now //Log::write('queue', $message); } } /** * Timestamped notification. * * @return void */ protected function _notify() { # log? if (Configure::read('Queue.notify')) { $folder = TMP; $file = $folder . 'queue_notification' . '.txt'; touch($file); } } /** * Returns a List of available QueueTasks and their individual configurations. * * @return array */ protected function _getTaskConf() { if (!is_array($this->_taskConf)) { $this->_taskConf = []; foreach ($this->tasks as $task) { list($pluginName, $taskName) = pluginSplit($task); $this->_taskConf[$taskName]['name'] = substr($taskName, 5); $this->_taskConf[$taskName]['plugin'] = $pluginName; if (property_exists($this->{$taskName}, 'timeout')) { $this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout; } else { $this->_taskConf[$taskName]['timeout'] = Configure::read('Queue.defaultworkertimeout'); } if (property_exists($this->{$taskName}, 'retries')) { $this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries; } else { $this->_taskConf[$taskName]['retries'] = Configure::read('Queue.defaultworkerretries'); } if (property_exists($this->{$taskName}, 'rate')) { $this->_taskConf[$taskName]['rate'] = $this->{$taskName}->rate; } } } return $this->_taskConf; } /** * Signal handling to queue worker for clean shutdown * * @param int $signal not used * @return void */ protected function _exit($signal) { $this->_exit = true; } /** * Destructor, removes pid-file */ public function __destruct() { $pidFilePath = Configure::read('Queue.pidfilepath'); if (!$pidFilePath) { return; } if (function_exists('posix_getpid')) { $pid = posix_getpid(); } else { $pid = $this->QueuedJobs->key(); } $file = $pidFilePath . 'queue_' . $pid . '.pid'; if (file_exists($file)) { unlink($file); } } /** * @return void */ protected function _displayAvailableTasks() { $this->out('Available Tasks:'); foreach ($this->taskNames as $loadedTask) { $this->out("\t" . '* ' . $this->_taskName($loadedTask)); } } }