@@ -10,63 +10,63 @@ |
||
10 | 10 | */ |
11 | 11 | class QueuedTasksTableTest extends TestCase |
12 | 12 | { |
13 | - /** |
|
14 | - * Test subject |
|
15 | - * |
|
16 | - * @var \Queue\Model\Table\QueuedTasksTable |
|
17 | - */ |
|
18 | - public $QueuedTasks; |
|
13 | + /** |
|
14 | + * Test subject |
|
15 | + * |
|
16 | + * @var \Queue\Model\Table\QueuedTasksTable |
|
17 | + */ |
|
18 | + public $QueuedTasks; |
|
19 | 19 | |
20 | - /** |
|
21 | - * Fixtures |
|
22 | - * |
|
23 | - * @var array |
|
24 | - */ |
|
25 | - public $fixtures = [ |
|
26 | - 'plugin.Queue.QueuedTasks' |
|
27 | - ]; |
|
20 | + /** |
|
21 | + * Fixtures |
|
22 | + * |
|
23 | + * @var array |
|
24 | + */ |
|
25 | + public $fixtures = [ |
|
26 | + 'plugin.Queue.QueuedTasks' |
|
27 | + ]; |
|
28 | 28 | |
29 | - /** |
|
30 | - * setUp method |
|
31 | - * |
|
32 | - * @return void |
|
33 | - */ |
|
34 | - public function setUp() |
|
35 | - { |
|
36 | - parent::setUp(); |
|
37 | - $config = TableRegistry::getTableLocator()->exists('QueuedTasks') ? [] : ['className' => QueuedTasksTable::class]; |
|
38 | - $this->QueuedTasks = TableRegistry::getTableLocator()->get('QueuedTasks', $config); |
|
39 | - } |
|
29 | + /** |
|
30 | + * setUp method |
|
31 | + * |
|
32 | + * @return void |
|
33 | + */ |
|
34 | + public function setUp() |
|
35 | + { |
|
36 | + parent::setUp(); |
|
37 | + $config = TableRegistry::getTableLocator()->exists('QueuedTasks') ? [] : ['className' => QueuedTasksTable::class]; |
|
38 | + $this->QueuedTasks = TableRegistry::getTableLocator()->get('QueuedTasks', $config); |
|
39 | + } |
|
40 | 40 | |
41 | - /** |
|
42 | - * tearDown method |
|
43 | - * |
|
44 | - * @return void |
|
45 | - */ |
|
46 | - public function tearDown() |
|
47 | - { |
|
48 | - unset($this->QueuedTasks); |
|
41 | + /** |
|
42 | + * tearDown method |
|
43 | + * |
|
44 | + * @return void |
|
45 | + */ |
|
46 | + public function tearDown() |
|
47 | + { |
|
48 | + unset($this->QueuedTasks); |
|
49 | 49 | |
50 | - parent::tearDown(); |
|
51 | - } |
|
50 | + parent::tearDown(); |
|
51 | + } |
|
52 | 52 | |
53 | - /** |
|
54 | - * Test initialize method |
|
55 | - * |
|
56 | - * @return void |
|
57 | - */ |
|
58 | - public function testInitialize() |
|
59 | - { |
|
60 | - $this->markTestIncomplete('Not implemented yet.'); |
|
61 | - } |
|
53 | + /** |
|
54 | + * Test initialize method |
|
55 | + * |
|
56 | + * @return void |
|
57 | + */ |
|
58 | + public function testInitialize() |
|
59 | + { |
|
60 | + $this->markTestIncomplete('Not implemented yet.'); |
|
61 | + } |
|
62 | 62 | |
63 | - /** |
|
64 | - * Test validationDefault method |
|
65 | - * |
|
66 | - * @return void |
|
67 | - */ |
|
68 | - public function testValidationDefault() |
|
69 | - { |
|
70 | - $this->markTestIncomplete('Not implemented yet.'); |
|
71 | - } |
|
63 | + /** |
|
64 | + * Test validationDefault method |
|
65 | + * |
|
66 | + * @return void |
|
67 | + */ |
|
68 | + public function testValidationDefault() |
|
69 | + { |
|
70 | + $this->markTestIncomplete('Not implemented yet.'); |
|
71 | + } |
|
72 | 72 | } |
@@ -2,14 +2,14 @@ discard block |
||
2 | 2 | use Cake\Datasource\ConnectionManager; |
3 | 3 | |
4 | 4 | if (!defined('DS')) { |
5 | - define('DS', DIRECTORY_SEPARATOR); |
|
5 | + define('DS', DIRECTORY_SEPARATOR); |
|
6 | 6 | } |
7 | 7 | if (!defined('WINDOWS')) { |
8 | - if (DS === '\\' || substr(PHP_OS, 0, 3) === 'WIN') { |
|
9 | - define('WINDOWS', true); |
|
10 | - } else { |
|
11 | - define('WINDOWS', false); |
|
12 | - } |
|
8 | + if (DS === '\\' || substr(PHP_OS, 0, 3) === 'WIN') { |
|
9 | + define('WINDOWS', true); |
|
10 | + } else { |
|
11 | + define('WINDOWS', false); |
|
12 | + } |
|
13 | 13 | } |
14 | 14 | |
15 | 15 | define('ROOT', dirname(__DIR__)); |
@@ -32,25 +32,25 @@ discard block |
||
32 | 32 | require CORE_PATH . 'config/bootstrap.php'; |
33 | 33 | |
34 | 34 | Cake\Core\Configure::write('App', [ |
35 | - 'namespace' => 'App', |
|
36 | - 'encoding' => 'UTF-8', |
|
37 | - 'paths' => [ |
|
38 | - 'templates' => [ROOT . DS . 'tests' . DS . 'test_app' . DS . 'src' . DS . 'Template' . DS], |
|
39 | - ] |
|
35 | + 'namespace' => 'App', |
|
36 | + 'encoding' => 'UTF-8', |
|
37 | + 'paths' => [ |
|
38 | + 'templates' => [ROOT . DS . 'tests' . DS . 'test_app' . DS . 'src' . DS . 'Template' . DS], |
|
39 | + ] |
|
40 | 40 | ]); |
41 | 41 | |
42 | 42 | Cake\Core\Configure::write('debug', true); |
43 | 43 | |
44 | 44 | Cake\Core\Configure::write('EmailTransport', [ |
45 | - 'default' => [ |
|
46 | - 'className' => 'Debug', |
|
47 | - ], |
|
45 | + 'default' => [ |
|
46 | + 'className' => 'Debug', |
|
47 | + ], |
|
48 | 48 | ]); |
49 | 49 | Cake\Core\Configure::write('Email', [ |
50 | - 'default' => [ |
|
51 | - 'transport' => 'default', |
|
52 | - 'from' => 'you@localhost', |
|
53 | - ], |
|
50 | + 'default' => [ |
|
51 | + 'transport' => 'default', |
|
52 | + 'from' => 'you@localhost', |
|
53 | + ], |
|
54 | 54 | ]); |
55 | 55 | |
56 | 56 | mb_internal_encoding('UTF-8'); |
@@ -61,24 +61,24 @@ discard block |
||
61 | 61 | $Tmp->create(TMP . 'cache/views', 0770); |
62 | 62 | |
63 | 63 | $cache = [ |
64 | - 'default' => [ |
|
65 | - 'engine' => 'File', |
|
66 | - 'path' => CACHE, |
|
67 | - ], |
|
68 | - '_cake_core_' => [ |
|
69 | - 'className' => 'File', |
|
70 | - 'prefix' => 'crud_myapp_cake_core_', |
|
71 | - 'path' => CACHE . 'persistent/', |
|
72 | - 'serialize' => true, |
|
73 | - 'duration' => '+10 seconds', |
|
74 | - ], |
|
75 | - '_cake_model_' => [ |
|
76 | - 'className' => 'File', |
|
77 | - 'prefix' => 'crud_my_app_cake_model_', |
|
78 | - 'path' => CACHE . 'models/', |
|
79 | - 'serialize' => 'File', |
|
80 | - 'duration' => '+10 seconds', |
|
81 | - ], |
|
64 | + 'default' => [ |
|
65 | + 'engine' => 'File', |
|
66 | + 'path' => CACHE, |
|
67 | + ], |
|
68 | + '_cake_core_' => [ |
|
69 | + 'className' => 'File', |
|
70 | + 'prefix' => 'crud_myapp_cake_core_', |
|
71 | + 'path' => CACHE . 'persistent/', |
|
72 | + 'serialize' => true, |
|
73 | + 'duration' => '+10 seconds', |
|
74 | + ], |
|
75 | + '_cake_model_' => [ |
|
76 | + 'className' => 'File', |
|
77 | + 'prefix' => 'crud_my_app_cake_model_', |
|
78 | + 'path' => CACHE . 'models/', |
|
79 | + 'serialize' => 'File', |
|
80 | + 'duration' => '+10 seconds', |
|
81 | + ], |
|
82 | 82 | ]; |
83 | 83 | |
84 | 84 | Cake\Cache\Cache::setConfig($cache); |
@@ -86,36 +86,36 @@ discard block |
||
86 | 86 | Cake\Core\Plugin::getCollection()->add(new \Queue\Plugin()); |
87 | 87 | |
88 | 88 | Cake\Mailer\TransportFactory::setConfig('default', [ |
89 | - 'className' => 'Debug', |
|
89 | + 'className' => 'Debug', |
|
90 | 90 | ]); |
91 | 91 | Cake\Mailer\TransportFactory::setConfig('queue', [ |
92 | - 'className' => 'Queue.Queue', |
|
92 | + 'className' => 'Queue.Queue', |
|
93 | 93 | ]); |
94 | 94 | Cake\Mailer\Email::setConfig('default', [ |
95 | - 'transport' => 'default', |
|
95 | + 'transport' => 'default', |
|
96 | 96 | ]); |
97 | 97 | |
98 | 98 | // Allow local overwrite |
99 | 99 | // E.g. in your console: export db_dsn="mysql://root:[email protected]/cake_test" |
100 | 100 | if (!getenv('db_class') && getenv('db_dsn')) { |
101 | - ConnectionManager::setConfig('test', ['url' => getenv('db_dsn')]); |
|
102 | - return; |
|
101 | + ConnectionManager::setConfig('test', ['url' => getenv('db_dsn')]); |
|
102 | + return; |
|
103 | 103 | } |
104 | 104 | if (!getenv('db_class')) { |
105 | - putenv('db_class=Cake\Database\Driver\Sqlite'); |
|
106 | - putenv('db_dsn=sqlite::memory:'); |
|
105 | + putenv('db_class=Cake\Database\Driver\Sqlite'); |
|
106 | + putenv('db_dsn=sqlite::memory:'); |
|
107 | 107 | } |
108 | 108 | |
109 | 109 | // Uses Travis config then (MySQL, Postgres, ...) |
110 | 110 | ConnectionManager::setConfig('test', [ |
111 | - 'className' => 'Cake\Database\Connection', |
|
112 | - 'driver' => getenv('db_class'), |
|
113 | - 'dsn' => getenv('db_dsn'), |
|
114 | - 'database' => getenv('db_database'), |
|
115 | - 'username' => getenv('db_username'), |
|
116 | - 'password' => getenv('db_password'), |
|
117 | - 'timezone' => 'UTC', |
|
118 | - 'quoteIdentifiers' => true, |
|
119 | - 'cacheMetadata' => true, |
|
111 | + 'className' => 'Cake\Database\Connection', |
|
112 | + 'driver' => getenv('db_class'), |
|
113 | + 'dsn' => getenv('db_dsn'), |
|
114 | + 'database' => getenv('db_database'), |
|
115 | + 'username' => getenv('db_username'), |
|
116 | + 'password' => getenv('db_password'), |
|
117 | + 'timezone' => 'UTC', |
|
118 | + 'quoteIdentifiers' => true, |
|
119 | + 'cacheMetadata' => true, |
|
120 | 120 | |
121 | 121 | ]); |
@@ -8,80 +8,80 @@ |
||
8 | 8 | class TaskFinder |
9 | 9 | { |
10 | 10 | |
11 | - /** |
|
12 | - * |
|
13 | - * @var array|null |
|
14 | - */ |
|
15 | - protected $tasks; |
|
11 | + /** |
|
12 | + * |
|
13 | + * @var array|null |
|
14 | + */ |
|
15 | + protected $tasks; |
|
16 | 16 | |
17 | - /** |
|
18 | - * Returns all possible Queue tasks. |
|
19 | - * |
|
20 | - * Makes sure that app tasks are prioritized over plugin ones. |
|
21 | - * |
|
22 | - * @return array |
|
23 | - */ |
|
24 | - public function allAppAndPluginTasks() |
|
25 | - { |
|
26 | - if ($this->tasks !== null) { |
|
27 | - return $this->tasks; |
|
28 | - } |
|
17 | + /** |
|
18 | + * Returns all possible Queue tasks. |
|
19 | + * |
|
20 | + * Makes sure that app tasks are prioritized over plugin ones. |
|
21 | + * |
|
22 | + * @return array |
|
23 | + */ |
|
24 | + public function allAppAndPluginTasks() |
|
25 | + { |
|
26 | + if ($this->tasks !== null) { |
|
27 | + return $this->tasks; |
|
28 | + } |
|
29 | 29 | |
30 | - $paths = App::path('Shell/Task'); |
|
31 | - $this->tasks = []; |
|
30 | + $paths = App::path('Shell/Task'); |
|
31 | + $this->tasks = []; |
|
32 | 32 | |
33 | - foreach ($paths as $path) { |
|
34 | - $Folder = new Folder($path); |
|
35 | - $this->tasks = $this->getAppPaths($Folder); |
|
36 | - } |
|
37 | - $plugins = Plugin::loaded(); |
|
38 | - foreach ($plugins as $plugin) { |
|
39 | - $pluginPaths = App::path('Shell/Task', $plugin); |
|
40 | - foreach ($pluginPaths as $pluginPath) { |
|
41 | - $Folder = new Folder($pluginPath); |
|
42 | - $pluginTasks = $this->getPluginPaths($Folder, $plugin); |
|
43 | - $this->tasks = array_merge($this->tasks, $pluginTasks); |
|
44 | - } |
|
45 | - } |
|
33 | + foreach ($paths as $path) { |
|
34 | + $Folder = new Folder($path); |
|
35 | + $this->tasks = $this->getAppPaths($Folder); |
|
36 | + } |
|
37 | + $plugins = Plugin::loaded(); |
|
38 | + foreach ($plugins as $plugin) { |
|
39 | + $pluginPaths = App::path('Shell/Task', $plugin); |
|
40 | + foreach ($pluginPaths as $pluginPath) { |
|
41 | + $Folder = new Folder($pluginPath); |
|
42 | + $pluginTasks = $this->getPluginPaths($Folder, $plugin); |
|
43 | + $this->tasks = array_merge($this->tasks, $pluginTasks); |
|
44 | + } |
|
45 | + } |
|
46 | 46 | |
47 | - return $this->tasks; |
|
48 | - } |
|
47 | + return $this->tasks; |
|
48 | + } |
|
49 | 49 | |
50 | - /** |
|
51 | - * |
|
52 | - * @param \Cake\Filesystem\Folder $Folder |
|
53 | - * |
|
54 | - * @return array |
|
55 | - */ |
|
56 | - protected function getAppPaths(Folder $Folder) |
|
57 | - { |
|
58 | - $res = array_merge($this->tasks, $Folder->find('Queue.+\.php')); |
|
59 | - foreach ($res as &$r) { |
|
60 | - $r = basename($r, 'Task.php'); |
|
61 | - } |
|
50 | + /** |
|
51 | + * |
|
52 | + * @param \Cake\Filesystem\Folder $Folder |
|
53 | + * |
|
54 | + * @return array |
|
55 | + */ |
|
56 | + protected function getAppPaths(Folder $Folder) |
|
57 | + { |
|
58 | + $res = array_merge($this->tasks, $Folder->find('Queue.+\.php')); |
|
59 | + foreach ($res as &$r) { |
|
60 | + $r = basename($r, 'Task.php'); |
|
61 | + } |
|
62 | 62 | |
63 | - return $res; |
|
64 | - } |
|
63 | + return $res; |
|
64 | + } |
|
65 | 65 | |
66 | - /** |
|
67 | - * |
|
68 | - * @param \Cake\Filesystem\Folder $Folder |
|
69 | - * @param string $plugin |
|
70 | - * |
|
71 | - * @return array |
|
72 | - */ |
|
73 | - protected function getPluginPaths(Folder $Folder, $plugin) |
|
74 | - { |
|
75 | - $res = $Folder->find('Queue.+Task\.php'); |
|
76 | - foreach ($res as $key => $r) { |
|
77 | - $name = basename($r, 'Task.php'); |
|
78 | - if (in_array($name, $this->tasks)) { |
|
79 | - unset($res[$key]); |
|
80 | - continue; |
|
81 | - } |
|
82 | - $res[$key] = $plugin . '.' . $name; |
|
83 | - } |
|
66 | + /** |
|
67 | + * |
|
68 | + * @param \Cake\Filesystem\Folder $Folder |
|
69 | + * @param string $plugin |
|
70 | + * |
|
71 | + * @return array |
|
72 | + */ |
|
73 | + protected function getPluginPaths(Folder $Folder, $plugin) |
|
74 | + { |
|
75 | + $res = $Folder->find('Queue.+Task\.php'); |
|
76 | + foreach ($res as $key => $r) { |
|
77 | + $name = basename($r, 'Task.php'); |
|
78 | + if (in_array($name, $this->tasks)) { |
|
79 | + unset($res[$key]); |
|
80 | + continue; |
|
81 | + } |
|
82 | + $res[$key] = $plugin . '.' . $name; |
|
83 | + } |
|
84 | 84 | |
85 | - return $res; |
|
86 | - } |
|
85 | + return $res; |
|
86 | + } |
|
87 | 87 | } |
@@ -6,66 +6,66 @@ |
||
6 | 6 | class Config |
7 | 7 | { |
8 | 8 | |
9 | - /** |
|
10 | - * |
|
11 | - * @return int |
|
12 | - */ |
|
13 | - public static function defaultWorkerTimeout() |
|
14 | - { |
|
15 | - return Configure::read('Queue.defaultWorkerTimeout', 600); // 10min |
|
16 | - } |
|
9 | + /** |
|
10 | + * |
|
11 | + * @return int |
|
12 | + */ |
|
13 | + public static function defaultWorkerTimeout() |
|
14 | + { |
|
15 | + return Configure::read('Queue.defaultWorkerTimeout', 600); // 10min |
|
16 | + } |
|
17 | 17 | |
18 | - /** |
|
19 | - * |
|
20 | - * @return int |
|
21 | - */ |
|
22 | - public static function workerMaxRuntime() |
|
23 | - { |
|
24 | - return Configure::read('Queue.workerMaxRuntime', 120); |
|
25 | - } |
|
18 | + /** |
|
19 | + * |
|
20 | + * @return int |
|
21 | + */ |
|
22 | + public static function workerMaxRuntime() |
|
23 | + { |
|
24 | + return Configure::read('Queue.workerMaxRuntime', 120); |
|
25 | + } |
|
26 | 26 | |
27 | - /** |
|
28 | - * |
|
29 | - * @return int |
|
30 | - */ |
|
31 | - public static function cleanupTimeout() |
|
32 | - { |
|
33 | - return Configure::read('Queue.cleanupTimeout', 2592000); // 30 days |
|
34 | - } |
|
27 | + /** |
|
28 | + * |
|
29 | + * @return int |
|
30 | + */ |
|
31 | + public static function cleanupTimeout() |
|
32 | + { |
|
33 | + return Configure::read('Queue.cleanupTimeout', 2592000); // 30 days |
|
34 | + } |
|
35 | 35 | |
36 | - /** |
|
37 | - * |
|
38 | - * @return int |
|
39 | - */ |
|
40 | - public static function sleepTime() |
|
41 | - { |
|
42 | - return Configure::read('Queue.sleepTime', 10); |
|
43 | - } |
|
36 | + /** |
|
37 | + * |
|
38 | + * @return int |
|
39 | + */ |
|
40 | + public static function sleepTime() |
|
41 | + { |
|
42 | + return Configure::read('Queue.sleepTime', 10); |
|
43 | + } |
|
44 | 44 | |
45 | - /** |
|
46 | - * |
|
47 | - * @return int |
|
48 | - */ |
|
49 | - public static function gcprob() |
|
50 | - { |
|
51 | - return Configure::read('Queue.gcprob', 10); |
|
52 | - } |
|
45 | + /** |
|
46 | + * |
|
47 | + * @return int |
|
48 | + */ |
|
49 | + public static function gcprob() |
|
50 | + { |
|
51 | + return Configure::read('Queue.gcprob', 10); |
|
52 | + } |
|
53 | 53 | |
54 | - /** |
|
55 | - * |
|
56 | - * @return int |
|
57 | - */ |
|
58 | - public static function defaultWorkerRetries() |
|
59 | - { |
|
60 | - return Configure::read('Queue.defaultWorkerRetries', 1); |
|
61 | - } |
|
54 | + /** |
|
55 | + * |
|
56 | + * @return int |
|
57 | + */ |
|
58 | + public static function defaultWorkerRetries() |
|
59 | + { |
|
60 | + return Configure::read('Queue.defaultWorkerRetries', 1); |
|
61 | + } |
|
62 | 62 | |
63 | - /** |
|
64 | - * |
|
65 | - * @return int |
|
66 | - */ |
|
67 | - public static function maxWorkers() |
|
68 | - { |
|
69 | - return Configure::read('Queue.maxWorkers', 1); |
|
70 | - } |
|
63 | + /** |
|
64 | + * |
|
65 | + * @return int |
|
66 | + */ |
|
67 | + public static function maxWorkers() |
|
68 | + { |
|
69 | + return Configure::read('Queue.maxWorkers', 1); |
|
70 | + } |
|
71 | 71 | } |
@@ -7,84 +7,84 @@ |
||
7 | 7 | class QueueExampleTask extends QueueTask implements AddInterface |
8 | 8 | { |
9 | 9 | |
10 | - /** |
|
11 | - * Timeout for run, after which the task is reassigned to a new worker. |
|
12 | - * |
|
13 | - * @var int |
|
14 | - */ |
|
15 | - public $timeout = 10; |
|
10 | + /** |
|
11 | + * Timeout for run, after which the task is reassigned to a new worker. |
|
12 | + * |
|
13 | + * @var int |
|
14 | + */ |
|
15 | + public $timeout = 10; |
|
16 | 16 | |
17 | - /** |
|
18 | - * Timeout for cleanup, after which completed jobs are deleted (in seconds). |
|
19 | - * |
|
20 | - * @var int |
|
21 | - */ |
|
22 | - public $cleanupTimeout = 600; |
|
17 | + /** |
|
18 | + * Timeout for cleanup, after which completed jobs are deleted (in seconds). |
|
19 | + * |
|
20 | + * @var int |
|
21 | + */ |
|
22 | + public $cleanupTimeout = 600; |
|
23 | 23 | |
24 | - /** |
|
25 | - * Number of times a failed instance of this task should be restarted before giving up. |
|
26 | - * |
|
27 | - * @var int |
|
28 | - */ |
|
29 | - public $retries = 0; |
|
24 | + /** |
|
25 | + * Number of times a failed instance of this task should be restarted before giving up. |
|
26 | + * |
|
27 | + * @var int |
|
28 | + */ |
|
29 | + public $retries = 0; |
|
30 | 30 | |
31 | - /** |
|
32 | - * Stores any failure messages triggered during run(). |
|
33 | - * |
|
34 | - * @var string |
|
35 | - */ |
|
36 | - public $failureMessage = ''; |
|
31 | + /** |
|
32 | + * Stores any failure messages triggered during run(). |
|
33 | + * |
|
34 | + * @var string |
|
35 | + */ |
|
36 | + public $failureMessage = ''; |
|
37 | 37 | |
38 | - /** |
|
39 | - * Example add functionality. |
|
40 | - * Will create one example job in the queue, which later will be executed using run(); |
|
41 | - * |
|
42 | - * To invoke from CLI execute: |
|
43 | - * - bin/cake queue add Example |
|
44 | - * |
|
45 | - * @return void |
|
46 | - */ |
|
47 | - public function add(): void |
|
48 | - { |
|
49 | - $this->out(__d('queue', 'CakePHP Queue Example task.')); |
|
50 | - $this->hr(); |
|
51 | - $this->out(__d('queue', 'This is a very simple example of a queueTask.')); |
|
52 | - $this->out(__d('queue', 'Now adding an example Task Job into the Queue.')); |
|
53 | - $this->out(__d('queue', 'This task will only produce some console output on the worker that it runs on.')); |
|
54 | - $this->out(' '); |
|
55 | - $this->out(__d('queue', 'To run a Worker use:')); |
|
56 | - $this->out(__d('queue', ' cake queue runworker')); |
|
57 | - $this->out(' '); |
|
58 | - $this->out(__d('queue', 'You can find the sourcecode of this task in: ')); |
|
59 | - $this->out(__FILE__); |
|
60 | - $this->out(' '); |
|
38 | + /** |
|
39 | + * Example add functionality. |
|
40 | + * Will create one example job in the queue, which later will be executed using run(); |
|
41 | + * |
|
42 | + * To invoke from CLI execute: |
|
43 | + * - bin/cake queue add Example |
|
44 | + * |
|
45 | + * @return void |
|
46 | + */ |
|
47 | + public function add(): void |
|
48 | + { |
|
49 | + $this->out(__d('queue', 'CakePHP Queue Example task.')); |
|
50 | + $this->hr(); |
|
51 | + $this->out(__d('queue', 'This is a very simple example of a queueTask.')); |
|
52 | + $this->out(__d('queue', 'Now adding an example Task Job into the Queue.')); |
|
53 | + $this->out(__d('queue', 'This task will only produce some console output on the worker that it runs on.')); |
|
54 | + $this->out(' '); |
|
55 | + $this->out(__d('queue', 'To run a Worker use:')); |
|
56 | + $this->out(__d('queue', ' cake queue runworker')); |
|
57 | + $this->out(' '); |
|
58 | + $this->out(__d('queue', 'You can find the sourcecode of this task in: ')); |
|
59 | + $this->out(__FILE__); |
|
60 | + $this->out(' '); |
|
61 | 61 | |
62 | - // Adding a task of type 'example' with no additionally passed data |
|
63 | - if ($this->QueuedTasks->createJob('Example')) { |
|
64 | - $this->out(__d('queue', 'OK, job created, now run the worker')); |
|
65 | - } else { |
|
66 | - $this->err(__d('queue', 'Could not create Job')); |
|
67 | - } |
|
68 | - } |
|
62 | + // Adding a task of type 'example' with no additionally passed data |
|
63 | + if ($this->QueuedTasks->createJob('Example')) { |
|
64 | + $this->out(__d('queue', 'OK, job created, now run the worker')); |
|
65 | + } else { |
|
66 | + $this->err(__d('queue', 'Could not create Job')); |
|
67 | + } |
|
68 | + } |
|
69 | 69 | |
70 | - /** |
|
71 | - * Example run function. |
|
72 | - * This function is executed, when a worker is executing a task. |
|
73 | - * The return parameter will determine, if the task will be marked completed, or be requeued. |
|
74 | - * |
|
75 | - * @param array $data |
|
76 | - * The array passed to QueuedTasksTable::createJob() |
|
77 | - * @param int $taskId |
|
78 | - * The id of the QueuedTask entity |
|
79 | - * @return void |
|
80 | - */ |
|
81 | - public function run(array $data, $taskId): void |
|
82 | - { |
|
83 | - $this->hr(); |
|
84 | - $this->out(__d('queue', 'CakePHP Queue Example task.')); |
|
85 | - $this->hr(); |
|
86 | - $this->out(__d('queue', ' ->Success, the Example Task was run.<-')); |
|
87 | - $this->out(' '); |
|
88 | - $this->out(' '); |
|
89 | - } |
|
70 | + /** |
|
71 | + * Example run function. |
|
72 | + * This function is executed, when a worker is executing a task. |
|
73 | + * The return parameter will determine, if the task will be marked completed, or be requeued. |
|
74 | + * |
|
75 | + * @param array $data |
|
76 | + * The array passed to QueuedTasksTable::createJob() |
|
77 | + * @param int $taskId |
|
78 | + * The id of the QueuedTask entity |
|
79 | + * @return void |
|
80 | + */ |
|
81 | + public function run(array $data, $taskId): void |
|
82 | + { |
|
83 | + $this->hr(); |
|
84 | + $this->out(__d('queue', 'CakePHP Queue Example task.')); |
|
85 | + $this->hr(); |
|
86 | + $this->out(__d('queue', ' ->Success, the Example Task was run.<-')); |
|
87 | + $this->out(' '); |
|
88 | + $this->out(' '); |
|
89 | + } |
|
90 | 90 | } |
@@ -18,58 +18,58 @@ |
||
18 | 18 | abstract class QueueTask extends Shell implements QueueTaskInterface |
19 | 19 | { |
20 | 20 | |
21 | - /** |
|
22 | - * |
|
23 | - * @var string |
|
24 | - */ |
|
25 | - public $queueModelClass = 'Queue.QueuedTasks'; |
|
21 | + /** |
|
22 | + * |
|
23 | + * @var string |
|
24 | + */ |
|
25 | + public $queueModelClass = 'Queue.QueuedTasks'; |
|
26 | 26 | |
27 | - /** |
|
28 | - * |
|
29 | - * @var \Queue\Model\Table\QueuedTasksTable |
|
30 | - */ |
|
31 | - public $QueuedTasks; |
|
27 | + /** |
|
28 | + * |
|
29 | + * @var \Queue\Model\Table\QueuedTasksTable |
|
30 | + */ |
|
31 | + public $QueuedTasks; |
|
32 | 32 | |
33 | - /** |
|
34 | - * Timeout for run, after which the Task is reassigned to a new worker. |
|
35 | - * |
|
36 | - * @var int |
|
37 | - */ |
|
38 | - public $timeout = 120; |
|
33 | + /** |
|
34 | + * Timeout for run, after which the Task is reassigned to a new worker. |
|
35 | + * |
|
36 | + * @var int |
|
37 | + */ |
|
38 | + public $timeout = 120; |
|
39 | 39 | |
40 | - /** |
|
41 | - * Number of times a failed instance of this task should be restarted before giving up. |
|
42 | - * |
|
43 | - * @var int |
|
44 | - */ |
|
45 | - public $retries = 1; |
|
40 | + /** |
|
41 | + * Number of times a failed instance of this task should be restarted before giving up. |
|
42 | + * |
|
43 | + * @var int |
|
44 | + */ |
|
45 | + public $retries = 1; |
|
46 | 46 | |
47 | - /** |
|
48 | - * |
|
49 | - * @param \Cake\Console\ConsoleIo|null $io |
|
50 | - * IO |
|
51 | - */ |
|
52 | - public function __construct(ConsoleIo $io = null) |
|
53 | - { |
|
54 | - parent::__construct($io); |
|
47 | + /** |
|
48 | + * |
|
49 | + * @param \Cake\Console\ConsoleIo|null $io |
|
50 | + * IO |
|
51 | + */ |
|
52 | + public function __construct(ConsoleIo $io = null) |
|
53 | + { |
|
54 | + parent::__construct($io); |
|
55 | 55 | |
56 | - $this->loadModel($this->queueModelClass); |
|
57 | - } |
|
56 | + $this->loadModel($this->queueModelClass); |
|
57 | + } |
|
58 | 58 | |
59 | - /** |
|
60 | - * |
|
61 | - * @return string |
|
62 | - * @throws \InvalidArgumentException |
|
63 | - */ |
|
64 | - protected function queueTaskName() |
|
65 | - { |
|
66 | - $class = get_class($this); |
|
59 | + /** |
|
60 | + * |
|
61 | + * @return string |
|
62 | + * @throws \InvalidArgumentException |
|
63 | + */ |
|
64 | + protected function queueTaskName() |
|
65 | + { |
|
66 | + $class = get_class($this); |
|
67 | 67 | |
68 | - preg_match('#\\\\Queue(.+)Task$#', $class, $matches); |
|
69 | - if (! $matches) { |
|
70 | - throw new InvalidArgumentException('Invalid class name: ' . $class); |
|
71 | - } |
|
68 | + preg_match('#\\\\Queue(.+)Task$#', $class, $matches); |
|
69 | + if (! $matches) { |
|
70 | + throw new InvalidArgumentException('Invalid class name: ' . $class); |
|
71 | + } |
|
72 | 72 | |
73 | - return $matches[1]; |
|
74 | - } |
|
73 | + return $matches[1]; |
|
74 | + } |
|
75 | 75 | } |
@@ -66,7 +66,7 @@ |
||
66 | 66 | $class = get_class($this); |
67 | 67 | |
68 | 68 | preg_match('#\\\\Queue(.+)Task$#', $class, $matches); |
69 | - if (! $matches) { |
|
69 | + if (!$matches) { |
|
70 | 70 | throw new InvalidArgumentException('Invalid class name: ' . $class); |
71 | 71 | } |
72 | 72 |
@@ -10,12 +10,12 @@ |
||
10 | 10 | interface AddInterface |
11 | 11 | { |
12 | 12 | |
13 | - /** |
|
14 | - * Allows adding a task to the queue. |
|
15 | - * |
|
16 | - * Will create one example task in the queue, which later will be executed using run(). |
|
17 | - * |
|
18 | - * @return void |
|
19 | - */ |
|
20 | - public function add(): void; |
|
13 | + /** |
|
14 | + * Allows adding a task to the queue. |
|
15 | + * |
|
16 | + * Will create one example task in the queue, which later will be executed using run(). |
|
17 | + * |
|
18 | + * @return void |
|
19 | + */ |
|
20 | + public function add(): void; |
|
21 | 21 | } |
@@ -13,14 +13,14 @@ |
||
13 | 13 | interface QueueTaskInterface |
14 | 14 | { |
15 | 15 | |
16 | - /** |
|
17 | - * Main execution of the task. |
|
18 | - * |
|
19 | - * @param array $data |
|
20 | - * The array passed to QueuedTasksTable::createJob() |
|
21 | - * @param int $taskId |
|
22 | - * The id of the QueuedTask entity |
|
23 | - * @return void |
|
24 | - */ |
|
25 | - public function run(array $data, $taskId): void; |
|
16 | + /** |
|
17 | + * Main execution of the task. |
|
18 | + * |
|
19 | + * @param array $data |
|
20 | + * The array passed to QueuedTasksTable::createJob() |
|
21 | + * @param int $taskId |
|
22 | + * The id of the QueuedTask entity |
|
23 | + * @return void |
|
24 | + */ |
|
25 | + public function run(array $data, $taskId): void; |
|
26 | 26 | } |
@@ -30,574 +30,574 @@ |
||
30 | 30 | class QueueShell extends Shell |
31 | 31 | { |
32 | 32 | |
33 | - /** |
|
34 | - * |
|
35 | - * @var string |
|
36 | - */ |
|
37 | - public $modelClass = 'Queue.QueuedTasks'; |
|
38 | - |
|
39 | - /** |
|
40 | - * |
|
41 | - * @var array|null |
|
42 | - */ |
|
43 | - protected $_taskConf; |
|
44 | - |
|
45 | - /** |
|
46 | - * |
|
47 | - * @var int |
|
48 | - */ |
|
49 | - protected $_time = 0; |
|
50 | - |
|
51 | - /** |
|
52 | - * |
|
53 | - * @var bool |
|
54 | - */ |
|
55 | - protected $_exit = false; |
|
56 | - |
|
57 | - /** |
|
58 | - * Overwrite shell initialize to dynamically load all Queue Related Tasks. |
|
59 | - * |
|
60 | - * @return void |
|
61 | - */ |
|
62 | - public function initialize() |
|
63 | - { |
|
64 | - $taskFinder = new TaskFinder(); |
|
65 | - $this->tasks = $taskFinder->allAppAndPluginTasks(); |
|
66 | - |
|
67 | - parent::initialize(); |
|
68 | - } |
|
69 | - |
|
70 | - /** |
|
71 | - * |
|
72 | - * @return void |
|
73 | - */ |
|
74 | - public function startup() |
|
75 | - { |
|
76 | - if ($this->param('quiet')) { |
|
77 | - $this->interactive = false; |
|
78 | - } |
|
79 | - |
|
80 | - parent::startup(); |
|
81 | - } |
|
82 | - |
|
83 | - /** |
|
84 | - * |
|
85 | - * @return string |
|
86 | - */ |
|
87 | - public function _getDescription() |
|
88 | - { |
|
89 | - $tasks = []; |
|
90 | - foreach ($this->taskNames as $loadedTask) { |
|
91 | - $tasks[] = "\t" . '* ' . $this->_taskName($loadedTask); |
|
92 | - } |
|
93 | - $tasks = implode(PHP_EOL, $tasks); |
|
94 | - |
|
95 | - $text = <<<TEXT |
|
33 | + /** |
|
34 | + * |
|
35 | + * @var string |
|
36 | + */ |
|
37 | + public $modelClass = 'Queue.QueuedTasks'; |
|
38 | + |
|
39 | + /** |
|
40 | + * |
|
41 | + * @var array|null |
|
42 | + */ |
|
43 | + protected $_taskConf; |
|
44 | + |
|
45 | + /** |
|
46 | + * |
|
47 | + * @var int |
|
48 | + */ |
|
49 | + protected $_time = 0; |
|
50 | + |
|
51 | + /** |
|
52 | + * |
|
53 | + * @var bool |
|
54 | + */ |
|
55 | + protected $_exit = false; |
|
56 | + |
|
57 | + /** |
|
58 | + * Overwrite shell initialize to dynamically load all Queue Related Tasks. |
|
59 | + * |
|
60 | + * @return void |
|
61 | + */ |
|
62 | + public function initialize() |
|
63 | + { |
|
64 | + $taskFinder = new TaskFinder(); |
|
65 | + $this->tasks = $taskFinder->allAppAndPluginTasks(); |
|
66 | + |
|
67 | + parent::initialize(); |
|
68 | + } |
|
69 | + |
|
70 | + /** |
|
71 | + * |
|
72 | + * @return void |
|
73 | + */ |
|
74 | + public function startup() |
|
75 | + { |
|
76 | + if ($this->param('quiet')) { |
|
77 | + $this->interactive = false; |
|
78 | + } |
|
79 | + |
|
80 | + parent::startup(); |
|
81 | + } |
|
82 | + |
|
83 | + /** |
|
84 | + * |
|
85 | + * @return string |
|
86 | + */ |
|
87 | + public function _getDescription() |
|
88 | + { |
|
89 | + $tasks = []; |
|
90 | + foreach ($this->taskNames as $loadedTask) { |
|
91 | + $tasks[] = "\t" . '* ' . $this->_taskName($loadedTask); |
|
92 | + } |
|
93 | + $tasks = implode(PHP_EOL, $tasks); |
|
94 | + |
|
95 | + $text = <<<TEXT |
|
96 | 96 | Simple and minimalistic job queue (or deferred-task) system. |
97 | 97 | |
98 | 98 | Available Tasks: |
99 | 99 | $tasks |
100 | 100 | TEXT; |
101 | - return $text; |
|
102 | - } |
|
103 | - |
|
104 | - /** |
|
105 | - * Look for a Queue Task of hte passed name and try to call add() on it. |
|
106 | - * A QueueTask may provide an add function to enable the user to create new jobs via commandline. |
|
107 | - * |
|
108 | - * @return void |
|
109 | - */ |
|
110 | - public function add() |
|
111 | - { |
|
112 | - if (count($this->args) < 1) { |
|
113 | - $this->out('Please call like this:'); |
|
114 | - $this->out(' bin/cake queue add <taskname>'); |
|
115 | - $this->_displayAvailableTasks(); |
|
116 | - |
|
117 | - return; |
|
118 | - } |
|
119 | - |
|
120 | - $name = Inflector::camelize($this->args[0]); |
|
121 | - if (in_array('Queue' . $name, $this->taskNames, true)) { |
|
122 | - /** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */ |
|
123 | - $task = $this->{'Queue' . $name}; |
|
124 | - if (! ($task instanceof AddInterface)) { |
|
125 | - $this->abort('This task does not support adding via CLI call'); |
|
126 | - } |
|
127 | - $task->add(); |
|
128 | - } else { |
|
129 | - $this->out('Error: Task not found: ' . $name); |
|
130 | - $this->_displayAvailableTasks(); |
|
131 | - } |
|
132 | - } |
|
133 | - |
|
134 | - /** |
|
135 | - * Output the task without Queue or Task |
|
136 | - * example: QueueImageTask becomes Image on display |
|
137 | - * |
|
138 | - * @param string $task |
|
139 | - * Task name |
|
140 | - * @return string Cleaned task name |
|
141 | - */ |
|
142 | - protected function _taskName($task) |
|
143 | - { |
|
144 | - if (strpos($task, 'Queue') === 0) { |
|
145 | - return substr($task, 5); |
|
146 | - } |
|
147 | - return $task; |
|
148 | - } |
|
149 | - |
|
150 | - /** |
|
151 | - * Run a QueueWorker loop. |
|
152 | - * Runs a Queue Worker process which will try to find unassigned jobs in the queue |
|
153 | - * which it may run and try to fetch and execute them. |
|
154 | - * |
|
155 | - * @return int|null |
|
156 | - */ |
|
157 | - public function runworker() |
|
158 | - { |
|
159 | - try { |
|
160 | - $pid = $this->_initPid(); |
|
161 | - } catch (PersistenceFailedException $exception) { |
|
162 | - $this->err($exception->getMessage()); |
|
163 | - $limit = (int) Configure::read('Queue.maxWorkers'); |
|
164 | - if ($limit) { |
|
165 | - $this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')'); |
|
166 | - } |
|
167 | - return static::CODE_ERROR; |
|
168 | - } |
|
169 | - |
|
170 | - // Enable Garbage Collector (PHP >= 5.3) |
|
171 | - if (function_exists('gc_enable')) { |
|
172 | - gc_enable(); |
|
173 | - } |
|
174 | - if (function_exists('pcntl_signal')) { |
|
175 | - pcntl_signal(SIGTERM, [ |
|
176 | - &$this, |
|
177 | - '_exit' |
|
178 | - ]); |
|
179 | - pcntl_signal(SIGINT, [ |
|
180 | - &$this, |
|
181 | - '_exit' |
|
182 | - ]); |
|
183 | - pcntl_signal(SIGTSTP, [ |
|
184 | - &$this, |
|
185 | - '_exit' |
|
186 | - ]); |
|
187 | - pcntl_signal(SIGQUIT, [ |
|
188 | - &$this, |
|
189 | - '_exit' |
|
190 | - ]); |
|
191 | - } |
|
192 | - $this->_exit = false; |
|
193 | - |
|
194 | - $startTime = time(); |
|
195 | - $types = $this->_stringToArray($this->param('type')); |
|
196 | - |
|
197 | - while (! $this->_exit) { |
|
198 | - $this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE); |
|
199 | - |
|
200 | - $QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types); |
|
201 | - |
|
202 | - if ($QueuedTask) { |
|
203 | - $this->runJob($QueuedTask, $pid); |
|
204 | - } elseif (Configure::read('Queue.exitWhenNothingToDo')) { |
|
205 | - $this->out('nothing to do, exiting.'); |
|
206 | - $this->_exit = true; |
|
207 | - } else { |
|
208 | - $this->out('nothing to do, sleeping.'); |
|
209 | - sleep(Config::sleepTime()); |
|
210 | - } |
|
211 | - |
|
212 | - // check if we are over the maximum runtime and end processing if so. |
|
213 | - if (Configure::readOrFail('Queue.workerMaxRuntime') && (time() - $startTime) >= Configure::readOrFail('Queue.workerMaxRuntime')) { |
|
214 | - $this->_exit = true; |
|
215 | - $this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.'); |
|
216 | - } |
|
217 | - if ($this->_exit || mt_rand(0, 100) > (100 - (int) Config::gcprob())) { |
|
218 | - $this->out(__d('queue', 'Performing old job cleanup.')); |
|
219 | - $this->QueuedTasks->cleanOldJobs(); |
|
220 | - } |
|
221 | - $this->hr(); |
|
222 | - } |
|
223 | - |
|
224 | - $this->_deletePid($pid); |
|
225 | - |
|
226 | - if ($this->param('verbose')) { |
|
227 | - $this->_log('endworker', $pid); |
|
228 | - } |
|
229 | - } |
|
230 | - |
|
231 | - /** |
|
232 | - * |
|
233 | - * @param \Queue\Model\Entity\QueuedTask $QueuedTask |
|
234 | - * @param string $pid |
|
235 | - * @return void |
|
236 | - */ |
|
237 | - protected function runJob(QueuedTask $QueuedTask, $pid) |
|
238 | - { |
|
239 | - $this->out('Running Job of type "' . $QueuedTask->task . '"'); |
|
240 | - $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id, $pid, false); |
|
241 | - $taskName = 'Queue' . $QueuedTask->task; |
|
242 | - |
|
243 | - try { |
|
244 | - $this->_time = time(); |
|
245 | - |
|
246 | - $data = unserialize($QueuedTask->data); |
|
247 | - /** @var \Queue\Shell\Task\QueueTask $task */ |
|
248 | - $task = $this->{$taskName}; |
|
249 | - if (! $task instanceof QueueTaskInterface) { |
|
250 | - throw new RuntimeException('Task must implement ' . QueueTaskInterface::class); |
|
251 | - } |
|
252 | - |
|
253 | - $return = $task->run((array) $data, $QueuedTask->id); |
|
254 | - if ($return !== null) { |
|
255 | - trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED); |
|
256 | - } |
|
257 | - $failureMessage = $taskName . ' failed'; |
|
258 | - } catch (Throwable $e) { |
|
259 | - $return = false; |
|
260 | - |
|
261 | - $failureMessage = get_class($e) . ': ' . $e->getMessage(); |
|
262 | - if (! ($e instanceof QueueException)) { |
|
263 | - $failureMessage .= "\n" . $e->getTraceAsString(); |
|
264 | - } |
|
265 | - |
|
266 | - $this->_logError($taskName . ' (job ' . $QueuedTask->id . ')' . "\n" . $failureMessage, $pid); |
|
267 | - } catch (Exception $e) { |
|
268 | - $return = false; |
|
269 | - |
|
270 | - $failureMessage = get_class($e) . ': ' . $e->getMessage(); |
|
271 | - $this->_logError($taskName . "\n" . $failureMessage, $pid); |
|
272 | - } |
|
273 | - |
|
274 | - if ($return === false) { |
|
275 | - $this->QueuedTasks->markJobFailed($QueuedTask, $failureMessage); |
|
276 | - $failedStatus = $this->QueuedTasks->getFailedStatus($QueuedTask, $this->_getTaskConf()); |
|
277 | - $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id . ' failed and ' . $failedStatus, $pid); |
|
278 | - $this->out('Job did not finish, ' . $failedStatus . ' after try ' . $QueuedTask->failed . '.'); |
|
279 | - return; |
|
280 | - } |
|
281 | - |
|
282 | - $this->QueuedTasks->markJobDone($QueuedTask); |
|
283 | - $this->out('Job Finished.'); |
|
284 | - } |
|
285 | - |
|
286 | - /** |
|
287 | - * Manually trigger a Finished job cleanup. |
|
288 | - * |
|
289 | - * @return void |
|
290 | - */ |
|
291 | - public function clean() |
|
292 | - { |
|
293 | - if (! Configure::read('Queue.cleanupTimeout')) { |
|
294 | - $this->abort('You disabled cleanuptimout in config. Aborting.'); |
|
295 | - } |
|
296 | - |
|
297 | - $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int) Configure::read('Queue.cleanupTimeout'))); |
|
298 | - $this->QueuedTasks->cleanOldJobs(); |
|
299 | - } |
|
300 | - |
|
301 | - /** |
|
302 | - * Display current settings |
|
303 | - * |
|
304 | - * @return void |
|
305 | - */ |
|
306 | - public function settings() |
|
307 | - { |
|
308 | - $this->out('Current Settings:'); |
|
309 | - $conf = (array) Configure::read('Queue'); |
|
310 | - foreach ($conf as $key => $val) { |
|
311 | - if ($val === false) { |
|
312 | - $val = 'no'; |
|
313 | - } |
|
314 | - if ($val === true) { |
|
315 | - $val = 'yes'; |
|
316 | - } |
|
317 | - $this->out('* ' . $key . ': ' . print_r($val, true)); |
|
318 | - } |
|
319 | - |
|
320 | - $this->out(); |
|
321 | - } |
|
322 | - |
|
323 | - /** |
|
324 | - * Display some statistics about Finished Jobs. |
|
325 | - * |
|
326 | - * @return void |
|
327 | - */ |
|
328 | - public function stats() |
|
329 | - { |
|
330 | - $this->out('Jobs currently in the queue:'); |
|
331 | - |
|
332 | - $types = $this->QueuedTasks->getTypes()->toArray(); |
|
333 | - foreach ($types as $type) { |
|
334 | - $this->out(' ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedTasks->getLength($type)); |
|
335 | - } |
|
336 | - $this->hr(); |
|
337 | - $this->out('Total unfinished jobs: ' . $this->QueuedTasks->getLength()); |
|
338 | - $this->hr(); |
|
339 | - $this->out('Finished job statistics:'); |
|
340 | - $data = $this->QueuedTasks->getStats(); |
|
341 | - foreach ($data as $item) { |
|
342 | - $this->out(' ' . $item['task'] . ': '); |
|
343 | - $this->out(' Finished Jobs in Database: ' . $item['num']); |
|
344 | - $this->out(' Average Job existence : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's'); |
|
345 | - $this->out(' Average Execution delay : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's'); |
|
346 | - $this->out(' Average Execution time : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's'); |
|
347 | - } |
|
348 | - } |
|
349 | - |
|
350 | - /** |
|
351 | - * Get option parser method to parse commandline options |
|
352 | - * |
|
353 | - * @return \Cake\Console\ConsoleOptionParser |
|
354 | - */ |
|
355 | - public function getOptionParser() |
|
356 | - { |
|
357 | - $subcommandParser = [ |
|
358 | - 'options' => [ /* |
|
101 | + return $text; |
|
102 | + } |
|
103 | + |
|
104 | + /** |
|
105 | + * Look for a Queue Task of hte passed name and try to call add() on it. |
|
106 | + * A QueueTask may provide an add function to enable the user to create new jobs via commandline. |
|
107 | + * |
|
108 | + * @return void |
|
109 | + */ |
|
110 | + public function add() |
|
111 | + { |
|
112 | + if (count($this->args) < 1) { |
|
113 | + $this->out('Please call like this:'); |
|
114 | + $this->out(' bin/cake queue add <taskname>'); |
|
115 | + $this->_displayAvailableTasks(); |
|
116 | + |
|
117 | + return; |
|
118 | + } |
|
119 | + |
|
120 | + $name = Inflector::camelize($this->args[0]); |
|
121 | + if (in_array('Queue' . $name, $this->taskNames, true)) { |
|
122 | + /** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */ |
|
123 | + $task = $this->{'Queue' . $name}; |
|
124 | + if (! ($task instanceof AddInterface)) { |
|
125 | + $this->abort('This task does not support adding via CLI call'); |
|
126 | + } |
|
127 | + $task->add(); |
|
128 | + } else { |
|
129 | + $this->out('Error: Task not found: ' . $name); |
|
130 | + $this->_displayAvailableTasks(); |
|
131 | + } |
|
132 | + } |
|
133 | + |
|
134 | + /** |
|
135 | + * Output the task without Queue or Task |
|
136 | + * example: QueueImageTask becomes Image on display |
|
137 | + * |
|
138 | + * @param string $task |
|
139 | + * Task name |
|
140 | + * @return string Cleaned task name |
|
141 | + */ |
|
142 | + protected function _taskName($task) |
|
143 | + { |
|
144 | + if (strpos($task, 'Queue') === 0) { |
|
145 | + return substr($task, 5); |
|
146 | + } |
|
147 | + return $task; |
|
148 | + } |
|
149 | + |
|
150 | + /** |
|
151 | + * Run a QueueWorker loop. |
|
152 | + * Runs a Queue Worker process which will try to find unassigned jobs in the queue |
|
153 | + * which it may run and try to fetch and execute them. |
|
154 | + * |
|
155 | + * @return int|null |
|
156 | + */ |
|
157 | + public function runworker() |
|
158 | + { |
|
159 | + try { |
|
160 | + $pid = $this->_initPid(); |
|
161 | + } catch (PersistenceFailedException $exception) { |
|
162 | + $this->err($exception->getMessage()); |
|
163 | + $limit = (int) Configure::read('Queue.maxWorkers'); |
|
164 | + if ($limit) { |
|
165 | + $this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')'); |
|
166 | + } |
|
167 | + return static::CODE_ERROR; |
|
168 | + } |
|
169 | + |
|
170 | + // Enable Garbage Collector (PHP >= 5.3) |
|
171 | + if (function_exists('gc_enable')) { |
|
172 | + gc_enable(); |
|
173 | + } |
|
174 | + if (function_exists('pcntl_signal')) { |
|
175 | + pcntl_signal(SIGTERM, [ |
|
176 | + &$this, |
|
177 | + '_exit' |
|
178 | + ]); |
|
179 | + pcntl_signal(SIGINT, [ |
|
180 | + &$this, |
|
181 | + '_exit' |
|
182 | + ]); |
|
183 | + pcntl_signal(SIGTSTP, [ |
|
184 | + &$this, |
|
185 | + '_exit' |
|
186 | + ]); |
|
187 | + pcntl_signal(SIGQUIT, [ |
|
188 | + &$this, |
|
189 | + '_exit' |
|
190 | + ]); |
|
191 | + } |
|
192 | + $this->_exit = false; |
|
193 | + |
|
194 | + $startTime = time(); |
|
195 | + $types = $this->_stringToArray($this->param('type')); |
|
196 | + |
|
197 | + while (! $this->_exit) { |
|
198 | + $this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE); |
|
199 | + |
|
200 | + $QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types); |
|
201 | + |
|
202 | + if ($QueuedTask) { |
|
203 | + $this->runJob($QueuedTask, $pid); |
|
204 | + } elseif (Configure::read('Queue.exitWhenNothingToDo')) { |
|
205 | + $this->out('nothing to do, exiting.'); |
|
206 | + $this->_exit = true; |
|
207 | + } else { |
|
208 | + $this->out('nothing to do, sleeping.'); |
|
209 | + sleep(Config::sleepTime()); |
|
210 | + } |
|
211 | + |
|
212 | + // check if we are over the maximum runtime and end processing if so. |
|
213 | + if (Configure::readOrFail('Queue.workerMaxRuntime') && (time() - $startTime) >= Configure::readOrFail('Queue.workerMaxRuntime')) { |
|
214 | + $this->_exit = true; |
|
215 | + $this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.'); |
|
216 | + } |
|
217 | + if ($this->_exit || mt_rand(0, 100) > (100 - (int) Config::gcprob())) { |
|
218 | + $this->out(__d('queue', 'Performing old job cleanup.')); |
|
219 | + $this->QueuedTasks->cleanOldJobs(); |
|
220 | + } |
|
221 | + $this->hr(); |
|
222 | + } |
|
223 | + |
|
224 | + $this->_deletePid($pid); |
|
225 | + |
|
226 | + if ($this->param('verbose')) { |
|
227 | + $this->_log('endworker', $pid); |
|
228 | + } |
|
229 | + } |
|
230 | + |
|
231 | + /** |
|
232 | + * |
|
233 | + * @param \Queue\Model\Entity\QueuedTask $QueuedTask |
|
234 | + * @param string $pid |
|
235 | + * @return void |
|
236 | + */ |
|
237 | + protected function runJob(QueuedTask $QueuedTask, $pid) |
|
238 | + { |
|
239 | + $this->out('Running Job of type "' . $QueuedTask->task . '"'); |
|
240 | + $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id, $pid, false); |
|
241 | + $taskName = 'Queue' . $QueuedTask->task; |
|
242 | + |
|
243 | + try { |
|
244 | + $this->_time = time(); |
|
245 | + |
|
246 | + $data = unserialize($QueuedTask->data); |
|
247 | + /** @var \Queue\Shell\Task\QueueTask $task */ |
|
248 | + $task = $this->{$taskName}; |
|
249 | + if (! $task instanceof QueueTaskInterface) { |
|
250 | + throw new RuntimeException('Task must implement ' . QueueTaskInterface::class); |
|
251 | + } |
|
252 | + |
|
253 | + $return = $task->run((array) $data, $QueuedTask->id); |
|
254 | + if ($return !== null) { |
|
255 | + trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED); |
|
256 | + } |
|
257 | + $failureMessage = $taskName . ' failed'; |
|
258 | + } catch (Throwable $e) { |
|
259 | + $return = false; |
|
260 | + |
|
261 | + $failureMessage = get_class($e) . ': ' . $e->getMessage(); |
|
262 | + if (! ($e instanceof QueueException)) { |
|
263 | + $failureMessage .= "\n" . $e->getTraceAsString(); |
|
264 | + } |
|
265 | + |
|
266 | + $this->_logError($taskName . ' (job ' . $QueuedTask->id . ')' . "\n" . $failureMessage, $pid); |
|
267 | + } catch (Exception $e) { |
|
268 | + $return = false; |
|
269 | + |
|
270 | + $failureMessage = get_class($e) . ': ' . $e->getMessage(); |
|
271 | + $this->_logError($taskName . "\n" . $failureMessage, $pid); |
|
272 | + } |
|
273 | + |
|
274 | + if ($return === false) { |
|
275 | + $this->QueuedTasks->markJobFailed($QueuedTask, $failureMessage); |
|
276 | + $failedStatus = $this->QueuedTasks->getFailedStatus($QueuedTask, $this->_getTaskConf()); |
|
277 | + $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id . ' failed and ' . $failedStatus, $pid); |
|
278 | + $this->out('Job did not finish, ' . $failedStatus . ' after try ' . $QueuedTask->failed . '.'); |
|
279 | + return; |
|
280 | + } |
|
281 | + |
|
282 | + $this->QueuedTasks->markJobDone($QueuedTask); |
|
283 | + $this->out('Job Finished.'); |
|
284 | + } |
|
285 | + |
|
286 | + /** |
|
287 | + * Manually trigger a Finished job cleanup. |
|
288 | + * |
|
289 | + * @return void |
|
290 | + */ |
|
291 | + public function clean() |
|
292 | + { |
|
293 | + if (! Configure::read('Queue.cleanupTimeout')) { |
|
294 | + $this->abort('You disabled cleanuptimout in config. Aborting.'); |
|
295 | + } |
|
296 | + |
|
297 | + $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int) Configure::read('Queue.cleanupTimeout'))); |
|
298 | + $this->QueuedTasks->cleanOldJobs(); |
|
299 | + } |
|
300 | + |
|
301 | + /** |
|
302 | + * Display current settings |
|
303 | + * |
|
304 | + * @return void |
|
305 | + */ |
|
306 | + public function settings() |
|
307 | + { |
|
308 | + $this->out('Current Settings:'); |
|
309 | + $conf = (array) Configure::read('Queue'); |
|
310 | + foreach ($conf as $key => $val) { |
|
311 | + if ($val === false) { |
|
312 | + $val = 'no'; |
|
313 | + } |
|
314 | + if ($val === true) { |
|
315 | + $val = 'yes'; |
|
316 | + } |
|
317 | + $this->out('* ' . $key . ': ' . print_r($val, true)); |
|
318 | + } |
|
319 | + |
|
320 | + $this->out(); |
|
321 | + } |
|
322 | + |
|
323 | + /** |
|
324 | + * Display some statistics about Finished Jobs. |
|
325 | + * |
|
326 | + * @return void |
|
327 | + */ |
|
328 | + public function stats() |
|
329 | + { |
|
330 | + $this->out('Jobs currently in the queue:'); |
|
331 | + |
|
332 | + $types = $this->QueuedTasks->getTypes()->toArray(); |
|
333 | + foreach ($types as $type) { |
|
334 | + $this->out(' ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedTasks->getLength($type)); |
|
335 | + } |
|
336 | + $this->hr(); |
|
337 | + $this->out('Total unfinished jobs: ' . $this->QueuedTasks->getLength()); |
|
338 | + $this->hr(); |
|
339 | + $this->out('Finished job statistics:'); |
|
340 | + $data = $this->QueuedTasks->getStats(); |
|
341 | + foreach ($data as $item) { |
|
342 | + $this->out(' ' . $item['task'] . ': '); |
|
343 | + $this->out(' Finished Jobs in Database: ' . $item['num']); |
|
344 | + $this->out(' Average Job existence : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's'); |
|
345 | + $this->out(' Average Execution delay : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's'); |
|
346 | + $this->out(' Average Execution time : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's'); |
|
347 | + } |
|
348 | + } |
|
349 | + |
|
350 | + /** |
|
351 | + * Get option parser method to parse commandline options |
|
352 | + * |
|
353 | + * @return \Cake\Console\ConsoleOptionParser |
|
354 | + */ |
|
355 | + public function getOptionParser() |
|
356 | + { |
|
357 | + $subcommandParser = [ |
|
358 | + 'options' => [ /* |
|
359 | 359 | * 'dry-run'=> array( |
360 | 360 | * 'short' => 'd', |
361 | 361 | * 'help' => 'Dry run the update, no jobs will actually be added.', |
362 | 362 | * 'boolean' => true |
363 | 363 | * ), |
364 | 364 | */ |
365 | - ] |
|
366 | - ]; |
|
367 | - $subcommandParserFull = $subcommandParser; |
|
368 | - $subcommandParserFull['options']['type'] = [ |
|
369 | - 'short' => 't', |
|
370 | - 'help' => 'Type (comma separated list possible)', |
|
371 | - 'default' => null |
|
372 | - ]; |
|
373 | - |
|
374 | - return parent::getOptionParser()->setDescription($this->_getDescription()) |
|
375 | - ->addSubcommand('clean', [ |
|
376 | - 'help' => 'Remove old jobs (cleanup)', |
|
377 | - 'parser' => $subcommandParser |
|
378 | - ]) |
|
379 | - ->addSubcommand('add', [ |
|
380 | - 'help' => 'Add Job', |
|
381 | - 'parser' => $subcommandParser |
|
382 | - ]) |
|
383 | - ->addSubcommand('stats', [ |
|
384 | - 'help' => 'Stats', |
|
385 | - 'parser' => $subcommandParserFull |
|
386 | - ]) |
|
387 | - ->addSubcommand('settings', [ |
|
388 | - 'help' => 'Settings', |
|
389 | - 'parser' => $subcommandParserFull |
|
390 | - ]) |
|
391 | - ->addSubcommand('runworker', [ |
|
392 | - 'help' => 'Run Worker', |
|
393 | - 'parser' => $subcommandParserFull |
|
394 | - ]); |
|
395 | - } |
|
396 | - |
|
397 | - /** |
|
398 | - * Timestamped log. |
|
399 | - * |
|
400 | - * @param string $message |
|
401 | - * Log type |
|
402 | - * @param string|null $pid |
|
403 | - * PID of the process |
|
404 | - * @param bool $addDetails |
|
405 | - * @return void |
|
406 | - */ |
|
407 | - protected function _log($message, $pid = null, $addDetails = true) |
|
408 | - { |
|
409 | - if (! Configure::read('Queue.log')) { |
|
410 | - return; |
|
411 | - } |
|
412 | - |
|
413 | - if ($addDetails) { |
|
414 | - $timeNeeded = $this->_timeNeeded(); |
|
415 | - $memoryUsage = $this->_memoryUsage(); |
|
416 | - $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']'; |
|
417 | - } |
|
418 | - |
|
419 | - if ($pid) { |
|
420 | - $message .= ' (pid ' . $pid . ')'; |
|
421 | - } |
|
422 | - Log::write('info', $message, [ |
|
423 | - 'scope' => 'queue' |
|
424 | - ]); |
|
425 | - } |
|
426 | - |
|
427 | - /** |
|
428 | - * |
|
429 | - * @param string $message |
|
430 | - * @param string|null $pid |
|
431 | - * PID of the process |
|
432 | - * @return void |
|
433 | - */ |
|
434 | - protected function _logError($message, $pid = null) |
|
435 | - { |
|
436 | - $timeNeeded = $this->_timeNeeded(); |
|
437 | - $memoryUsage = $this->_memoryUsage(); |
|
438 | - $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']'; |
|
439 | - |
|
440 | - if ($pid) { |
|
441 | - $message .= ' (pid ' . $pid . ')'; |
|
442 | - } |
|
443 | - |
|
444 | - Log::write('error', $message); |
|
445 | - } |
|
446 | - |
|
447 | - /** |
|
448 | - * Returns a List of available QueueTasks and their individual configurations. |
|
449 | - * |
|
450 | - * @return array |
|
451 | - */ |
|
452 | - protected function _getTaskConf() |
|
453 | - { |
|
454 | - if (! is_array($this->_taskConf)) { |
|
455 | - $this->_taskConf = []; |
|
456 | - foreach ($this->tasks as $task) { |
|
457 | - list ($pluginName, $taskName) = pluginSplit($task); |
|
458 | - |
|
459 | - $this->_taskConf[$taskName]['name'] = substr($taskName, 5); |
|
460 | - $this->_taskConf[$taskName]['plugin'] = $pluginName; |
|
461 | - if (property_exists($this->{$taskName}, 'timeout')) { |
|
462 | - $this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout; |
|
463 | - } else { |
|
464 | - $this->_taskConf[$taskName]['timeout'] = Config::defaultWorkerTimeout(); |
|
465 | - } |
|
466 | - if (property_exists($this->{$taskName}, 'retries')) { |
|
467 | - $this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries; |
|
468 | - } else { |
|
469 | - $this->_taskConf[$taskName]['retries'] = Config::defaultWorkerRetries(); |
|
470 | - } |
|
471 | - if (property_exists($this->{$taskName}, 'cleanupTimeout')) { |
|
472 | - $this->_taskConf[$taskName]['cleanupTimeout'] = $this->{$taskName}->cleanupTimeout; |
|
473 | - } else { |
|
474 | - $this->_taskConf[$taskName]['cleanupTimeout'] = Config::cleanupTimeout(); |
|
475 | - } |
|
476 | - } |
|
477 | - } |
|
478 | - return $this->_taskConf; |
|
479 | - } |
|
480 | - |
|
481 | - /** |
|
482 | - * Signal handling to queue worker for clean shutdown |
|
483 | - * |
|
484 | - * @param int $signal |
|
485 | - * @return void |
|
486 | - */ |
|
487 | - protected function _exit($signal) |
|
488 | - { |
|
489 | - $this->out(__d('queue', 'Caught %d signal, exiting.', $signal)); |
|
490 | - $this->_exit = true; |
|
491 | - } |
|
492 | - |
|
493 | - /** |
|
494 | - * |
|
495 | - * @return void |
|
496 | - */ |
|
497 | - protected function _displayAvailableTasks() |
|
498 | - { |
|
499 | - $this->out('Available Tasks:'); |
|
500 | - foreach ($this->taskNames as $loadedTask) { |
|
501 | - $this->out("\t" . '* ' . $this->_taskName($loadedTask)); |
|
502 | - } |
|
503 | - } |
|
504 | - |
|
505 | - /** |
|
506 | - * |
|
507 | - * @return string |
|
508 | - */ |
|
509 | - protected function _initPid() |
|
510 | - { |
|
511 | - $this->_pid = $this->_retrievePid(); |
|
512 | - return $this->_pid; |
|
513 | - } |
|
514 | - |
|
515 | - /** |
|
516 | - * |
|
517 | - * @return string |
|
518 | - */ |
|
519 | - protected function _retrievePid() |
|
520 | - { |
|
521 | - if (function_exists('posix_getpid')) { |
|
522 | - $pid = (string) posix_getpid(); |
|
523 | - } else { |
|
524 | - $pid = $this->QueuedTasks->key(); |
|
525 | - } |
|
526 | - return $pid; |
|
527 | - } |
|
528 | - |
|
529 | - /** |
|
530 | - * |
|
531 | - * @return string Memory usage in MB. |
|
532 | - */ |
|
533 | - protected function _memoryUsage() |
|
534 | - { |
|
535 | - $limit = ini_get('memory_limit'); |
|
536 | - |
|
537 | - $used = number_format(memory_get_peak_usage(true) / (1024 * 1024), 0) . 'MB'; |
|
538 | - if ($limit !== '-1') { |
|
539 | - $used .= '/' . $limit; |
|
540 | - } |
|
541 | - |
|
542 | - return $used; |
|
543 | - } |
|
544 | - |
|
545 | - /** |
|
546 | - * |
|
547 | - * @param string|null $pid |
|
548 | - * |
|
549 | - * @return void |
|
550 | - */ |
|
551 | - protected function _deletePid($pid) |
|
552 | - { |
|
553 | - if (! $pid) { |
|
554 | - $pid = $this->_pid; |
|
555 | - } |
|
556 | - if (! $pid) { |
|
557 | - return; |
|
558 | - } |
|
559 | - } |
|
560 | - |
|
561 | - /** |
|
562 | - * |
|
563 | - * @return string |
|
564 | - */ |
|
565 | - protected function _timeNeeded() |
|
566 | - { |
|
567 | - $diff = $this->_time() - $this->_time($this->_time); |
|
568 | - $seconds = max($diff, 1); |
|
569 | - |
|
570 | - return $seconds . 's'; |
|
571 | - } |
|
572 | - |
|
573 | - /** |
|
574 | - * |
|
575 | - * @param int|null $providedTime |
|
576 | - * |
|
577 | - * @return int |
|
578 | - */ |
|
579 | - protected function _time($providedTime = null) |
|
580 | - { |
|
581 | - if ($providedTime) { |
|
582 | - return $providedTime; |
|
583 | - } |
|
584 | - |
|
585 | - return time(); |
|
586 | - } |
|
587 | - |
|
588 | - /** |
|
589 | - * |
|
590 | - * @param string|null $param |
|
591 | - * @return array |
|
592 | - */ |
|
593 | - protected function _stringToArray($param) |
|
594 | - { |
|
595 | - if (! $param) { |
|
596 | - return []; |
|
597 | - } |
|
598 | - |
|
599 | - $array = Text::tokenize($param); |
|
600 | - |
|
601 | - return array_filter($array); |
|
602 | - } |
|
365 | + ] |
|
366 | + ]; |
|
367 | + $subcommandParserFull = $subcommandParser; |
|
368 | + $subcommandParserFull['options']['type'] = [ |
|
369 | + 'short' => 't', |
|
370 | + 'help' => 'Type (comma separated list possible)', |
|
371 | + 'default' => null |
|
372 | + ]; |
|
373 | + |
|
374 | + return parent::getOptionParser()->setDescription($this->_getDescription()) |
|
375 | + ->addSubcommand('clean', [ |
|
376 | + 'help' => 'Remove old jobs (cleanup)', |
|
377 | + 'parser' => $subcommandParser |
|
378 | + ]) |
|
379 | + ->addSubcommand('add', [ |
|
380 | + 'help' => 'Add Job', |
|
381 | + 'parser' => $subcommandParser |
|
382 | + ]) |
|
383 | + ->addSubcommand('stats', [ |
|
384 | + 'help' => 'Stats', |
|
385 | + 'parser' => $subcommandParserFull |
|
386 | + ]) |
|
387 | + ->addSubcommand('settings', [ |
|
388 | + 'help' => 'Settings', |
|
389 | + 'parser' => $subcommandParserFull |
|
390 | + ]) |
|
391 | + ->addSubcommand('runworker', [ |
|
392 | + 'help' => 'Run Worker', |
|
393 | + 'parser' => $subcommandParserFull |
|
394 | + ]); |
|
395 | + } |
|
396 | + |
|
397 | + /** |
|
398 | + * Timestamped log. |
|
399 | + * |
|
400 | + * @param string $message |
|
401 | + * Log type |
|
402 | + * @param string|null $pid |
|
403 | + * PID of the process |
|
404 | + * @param bool $addDetails |
|
405 | + * @return void |
|
406 | + */ |
|
407 | + protected function _log($message, $pid = null, $addDetails = true) |
|
408 | + { |
|
409 | + if (! Configure::read('Queue.log')) { |
|
410 | + return; |
|
411 | + } |
|
412 | + |
|
413 | + if ($addDetails) { |
|
414 | + $timeNeeded = $this->_timeNeeded(); |
|
415 | + $memoryUsage = $this->_memoryUsage(); |
|
416 | + $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']'; |
|
417 | + } |
|
418 | + |
|
419 | + if ($pid) { |
|
420 | + $message .= ' (pid ' . $pid . ')'; |
|
421 | + } |
|
422 | + Log::write('info', $message, [ |
|
423 | + 'scope' => 'queue' |
|
424 | + ]); |
|
425 | + } |
|
426 | + |
|
427 | + /** |
|
428 | + * |
|
429 | + * @param string $message |
|
430 | + * @param string|null $pid |
|
431 | + * PID of the process |
|
432 | + * @return void |
|
433 | + */ |
|
434 | + protected function _logError($message, $pid = null) |
|
435 | + { |
|
436 | + $timeNeeded = $this->_timeNeeded(); |
|
437 | + $memoryUsage = $this->_memoryUsage(); |
|
438 | + $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']'; |
|
439 | + |
|
440 | + if ($pid) { |
|
441 | + $message .= ' (pid ' . $pid . ')'; |
|
442 | + } |
|
443 | + |
|
444 | + Log::write('error', $message); |
|
445 | + } |
|
446 | + |
|
447 | + /** |
|
448 | + * Returns a List of available QueueTasks and their individual configurations. |
|
449 | + * |
|
450 | + * @return array |
|
451 | + */ |
|
452 | + protected function _getTaskConf() |
|
453 | + { |
|
454 | + if (! is_array($this->_taskConf)) { |
|
455 | + $this->_taskConf = []; |
|
456 | + foreach ($this->tasks as $task) { |
|
457 | + list ($pluginName, $taskName) = pluginSplit($task); |
|
458 | + |
|
459 | + $this->_taskConf[$taskName]['name'] = substr($taskName, 5); |
|
460 | + $this->_taskConf[$taskName]['plugin'] = $pluginName; |
|
461 | + if (property_exists($this->{$taskName}, 'timeout')) { |
|
462 | + $this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout; |
|
463 | + } else { |
|
464 | + $this->_taskConf[$taskName]['timeout'] = Config::defaultWorkerTimeout(); |
|
465 | + } |
|
466 | + if (property_exists($this->{$taskName}, 'retries')) { |
|
467 | + $this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries; |
|
468 | + } else { |
|
469 | + $this->_taskConf[$taskName]['retries'] = Config::defaultWorkerRetries(); |
|
470 | + } |
|
471 | + if (property_exists($this->{$taskName}, 'cleanupTimeout')) { |
|
472 | + $this->_taskConf[$taskName]['cleanupTimeout'] = $this->{$taskName}->cleanupTimeout; |
|
473 | + } else { |
|
474 | + $this->_taskConf[$taskName]['cleanupTimeout'] = Config::cleanupTimeout(); |
|
475 | + } |
|
476 | + } |
|
477 | + } |
|
478 | + return $this->_taskConf; |
|
479 | + } |
|
480 | + |
|
481 | + /** |
|
482 | + * Signal handling to queue worker for clean shutdown |
|
483 | + * |
|
484 | + * @param int $signal |
|
485 | + * @return void |
|
486 | + */ |
|
487 | + protected function _exit($signal) |
|
488 | + { |
|
489 | + $this->out(__d('queue', 'Caught %d signal, exiting.', $signal)); |
|
490 | + $this->_exit = true; |
|
491 | + } |
|
492 | + |
|
493 | + /** |
|
494 | + * |
|
495 | + * @return void |
|
496 | + */ |
|
497 | + protected function _displayAvailableTasks() |
|
498 | + { |
|
499 | + $this->out('Available Tasks:'); |
|
500 | + foreach ($this->taskNames as $loadedTask) { |
|
501 | + $this->out("\t" . '* ' . $this->_taskName($loadedTask)); |
|
502 | + } |
|
503 | + } |
|
504 | + |
|
505 | + /** |
|
506 | + * |
|
507 | + * @return string |
|
508 | + */ |
|
509 | + protected function _initPid() |
|
510 | + { |
|
511 | + $this->_pid = $this->_retrievePid(); |
|
512 | + return $this->_pid; |
|
513 | + } |
|
514 | + |
|
515 | + /** |
|
516 | + * |
|
517 | + * @return string |
|
518 | + */ |
|
519 | + protected function _retrievePid() |
|
520 | + { |
|
521 | + if (function_exists('posix_getpid')) { |
|
522 | + $pid = (string) posix_getpid(); |
|
523 | + } else { |
|
524 | + $pid = $this->QueuedTasks->key(); |
|
525 | + } |
|
526 | + return $pid; |
|
527 | + } |
|
528 | + |
|
529 | + /** |
|
530 | + * |
|
531 | + * @return string Memory usage in MB. |
|
532 | + */ |
|
533 | + protected function _memoryUsage() |
|
534 | + { |
|
535 | + $limit = ini_get('memory_limit'); |
|
536 | + |
|
537 | + $used = number_format(memory_get_peak_usage(true) / (1024 * 1024), 0) . 'MB'; |
|
538 | + if ($limit !== '-1') { |
|
539 | + $used .= '/' . $limit; |
|
540 | + } |
|
541 | + |
|
542 | + return $used; |
|
543 | + } |
|
544 | + |
|
545 | + /** |
|
546 | + * |
|
547 | + * @param string|null $pid |
|
548 | + * |
|
549 | + * @return void |
|
550 | + */ |
|
551 | + protected function _deletePid($pid) |
|
552 | + { |
|
553 | + if (! $pid) { |
|
554 | + $pid = $this->_pid; |
|
555 | + } |
|
556 | + if (! $pid) { |
|
557 | + return; |
|
558 | + } |
|
559 | + } |
|
560 | + |
|
561 | + /** |
|
562 | + * |
|
563 | + * @return string |
|
564 | + */ |
|
565 | + protected function _timeNeeded() |
|
566 | + { |
|
567 | + $diff = $this->_time() - $this->_time($this->_time); |
|
568 | + $seconds = max($diff, 1); |
|
569 | + |
|
570 | + return $seconds . 's'; |
|
571 | + } |
|
572 | + |
|
573 | + /** |
|
574 | + * |
|
575 | + * @param int|null $providedTime |
|
576 | + * |
|
577 | + * @return int |
|
578 | + */ |
|
579 | + protected function _time($providedTime = null) |
|
580 | + { |
|
581 | + if ($providedTime) { |
|
582 | + return $providedTime; |
|
583 | + } |
|
584 | + |
|
585 | + return time(); |
|
586 | + } |
|
587 | + |
|
588 | + /** |
|
589 | + * |
|
590 | + * @param string|null $param |
|
591 | + * @return array |
|
592 | + */ |
|
593 | + protected function _stringToArray($param) |
|
594 | + { |
|
595 | + if (! $param) { |
|
596 | + return []; |
|
597 | + } |
|
598 | + |
|
599 | + $array = Text::tokenize($param); |
|
600 | + |
|
601 | + return array_filter($array); |
|
602 | + } |
|
603 | 603 | } |
@@ -17,7 +17,7 @@ discard block |
||
17 | 17 | use Queue\Shell\Task\QueueTaskInterface; |
18 | 18 | use RuntimeException; |
19 | 19 | use Throwable; |
20 | -declare(ticks = 1); |
|
20 | +declare(ticks=1); |
|
21 | 21 | |
22 | 22 | /** |
23 | 23 | * Main shell to init and run queue workers. |
@@ -121,7 +121,7 @@ discard block |
||
121 | 121 | if (in_array('Queue' . $name, $this->taskNames, true)) { |
122 | 122 | /** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */ |
123 | 123 | $task = $this->{'Queue' . $name}; |
124 | - if (! ($task instanceof AddInterface)) { |
|
124 | + if (!($task instanceof AddInterface)) { |
|
125 | 125 | $this->abort('This task does not support adding via CLI call'); |
126 | 126 | } |
127 | 127 | $task->add(); |
@@ -160,7 +160,7 @@ discard block |
||
160 | 160 | $pid = $this->_initPid(); |
161 | 161 | } catch (PersistenceFailedException $exception) { |
162 | 162 | $this->err($exception->getMessage()); |
163 | - $limit = (int) Configure::read('Queue.maxWorkers'); |
|
163 | + $limit = (int)Configure::read('Queue.maxWorkers'); |
|
164 | 164 | if ($limit) { |
165 | 165 | $this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')'); |
166 | 166 | } |
@@ -194,7 +194,7 @@ discard block |
||
194 | 194 | $startTime = time(); |
195 | 195 | $types = $this->_stringToArray($this->param('type')); |
196 | 196 | |
197 | - while (! $this->_exit) { |
|
197 | + while (!$this->_exit) { |
|
198 | 198 | $this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE); |
199 | 199 | |
200 | 200 | $QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types); |
@@ -214,7 +214,7 @@ discard block |
||
214 | 214 | $this->_exit = true; |
215 | 215 | $this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.'); |
216 | 216 | } |
217 | - if ($this->_exit || mt_rand(0, 100) > (100 - (int) Config::gcprob())) { |
|
217 | + if ($this->_exit || mt_rand(0, 100) > (100 - (int)Config::gcprob())) { |
|
218 | 218 | $this->out(__d('queue', 'Performing old job cleanup.')); |
219 | 219 | $this->QueuedTasks->cleanOldJobs(); |
220 | 220 | } |
@@ -246,11 +246,11 @@ discard block |
||
246 | 246 | $data = unserialize($QueuedTask->data); |
247 | 247 | /** @var \Queue\Shell\Task\QueueTask $task */ |
248 | 248 | $task = $this->{$taskName}; |
249 | - if (! $task instanceof QueueTaskInterface) { |
|
249 | + if (!$task instanceof QueueTaskInterface) { |
|
250 | 250 | throw new RuntimeException('Task must implement ' . QueueTaskInterface::class); |
251 | 251 | } |
252 | 252 | |
253 | - $return = $task->run((array) $data, $QueuedTask->id); |
|
253 | + $return = $task->run((array)$data, $QueuedTask->id); |
|
254 | 254 | if ($return !== null) { |
255 | 255 | trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED); |
256 | 256 | } |
@@ -259,7 +259,7 @@ discard block |
||
259 | 259 | $return = false; |
260 | 260 | |
261 | 261 | $failureMessage = get_class($e) . ': ' . $e->getMessage(); |
262 | - if (! ($e instanceof QueueException)) { |
|
262 | + if (!($e instanceof QueueException)) { |
|
263 | 263 | $failureMessage .= "\n" . $e->getTraceAsString(); |
264 | 264 | } |
265 | 265 | |
@@ -290,11 +290,11 @@ discard block |
||
290 | 290 | */ |
291 | 291 | public function clean() |
292 | 292 | { |
293 | - if (! Configure::read('Queue.cleanupTimeout')) { |
|
293 | + if (!Configure::read('Queue.cleanupTimeout')) { |
|
294 | 294 | $this->abort('You disabled cleanuptimout in config. Aborting.'); |
295 | 295 | } |
296 | 296 | |
297 | - $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int) Configure::read('Queue.cleanupTimeout'))); |
|
297 | + $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int)Configure::read('Queue.cleanupTimeout'))); |
|
298 | 298 | $this->QueuedTasks->cleanOldJobs(); |
299 | 299 | } |
300 | 300 | |
@@ -306,7 +306,7 @@ discard block |
||
306 | 306 | public function settings() |
307 | 307 | { |
308 | 308 | $this->out('Current Settings:'); |
309 | - $conf = (array) Configure::read('Queue'); |
|
309 | + $conf = (array)Configure::read('Queue'); |
|
310 | 310 | foreach ($conf as $key => $val) { |
311 | 311 | if ($val === false) { |
312 | 312 | $val = 'no'; |
@@ -355,7 +355,7 @@ discard block |
||
355 | 355 | public function getOptionParser() |
356 | 356 | { |
357 | 357 | $subcommandParser = [ |
358 | - 'options' => [ /* |
|
358 | + 'options' => [/* |
|
359 | 359 | * 'dry-run'=> array( |
360 | 360 | * 'short' => 'd', |
361 | 361 | * 'help' => 'Dry run the update, no jobs will actually be added.', |
@@ -406,7 +406,7 @@ discard block |
||
406 | 406 | */ |
407 | 407 | protected function _log($message, $pid = null, $addDetails = true) |
408 | 408 | { |
409 | - if (! Configure::read('Queue.log')) { |
|
409 | + if (!Configure::read('Queue.log')) { |
|
410 | 410 | return; |
411 | 411 | } |
412 | 412 | |
@@ -451,7 +451,7 @@ discard block |
||
451 | 451 | */ |
452 | 452 | protected function _getTaskConf() |
453 | 453 | { |
454 | - if (! is_array($this->_taskConf)) { |
|
454 | + if (!is_array($this->_taskConf)) { |
|
455 | 455 | $this->_taskConf = []; |
456 | 456 | foreach ($this->tasks as $task) { |
457 | 457 | list ($pluginName, $taskName) = pluginSplit($task); |
@@ -519,7 +519,7 @@ discard block |
||
519 | 519 | protected function _retrievePid() |
520 | 520 | { |
521 | 521 | if (function_exists('posix_getpid')) { |
522 | - $pid = (string) posix_getpid(); |
|
522 | + $pid = (string)posix_getpid(); |
|
523 | 523 | } else { |
524 | 524 | $pid = $this->QueuedTasks->key(); |
525 | 525 | } |
@@ -550,10 +550,10 @@ discard block |
||
550 | 550 | */ |
551 | 551 | protected function _deletePid($pid) |
552 | 552 | { |
553 | - if (! $pid) { |
|
553 | + if (!$pid) { |
|
554 | 554 | $pid = $this->_pid; |
555 | 555 | } |
556 | - if (! $pid) { |
|
556 | + if (!$pid) { |
|
557 | 557 | return; |
558 | 558 | } |
559 | 559 | } |
@@ -592,7 +592,7 @@ discard block |
||
592 | 592 | */ |
593 | 593 | protected function _stringToArray($param) |
594 | 594 | { |
595 | - if (! $param) { |
|
595 | + if (!$param) { |
|
596 | 596 | return []; |
597 | 597 | } |
598 | 598 |