@@ -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 | } |
@@ -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 | } |
@@ -20,21 +20,21 @@ |
||
20 | 20 | class QueuedTask extends Entity |
21 | 21 | { |
22 | 22 | |
23 | - /** |
|
24 | - * |
|
25 | - * {@inheritdoc} |
|
26 | - * |
|
27 | - * @var array |
|
28 | - */ |
|
29 | - protected $_accessible = [ |
|
30 | - 'task' => true, |
|
31 | - 'data' => true, |
|
32 | - 'not_before' => true, |
|
33 | - 'fetched' => true, |
|
34 | - 'completed' => true, |
|
35 | - 'failed_count' => true, |
|
36 | - 'failure_message' => true, |
|
37 | - 'worker_key' => true, |
|
38 | - 'created' => true |
|
39 | - ]; |
|
23 | + /** |
|
24 | + * |
|
25 | + * {@inheritdoc} |
|
26 | + * |
|
27 | + * @var array |
|
28 | + */ |
|
29 | + protected $_accessible = [ |
|
30 | + 'task' => true, |
|
31 | + 'data' => true, |
|
32 | + 'not_before' => true, |
|
33 | + 'fetched' => true, |
|
34 | + 'completed' => true, |
|
35 | + 'failed_count' => true, |
|
36 | + 'failure_message' => true, |
|
37 | + 'worker_key' => true, |
|
38 | + 'created' => true |
|
39 | + ]; |
|
40 | 40 | } |
@@ -1,18 +1,18 @@ |
||
1 | 1 | <?php |
2 | 2 | |
3 | 3 | return [ |
4 | - 'Queue' => [ |
|
5 | - // time (in seconds) after which a job is requeued if the worker doesn't report back |
|
6 | - 'defaultWorkerTimeout' => 1800, |
|
4 | + 'Queue' => [ |
|
5 | + // time (in seconds) after which a job is requeued if the worker doesn't report back |
|
6 | + 'defaultWorkerTimeout' => 1800, |
|
7 | 7 | |
8 | - // seconds of running time after which the worker will terminate (0 = unlimited) |
|
9 | - 'workerMaxRuntime' => 120, |
|
8 | + // seconds of running time after which the worker will terminate (0 = unlimited) |
|
9 | + 'workerMaxRuntime' => 120, |
|
10 | 10 | |
11 | - // minimum time (in seconds) which a task remains in the database before being cleaned up. |
|
12 | - 'cleanupTimeout' => 2592000, // 30 days |
|
11 | + // minimum time (in seconds) which a task remains in the database before being cleaned up. |
|
12 | + 'cleanupTimeout' => 2592000, // 30 days |
|
13 | 13 | |
14 | - /* Optional */ |
|
14 | + /* Optional */ |
|
15 | 15 | |
16 | - 'isSearchEnabled' => true, |
|
17 | - ], |
|
16 | + 'isSearchEnabled' => true, |
|
17 | + ], |
|
18 | 18 | ]; |
@@ -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 The directory |
|
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 The directory |
|
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 The directory |
|
69 | - * @param string $plugin The plugin name |
|
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 The directory |
|
69 | + * @param string $plugin The plugin name |
|
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 | } |
@@ -18,57 +18,57 @@ |
||
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 IO |
|
50 | - */ |
|
51 | - public function __construct(ConsoleIo $io = null) |
|
52 | - { |
|
53 | - parent::__construct($io); |
|
47 | + /** |
|
48 | + * |
|
49 | + * @param \Cake\Console\ConsoleIo|null $io IO |
|
50 | + */ |
|
51 | + public function __construct(ConsoleIo $io = null) |
|
52 | + { |
|
53 | + parent::__construct($io); |
|
54 | 54 | |
55 | - $this->loadModel($this->queueModelClass); |
|
56 | - } |
|
55 | + $this->loadModel($this->queueModelClass); |
|
56 | + } |
|
57 | 57 | |
58 | - /** |
|
59 | - * |
|
60 | - * @return string |
|
61 | - * @throws \InvalidArgumentException |
|
62 | - */ |
|
63 | - protected function queueTaskName() |
|
64 | - { |
|
65 | - $class = get_class($this); |
|
58 | + /** |
|
59 | + * |
|
60 | + * @return string |
|
61 | + * @throws \InvalidArgumentException |
|
62 | + */ |
|
63 | + protected function queueTaskName() |
|
64 | + { |
|
65 | + $class = get_class($this); |
|
66 | 66 | |
67 | - preg_match('#\\\\Queue(.+)Task$#', $class, $matches); |
|
68 | - if (!$matches) { |
|
69 | - throw new InvalidArgumentException('Invalid class name: ' . $class); |
|
70 | - } |
|
67 | + preg_match('#\\\\Queue(.+)Task$#', $class, $matches); |
|
68 | + if (!$matches) { |
|
69 | + throw new InvalidArgumentException('Invalid class name: ' . $class); |
|
70 | + } |
|
71 | 71 | |
72 | - return $matches[1]; |
|
73 | - } |
|
72 | + return $matches[1]; |
|
73 | + } |
|
74 | 74 | } |
@@ -7,82 +7,82 @@ |
||
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 The array passed to QueuedTasksTable::createJob() |
|
76 | - * @param int $taskId The id of the QueuedTask entity |
|
77 | - * @return void |
|
78 | - */ |
|
79 | - public function run(array $data, $taskId): void |
|
80 | - { |
|
81 | - $this->hr(); |
|
82 | - $this->out(__d('queue', 'CakePHP Queue Example task.')); |
|
83 | - $this->hr(); |
|
84 | - $this->out(__d('queue', ' ->Success, the Example Task was run.<-')); |
|
85 | - $this->out(' '); |
|
86 | - $this->out(' '); |
|
87 | - } |
|
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 The array passed to QueuedTasksTable::createJob() |
|
76 | + * @param int $taskId The id of the QueuedTask entity |
|
77 | + * @return void |
|
78 | + */ |
|
79 | + public function run(array $data, $taskId): void |
|
80 | + { |
|
81 | + $this->hr(); |
|
82 | + $this->out(__d('queue', 'CakePHP Queue Example task.')); |
|
83 | + $this->hr(); |
|
84 | + $this->out(__d('queue', ' ->Success, the Example Task was run.<-')); |
|
85 | + $this->out(' '); |
|
86 | + $this->out(' '); |
|
87 | + } |
|
88 | 88 | } |
@@ -13,12 +13,12 @@ |
||
13 | 13 | interface QueueTaskInterface |
14 | 14 | { |
15 | 15 | |
16 | - /** |
|
17 | - * Main execution of the task. |
|
18 | - * |
|
19 | - * @param array $data The array passed to QueuedTasksTable::createJob() |
|
20 | - * @param int $taskId The id of the QueuedTask entity |
|
21 | - * @return void |
|
22 | - */ |
|
23 | - public function run(array $data, $taskId): void; |
|
16 | + /** |
|
17 | + * Main execution of the task. |
|
18 | + * |
|
19 | + * @param array $data The array passed to QueuedTasksTable::createJob() |
|
20 | + * @param int $taskId The id of the QueuedTask entity |
|
21 | + * @return void |
|
22 | + */ |
|
23 | + public function run(array $data, $taskId): void; |
|
24 | 24 | } |
@@ -189,7 +189,7 @@ discard block |
||
189 | 189 | { |
190 | 190 | $driverName = $this->_getDriverName(); |
191 | 191 | $options = [ |
192 | - 'fields' => function (Query $query) use ($driverName) { |
|
192 | + 'fields' => function(Query $query) use ($driverName) { |
|
193 | 193 | $alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)'); |
194 | 194 | $runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)'); |
195 | 195 | $fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(not_before is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(not_before))'); |
@@ -236,7 +236,7 @@ discard block |
||
236 | 236 | public function getFullStats($taskName = null) |
237 | 237 | { |
238 | 238 | $driverName = $this->_getDriverName(); |
239 | - $fields = function (Query $query) use ($driverName) { |
|
239 | + $fields = function(Query $query) use ($driverName) { |
|
240 | 240 | $runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)'); |
241 | 241 | switch ($driverName) { |
242 | 242 | case static::DRIVER_SQLSERVER: |
@@ -370,7 +370,7 @@ discard block |
||
370 | 370 | } |
371 | 371 | |
372 | 372 | /** @var \Queue\Model\Entity\QueuedTask|null $task */ |
373 | - $task = $this->getConnection()->transactional(function () use ($query, $options, $now) { |
|
373 | + $task = $this->getConnection()->transactional(function() use ($query, $options, $now) { |
|
374 | 374 | $task = $query->find('all', $options) |
375 | 375 | ->enableAutoFields(true) |
376 | 376 | ->epilog('FOR UPDATE') |
@@ -28,596 +28,596 @@ |
||
28 | 28 | class QueuedTasksTable extends Table |
29 | 29 | { |
30 | 30 | |
31 | - const DRIVER_MYSQL = 'Mysql'; |
|
32 | - |
|
33 | - const DRIVER_POSTGRES = 'Postgres'; |
|
34 | - |
|
35 | - const DRIVER_SQLSERVER = 'Sqlserver'; |
|
36 | - |
|
37 | - const STATS_LIMIT = 100000; |
|
38 | - |
|
39 | - /** |
|
40 | - * |
|
41 | - * @var string|null |
|
42 | - */ |
|
43 | - protected $_key; |
|
44 | - |
|
45 | - /** |
|
46 | - * Initialize method |
|
47 | - * |
|
48 | - * @param array $config The configuration for the Table. |
|
49 | - * @return void |
|
50 | - */ |
|
51 | - public function initialize(array $config) |
|
52 | - { |
|
53 | - parent::initialize($config); |
|
54 | - |
|
55 | - $this->setTable('queued_tasks'); |
|
56 | - $this->setDisplayField('id'); |
|
57 | - $this->setPrimaryKey('id'); |
|
58 | - |
|
59 | - $this->addBehavior('Timestamp'); |
|
60 | - } |
|
61 | - |
|
62 | - /** |
|
63 | - * set connection name |
|
64 | - * |
|
65 | - * @return string |
|
66 | - */ |
|
67 | - public static function defaultConnectionName() |
|
68 | - { |
|
69 | - $connection = Configure::read('Queue.connection'); |
|
70 | - if (!empty($connection)) { |
|
71 | - return $connection; |
|
72 | - } |
|
73 | - |
|
74 | - return parent::defaultConnectionName(); |
|
75 | - } |
|
76 | - |
|
77 | - /** |
|
78 | - * |
|
79 | - * @param \Cake\Event\Event $event Model event |
|
80 | - * @param \ArrayObject $data The data |
|
81 | - * @param \ArrayObject $options The options |
|
82 | - * @return void |
|
83 | - */ |
|
84 | - public function beforeMarshal(Event $event, ArrayObject $data, ArrayObject $options) |
|
85 | - { |
|
86 | - if (isset($data['data']) && $data['data'] === '') { |
|
87 | - $data['data'] = null; |
|
88 | - } |
|
89 | - } |
|
90 | - |
|
91 | - /** |
|
92 | - * Adds a new job to the queue. |
|
93 | - * |
|
94 | - * @param string $taskName Task name |
|
95 | - * @param array|null $data Array of data |
|
96 | - * @param string $notBefore A datetime which indicates when the job may be executed |
|
97 | - * @return \Queue\Model\Entity\QueuedTask Saved job entity |
|
98 | - */ |
|
99 | - public function createJob($taskName, array $data = null, string $notBefore = null) |
|
100 | - { |
|
101 | - $task = [ |
|
102 | - 'task' => $taskName, |
|
103 | - 'data' => serialize($data), |
|
104 | - 'not_before' => $this->getDateTime() |
|
105 | - ]; |
|
106 | - |
|
107 | - if (!empty($notBefore)) { |
|
108 | - $task['not_before'] = $this->getDateTime(strtotime($notBefore)); |
|
109 | - } |
|
110 | - |
|
111 | - $queuedTask = $this->newEntity($task); |
|
112 | - var_dump($task); |
|
113 | - var_dump($queuedTask); |
|
114 | - |
|
115 | - return $this->saveOrFail($queuedTask); |
|
116 | - } |
|
117 | - |
|
118 | - /** |
|
119 | - * |
|
120 | - * @param string|null $taskName Task name |
|
121 | - * @return bool |
|
122 | - */ |
|
123 | - public function isQueued($taskName = null) |
|
124 | - { |
|
125 | - $conditions = [ |
|
126 | - 'completed IS' => null |
|
127 | - ]; |
|
128 | - if ($taskName) { |
|
129 | - $conditions['task'] = $taskName; |
|
130 | - } |
|
131 | - |
|
132 | - return (bool)$this->find() |
|
133 | - ->where($conditions) |
|
134 | - ->select([ |
|
135 | - 'id' |
|
136 | - ]) |
|
137 | - ->first(); |
|
138 | - } |
|
139 | - |
|
140 | - /** |
|
141 | - * Returns the number of items in the queue. |
|
142 | - * Either returns the number of ALL pending jobs, or the number of pending jobs of the passed type. |
|
143 | - * |
|
144 | - * @param string|null $taskName Task type to Count |
|
145 | - * @return int |
|
146 | - */ |
|
147 | - public function getLength($taskName = null) |
|
148 | - { |
|
149 | - $findConf = [ |
|
150 | - 'conditions' => [ |
|
151 | - 'completed IS' => null |
|
152 | - ] |
|
153 | - ]; |
|
154 | - if ($taskName !== null) { |
|
155 | - $findConf['conditions']['task'] = $taskName; |
|
156 | - } |
|
157 | - |
|
158 | - return $this->find('all', $findConf)->count(); |
|
159 | - } |
|
160 | - |
|
161 | - /** |
|
162 | - * Return a list of all task types in the Queue. |
|
163 | - * |
|
164 | - * @return \Cake\ORM\Query |
|
165 | - */ |
|
166 | - public function getTypes() |
|
167 | - { |
|
168 | - $findCond = [ |
|
169 | - 'fields' => [ |
|
170 | - 'task' |
|
171 | - ], |
|
172 | - 'group' => [ |
|
173 | - 'task' |
|
174 | - ], |
|
175 | - 'keyField' => 'task', |
|
176 | - 'valueField' => 'task' |
|
177 | - ]; |
|
178 | - |
|
179 | - return $this->find('list', $findCond); |
|
180 | - } |
|
181 | - |
|
182 | - /** |
|
183 | - * Return some statistics about finished jobs still in the Database. |
|
184 | - * TO-DO: rewrite as virtual field |
|
185 | - * |
|
186 | - * @return \Cake\ORM\Query |
|
187 | - */ |
|
188 | - public function getStats() |
|
189 | - { |
|
190 | - $driverName = $this->_getDriverName(); |
|
191 | - $options = [ |
|
192 | - 'fields' => function (Query $query) use ($driverName) { |
|
193 | - $alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)'); |
|
194 | - $runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)'); |
|
195 | - $fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(not_before is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(not_before))'); |
|
196 | - switch ($driverName) { |
|
197 | - case static::DRIVER_SQLSERVER: |
|
198 | - $alltime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', created)"); |
|
199 | - $runtime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)"); |
|
200 | - $fetchdelay = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', fetched) - (CASE WHEN not_before IS NULL THEN DATEDIFF(s, '1970-01-01 00:00:00', created) ELSE DATEDIFF(s, '1970-01-01 00:00:00', not_before) END)"); |
|
201 | - break; |
|
202 | - } |
|
203 | - /** |
|
204 | - * |
|
205 | - * @var \Cake\ORM\Query |
|
206 | - */ |
|
207 | - return [ |
|
208 | - 'task', |
|
209 | - 'num' => $query->func()->count('*'), |
|
210 | - 'alltime' => $alltime, |
|
211 | - 'runtime' => $runtime, |
|
212 | - 'fetchdelay' => $fetchdelay |
|
213 | - ]; |
|
214 | - }, |
|
215 | - 'conditions' => [ |
|
216 | - 'completed IS NOT' => null |
|
217 | - ], |
|
218 | - 'group' => [ |
|
219 | - 'task' |
|
220 | - ] |
|
221 | - ]; |
|
222 | - |
|
223 | - return $this->find('all', $options); |
|
224 | - } |
|
225 | - |
|
226 | - /** |
|
227 | - * Returns [ |
|
228 | - * 'Task' => [ |
|
229 | - * 'YYYY-MM-DD' => INT, |
|
230 | - * ... |
|
231 | - * ] |
|
232 | - * ] |
|
233 | - * |
|
234 | - * @param string|null $taskName The task name |
|
235 | - * @return array |
|
236 | - */ |
|
237 | - public function getFullStats($taskName = null) |
|
238 | - { |
|
239 | - $driverName = $this->_getDriverName(); |
|
240 | - $fields = function (Query $query) use ($driverName) { |
|
241 | - $runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)'); |
|
242 | - switch ($driverName) { |
|
243 | - case static::DRIVER_SQLSERVER: |
|
244 | - $runtime = $query->newExpr("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)"); |
|
245 | - break; |
|
246 | - } |
|
247 | - |
|
248 | - return [ |
|
249 | - 'task', |
|
250 | - 'created', |
|
251 | - 'duration' => $runtime |
|
252 | - ]; |
|
253 | - }; |
|
254 | - |
|
255 | - $conditions = [ |
|
256 | - 'completed IS NOT' => null |
|
257 | - ]; |
|
258 | - if ($taskName) { |
|
259 | - $conditions['task'] = $taskName; |
|
260 | - } |
|
261 | - |
|
262 | - $tasks = $this->find() |
|
263 | - ->select($fields) |
|
264 | - ->where($conditions) |
|
265 | - ->enableHydration(false) |
|
266 | - ->orderDesc('id') |
|
267 | - ->limit(static::STATS_LIMIT) |
|
268 | - ->all() |
|
269 | - ->toArray(); |
|
270 | - |
|
271 | - $result = []; |
|
272 | - |
|
273 | - $days = []; |
|
274 | - |
|
275 | - foreach ($tasks as $task) { |
|
276 | - /** @var \DateTime $created */ |
|
277 | - $created = $task['created']; |
|
278 | - $day = $created->format('Y-m-d'); |
|
279 | - if (!isset($days[$day])) { |
|
280 | - $days[$day] = $day; |
|
281 | - } |
|
282 | - |
|
283 | - $result[$task['task']][$day][] = $task['duration']; |
|
284 | - } |
|
285 | - |
|
286 | - foreach ($result as $type => $tasks) { |
|
287 | - foreach ($tasks as $day => $durations) { |
|
288 | - $average = array_sum($durations) / count($durations); |
|
289 | - $result[$type][$day] = (int)$average; |
|
290 | - } |
|
291 | - |
|
292 | - foreach ($days as $day) { |
|
293 | - if (isset($result[$type][$day])) { |
|
294 | - continue; |
|
295 | - } |
|
296 | - |
|
297 | - $result[$type][$day] = 0; |
|
298 | - } |
|
299 | - |
|
300 | - ksort($result[$type]); |
|
301 | - } |
|
302 | - |
|
303 | - return $result; |
|
304 | - } |
|
305 | - |
|
306 | - /** |
|
307 | - * Look for a new job that can be processed with the current abilities and |
|
308 | - * from the specified group (or any if null). |
|
309 | - * |
|
310 | - * @param array $capabilities Available QueueWorkerTasks. |
|
311 | - * @param array $types Request a job from these types (or exclude certain types), or any otherwise. |
|
312 | - * @return \Queue\Model\Entity\QueuedTask|null |
|
313 | - */ |
|
314 | - public function requestJob(array $capabilities, array $types = []) |
|
315 | - { |
|
316 | - $now = $this->getDateTime(); |
|
317 | - $nowStr = $now->toDateTimeString(); |
|
318 | - $driverName = $this->_getDriverName(); |
|
319 | - |
|
320 | - $query = $this->find(); |
|
321 | - $age = $query->newExpr()->add('IFNULL(TIMESTAMPDIFF(SECOND, "' . $nowStr . '", not_before), 0)'); |
|
322 | - switch ($driverName) { |
|
323 | - case static::DRIVER_SQLSERVER: |
|
324 | - $age = $query->newExpr()->add('ISNULL(DATEDIFF(SECOND, GETDATE(), not_before), 0)'); |
|
325 | - break; |
|
326 | - case static::DRIVER_POSTGRES: |
|
327 | - $age = $query->newExpr()->add('COALESCE((EXTRACT(EPOCH FROM now()) - EXTRACT(EPOCH FROM not_before)), 0)'); |
|
328 | - break; |
|
329 | - } |
|
330 | - $options = [ |
|
331 | - 'conditions' => [ |
|
332 | - 'completed IS' => null, |
|
333 | - 'OR' => [] |
|
334 | - ], |
|
335 | - 'fields' => [ |
|
336 | - 'age' => $age |
|
337 | - ], |
|
338 | - 'order' => [ |
|
339 | - 'age' => 'ASC', |
|
340 | - 'id' => 'ASC' |
|
341 | - ] |
|
342 | - ]; |
|
343 | - |
|
344 | - if ($types) { |
|
345 | - $options['conditions'] = $this->addFilter($options['conditions'], 'task', $types); |
|
346 | - } |
|
347 | - |
|
348 | - // Generate the task specific conditions. |
|
349 | - foreach ($capabilities as $task) { |
|
350 | - list ($plugin, $name) = pluginSplit($task['name']); |
|
351 | - $timeoutAt = $now->copy(); |
|
352 | - $tmp = [ |
|
353 | - 'task' => $name, |
|
354 | - 'AND' => [ |
|
355 | - [ |
|
356 | - 'OR' => [ |
|
357 | - 'not_before <' => $nowStr, |
|
358 | - 'not_before IS' => null |
|
359 | - ] |
|
360 | - ], |
|
361 | - [ |
|
362 | - 'OR' => [ |
|
363 | - 'fetched <' => $timeoutAt->subSeconds($task['timeout']), |
|
364 | - 'fetched IS' => null |
|
365 | - ] |
|
366 | - ] |
|
367 | - ], |
|
368 | - 'failed_count <' => ($task['retries'] + 1) |
|
369 | - ]; |
|
370 | - $options['conditions']['OR'][] = $tmp; |
|
371 | - } |
|
372 | - |
|
373 | - /** @var \Queue\Model\Entity\QueuedTask|null $task */ |
|
374 | - $task = $this->getConnection()->transactional(function () use ($query, $options, $now) { |
|
375 | - $task = $query->find('all', $options) |
|
376 | - ->enableAutoFields(true) |
|
377 | - ->epilog('FOR UPDATE') |
|
378 | - ->first(); |
|
379 | - |
|
380 | - if (!$task) { |
|
381 | - return null; |
|
382 | - } |
|
383 | - |
|
384 | - $key = $this->key(); |
|
385 | - $task = $this->patchEntity($task, [ |
|
386 | - 'worker_key' => $key, |
|
387 | - 'fetched' => $now |
|
388 | - ]); |
|
389 | - |
|
390 | - return $this->saveOrFail($task); |
|
391 | - }); |
|
392 | - |
|
393 | - if (!$task) { |
|
394 | - return null; |
|
395 | - } |
|
396 | - |
|
397 | - return $task; |
|
398 | - } |
|
399 | - |
|
400 | - /** |
|
401 | - * Mark a task as Completed, removing it from the queue. |
|
402 | - * |
|
403 | - * @param \Queue\Model\Entity\QueuedTask $task Task |
|
404 | - * @return bool Success |
|
405 | - */ |
|
406 | - public function markJobDone(QueuedTask $task) |
|
407 | - { |
|
408 | - $fields = [ |
|
409 | - 'completed' => $this->getDateTime() |
|
410 | - ]; |
|
411 | - $task = $this->patchEntity($task, $fields); |
|
412 | - |
|
413 | - return (bool)$this->save($task); |
|
414 | - } |
|
415 | - |
|
416 | - /** |
|
417 | - * Mark a job as Failed, incrementing the failed-counter and Requeueing it. |
|
418 | - * |
|
419 | - * @param \Queue\Model\Entity\QueuedTask $task Task |
|
420 | - * @param string|null $failureMessage Optional message to append to the failure_message field. |
|
421 | - * @return bool Success |
|
422 | - */ |
|
423 | - public function markJobFailed(QueuedTask $task, $failureMessage = null) |
|
424 | - { |
|
425 | - $fields = [ |
|
426 | - 'failed_count' => $task->failed_count + 1, |
|
427 | - 'failure_message' => $failureMessage |
|
428 | - ]; |
|
429 | - $task = $this->patchEntity($task, $fields); |
|
430 | - |
|
431 | - return (bool)$this->save($task); |
|
432 | - } |
|
433 | - |
|
434 | - /** |
|
435 | - * Reset current jobs |
|
436 | - * |
|
437 | - * @param int|null $id ID |
|
438 | - * |
|
439 | - * @return int Success |
|
440 | - */ |
|
441 | - public function reset($id = null) |
|
442 | - { |
|
443 | - $fields = [ |
|
444 | - 'completed' => null, |
|
445 | - 'fetched' => null, |
|
446 | - 'failed_count' => 0, |
|
447 | - 'worker_key' => null, |
|
448 | - 'failure_message' => null |
|
449 | - ]; |
|
450 | - $conditions = [ |
|
451 | - 'completed IS' => null |
|
452 | - ]; |
|
453 | - if ($id) { |
|
454 | - $conditions['id'] = $id; |
|
455 | - } |
|
456 | - |
|
457 | - return $this->updateAll($fields, $conditions); |
|
458 | - } |
|
459 | - |
|
460 | - /** |
|
461 | - * |
|
462 | - * @param string $taskName Task name |
|
463 | - * |
|
464 | - * @return int |
|
465 | - */ |
|
466 | - public function rerun($taskName) |
|
467 | - { |
|
468 | - $fields = [ |
|
469 | - 'completed' => null, |
|
470 | - 'fetched' => null, |
|
471 | - 'failed_count' => 0, |
|
472 | - 'worker_key' => null, |
|
473 | - 'failure_message' => null |
|
474 | - ]; |
|
475 | - $conditions = [ |
|
476 | - 'completed IS NOT' => null, |
|
477 | - 'task' => $taskName |
|
478 | - ]; |
|
479 | - |
|
480 | - return $this->updateAll($fields, $conditions); |
|
481 | - } |
|
482 | - |
|
483 | - /** |
|
484 | - * Cleanup/Delete Completed Tasks. |
|
485 | - * |
|
486 | - * @return void |
|
487 | - */ |
|
488 | - public function cleanOldJobs() |
|
489 | - { |
|
490 | - if (!Configure::read('Queue.cleanuptimeout')) { |
|
491 | - return; |
|
492 | - } |
|
493 | - |
|
494 | - $this->deleteAll([ |
|
495 | - 'completed <' => time() - (int)Configure::read('Queue.cleanuptimeout') |
|
496 | - ]); |
|
497 | - } |
|
498 | - |
|
499 | - /** |
|
500 | - * |
|
501 | - * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task |
|
502 | - * @param array $taskConfiguration Task configuration |
|
503 | - * @return string |
|
504 | - */ |
|
505 | - public function getFailedStatus($queuedTask, array $taskConfiguration) |
|
506 | - { |
|
507 | - $failureMessageRequeued = 'requeued'; |
|
508 | - |
|
509 | - $queuedTaskName = 'Queue' . $queuedTask->task; |
|
510 | - if (empty($taskConfiguration[$queuedTaskName])) { |
|
511 | - return $failureMessageRequeued; |
|
512 | - } |
|
513 | - $retries = $taskConfiguration[$queuedTaskName]['retries']; |
|
514 | - if ($queuedTask->failed_count <= $retries) { |
|
515 | - return $failureMessageRequeued; |
|
516 | - } |
|
517 | - |
|
518 | - return 'aborted'; |
|
519 | - } |
|
520 | - |
|
521 | - /** |
|
522 | - * Generates a unique Identifier for the current worker thread. |
|
523 | - * |
|
524 | - * Useful to identify the currently running processes for this thread. |
|
525 | - * |
|
526 | - * @return string Identifier |
|
527 | - */ |
|
528 | - public function key() |
|
529 | - { |
|
530 | - if ($this->_key !== null) { |
|
531 | - return $this->_key; |
|
532 | - } |
|
533 | - $this->_key = sha1(microtime()); |
|
534 | - if (!$this->_key) { |
|
535 | - throw new RuntimeException('Invalid key generated'); |
|
536 | - } |
|
537 | - |
|
538 | - return $this->_key; |
|
539 | - } |
|
540 | - |
|
541 | - /** |
|
542 | - * Resets worker Identifier |
|
543 | - * |
|
544 | - * @return void |
|
545 | - */ |
|
546 | - public function clearKey() |
|
547 | - { |
|
548 | - $this->_key = null; |
|
549 | - } |
|
550 | - |
|
551 | - /** |
|
552 | - * truncate() |
|
553 | - * |
|
554 | - * @return void |
|
555 | - */ |
|
556 | - public function truncate() |
|
557 | - { |
|
558 | - $sql = $this->getSchema()->truncateSql($this->_connection); |
|
559 | - foreach ($sql as $snippet) { |
|
560 | - $this->_connection->execute($snippet); |
|
561 | - } |
|
562 | - } |
|
563 | - |
|
564 | - /** |
|
565 | - * get the name of the driver |
|
566 | - * |
|
567 | - * @return string |
|
568 | - */ |
|
569 | - protected function _getDriverName() |
|
570 | - { |
|
571 | - $className = explode('\\', $this->getConnection()->config()['driver']); |
|
572 | - $name = end($className); |
|
573 | - |
|
574 | - return $name; |
|
575 | - } |
|
576 | - |
|
577 | - /** |
|
578 | - * |
|
579 | - * @param array $conditions Conditions |
|
580 | - * @param string $key Key |
|
581 | - * @param array $values Values |
|
582 | - * @return array |
|
583 | - */ |
|
584 | - protected function addFilter(array $conditions, $key, array $values) |
|
585 | - { |
|
586 | - $include = []; |
|
587 | - $exclude = []; |
|
588 | - foreach ($values as $value) { |
|
589 | - if (substr($value, 0, 1) === '-') { |
|
590 | - $exclude[] = substr($value, 1); |
|
591 | - } else { |
|
592 | - $include[] = $value; |
|
593 | - } |
|
594 | - } |
|
595 | - |
|
596 | - if ($include) { |
|
597 | - $conditions[$key . ' IN'] = $include; |
|
598 | - } |
|
599 | - if ($exclude) { |
|
600 | - $conditions[$key . ' NOT IN'] = $exclude; |
|
601 | - } |
|
602 | - |
|
603 | - return $conditions; |
|
604 | - } |
|
605 | - |
|
606 | - /** |
|
607 | - * Returns a DateTime object from different input. |
|
608 | - * |
|
609 | - * Without argument this will be "now". |
|
610 | - * |
|
611 | - * @param int|string|\Cake\I18n\FrozenTime|\Cake\I18n\Time|null $notBefore Not before time |
|
612 | - * |
|
613 | - * @return \Cake\I18n\FrozenTime|\Cake\I18n\Time |
|
614 | - */ |
|
615 | - protected function getDateTime($notBefore = null) |
|
616 | - { |
|
617 | - if (is_object($notBefore)) { |
|
618 | - return $notBefore; |
|
619 | - } |
|
620 | - |
|
621 | - return new FrozenTime($notBefore); |
|
622 | - } |
|
31 | + const DRIVER_MYSQL = 'Mysql'; |
|
32 | + |
|
33 | + const DRIVER_POSTGRES = 'Postgres'; |
|
34 | + |
|
35 | + const DRIVER_SQLSERVER = 'Sqlserver'; |
|
36 | + |
|
37 | + const STATS_LIMIT = 100000; |
|
38 | + |
|
39 | + /** |
|
40 | + * |
|
41 | + * @var string|null |
|
42 | + */ |
|
43 | + protected $_key; |
|
44 | + |
|
45 | + /** |
|
46 | + * Initialize method |
|
47 | + * |
|
48 | + * @param array $config The configuration for the Table. |
|
49 | + * @return void |
|
50 | + */ |
|
51 | + public function initialize(array $config) |
|
52 | + { |
|
53 | + parent::initialize($config); |
|
54 | + |
|
55 | + $this->setTable('queued_tasks'); |
|
56 | + $this->setDisplayField('id'); |
|
57 | + $this->setPrimaryKey('id'); |
|
58 | + |
|
59 | + $this->addBehavior('Timestamp'); |
|
60 | + } |
|
61 | + |
|
62 | + /** |
|
63 | + * set connection name |
|
64 | + * |
|
65 | + * @return string |
|
66 | + */ |
|
67 | + public static function defaultConnectionName() |
|
68 | + { |
|
69 | + $connection = Configure::read('Queue.connection'); |
|
70 | + if (!empty($connection)) { |
|
71 | + return $connection; |
|
72 | + } |
|
73 | + |
|
74 | + return parent::defaultConnectionName(); |
|
75 | + } |
|
76 | + |
|
77 | + /** |
|
78 | + * |
|
79 | + * @param \Cake\Event\Event $event Model event |
|
80 | + * @param \ArrayObject $data The data |
|
81 | + * @param \ArrayObject $options The options |
|
82 | + * @return void |
|
83 | + */ |
|
84 | + public function beforeMarshal(Event $event, ArrayObject $data, ArrayObject $options) |
|
85 | + { |
|
86 | + if (isset($data['data']) && $data['data'] === '') { |
|
87 | + $data['data'] = null; |
|
88 | + } |
|
89 | + } |
|
90 | + |
|
91 | + /** |
|
92 | + * Adds a new job to the queue. |
|
93 | + * |
|
94 | + * @param string $taskName Task name |
|
95 | + * @param array|null $data Array of data |
|
96 | + * @param string $notBefore A datetime which indicates when the job may be executed |
|
97 | + * @return \Queue\Model\Entity\QueuedTask Saved job entity |
|
98 | + */ |
|
99 | + public function createJob($taskName, array $data = null, string $notBefore = null) |
|
100 | + { |
|
101 | + $task = [ |
|
102 | + 'task' => $taskName, |
|
103 | + 'data' => serialize($data), |
|
104 | + 'not_before' => $this->getDateTime() |
|
105 | + ]; |
|
106 | + |
|
107 | + if (!empty($notBefore)) { |
|
108 | + $task['not_before'] = $this->getDateTime(strtotime($notBefore)); |
|
109 | + } |
|
110 | + |
|
111 | + $queuedTask = $this->newEntity($task); |
|
112 | + var_dump($task); |
|
113 | + var_dump($queuedTask); |
|
114 | + |
|
115 | + return $this->saveOrFail($queuedTask); |
|
116 | + } |
|
117 | + |
|
118 | + /** |
|
119 | + * |
|
120 | + * @param string|null $taskName Task name |
|
121 | + * @return bool |
|
122 | + */ |
|
123 | + public function isQueued($taskName = null) |
|
124 | + { |
|
125 | + $conditions = [ |
|
126 | + 'completed IS' => null |
|
127 | + ]; |
|
128 | + if ($taskName) { |
|
129 | + $conditions['task'] = $taskName; |
|
130 | + } |
|
131 | + |
|
132 | + return (bool)$this->find() |
|
133 | + ->where($conditions) |
|
134 | + ->select([ |
|
135 | + 'id' |
|
136 | + ]) |
|
137 | + ->first(); |
|
138 | + } |
|
139 | + |
|
140 | + /** |
|
141 | + * Returns the number of items in the queue. |
|
142 | + * Either returns the number of ALL pending jobs, or the number of pending jobs of the passed type. |
|
143 | + * |
|
144 | + * @param string|null $taskName Task type to Count |
|
145 | + * @return int |
|
146 | + */ |
|
147 | + public function getLength($taskName = null) |
|
148 | + { |
|
149 | + $findConf = [ |
|
150 | + 'conditions' => [ |
|
151 | + 'completed IS' => null |
|
152 | + ] |
|
153 | + ]; |
|
154 | + if ($taskName !== null) { |
|
155 | + $findConf['conditions']['task'] = $taskName; |
|
156 | + } |
|
157 | + |
|
158 | + return $this->find('all', $findConf)->count(); |
|
159 | + } |
|
160 | + |
|
161 | + /** |
|
162 | + * Return a list of all task types in the Queue. |
|
163 | + * |
|
164 | + * @return \Cake\ORM\Query |
|
165 | + */ |
|
166 | + public function getTypes() |
|
167 | + { |
|
168 | + $findCond = [ |
|
169 | + 'fields' => [ |
|
170 | + 'task' |
|
171 | + ], |
|
172 | + 'group' => [ |
|
173 | + 'task' |
|
174 | + ], |
|
175 | + 'keyField' => 'task', |
|
176 | + 'valueField' => 'task' |
|
177 | + ]; |
|
178 | + |
|
179 | + return $this->find('list', $findCond); |
|
180 | + } |
|
181 | + |
|
182 | + /** |
|
183 | + * Return some statistics about finished jobs still in the Database. |
|
184 | + * TO-DO: rewrite as virtual field |
|
185 | + * |
|
186 | + * @return \Cake\ORM\Query |
|
187 | + */ |
|
188 | + public function getStats() |
|
189 | + { |
|
190 | + $driverName = $this->_getDriverName(); |
|
191 | + $options = [ |
|
192 | + 'fields' => function (Query $query) use ($driverName) { |
|
193 | + $alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)'); |
|
194 | + $runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)'); |
|
195 | + $fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(not_before is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(not_before))'); |
|
196 | + switch ($driverName) { |
|
197 | + case static::DRIVER_SQLSERVER: |
|
198 | + $alltime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', created)"); |
|
199 | + $runtime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)"); |
|
200 | + $fetchdelay = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', fetched) - (CASE WHEN not_before IS NULL THEN DATEDIFF(s, '1970-01-01 00:00:00', created) ELSE DATEDIFF(s, '1970-01-01 00:00:00', not_before) END)"); |
|
201 | + break; |
|
202 | + } |
|
203 | + /** |
|
204 | + * |
|
205 | + * @var \Cake\ORM\Query |
|
206 | + */ |
|
207 | + return [ |
|
208 | + 'task', |
|
209 | + 'num' => $query->func()->count('*'), |
|
210 | + 'alltime' => $alltime, |
|
211 | + 'runtime' => $runtime, |
|
212 | + 'fetchdelay' => $fetchdelay |
|
213 | + ]; |
|
214 | + }, |
|
215 | + 'conditions' => [ |
|
216 | + 'completed IS NOT' => null |
|
217 | + ], |
|
218 | + 'group' => [ |
|
219 | + 'task' |
|
220 | + ] |
|
221 | + ]; |
|
222 | + |
|
223 | + return $this->find('all', $options); |
|
224 | + } |
|
225 | + |
|
226 | + /** |
|
227 | + * Returns [ |
|
228 | + * 'Task' => [ |
|
229 | + * 'YYYY-MM-DD' => INT, |
|
230 | + * ... |
|
231 | + * ] |
|
232 | + * ] |
|
233 | + * |
|
234 | + * @param string|null $taskName The task name |
|
235 | + * @return array |
|
236 | + */ |
|
237 | + public function getFullStats($taskName = null) |
|
238 | + { |
|
239 | + $driverName = $this->_getDriverName(); |
|
240 | + $fields = function (Query $query) use ($driverName) { |
|
241 | + $runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)'); |
|
242 | + switch ($driverName) { |
|
243 | + case static::DRIVER_SQLSERVER: |
|
244 | + $runtime = $query->newExpr("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)"); |
|
245 | + break; |
|
246 | + } |
|
247 | + |
|
248 | + return [ |
|
249 | + 'task', |
|
250 | + 'created', |
|
251 | + 'duration' => $runtime |
|
252 | + ]; |
|
253 | + }; |
|
254 | + |
|
255 | + $conditions = [ |
|
256 | + 'completed IS NOT' => null |
|
257 | + ]; |
|
258 | + if ($taskName) { |
|
259 | + $conditions['task'] = $taskName; |
|
260 | + } |
|
261 | + |
|
262 | + $tasks = $this->find() |
|
263 | + ->select($fields) |
|
264 | + ->where($conditions) |
|
265 | + ->enableHydration(false) |
|
266 | + ->orderDesc('id') |
|
267 | + ->limit(static::STATS_LIMIT) |
|
268 | + ->all() |
|
269 | + ->toArray(); |
|
270 | + |
|
271 | + $result = []; |
|
272 | + |
|
273 | + $days = []; |
|
274 | + |
|
275 | + foreach ($tasks as $task) { |
|
276 | + /** @var \DateTime $created */ |
|
277 | + $created = $task['created']; |
|
278 | + $day = $created->format('Y-m-d'); |
|
279 | + if (!isset($days[$day])) { |
|
280 | + $days[$day] = $day; |
|
281 | + } |
|
282 | + |
|
283 | + $result[$task['task']][$day][] = $task['duration']; |
|
284 | + } |
|
285 | + |
|
286 | + foreach ($result as $type => $tasks) { |
|
287 | + foreach ($tasks as $day => $durations) { |
|
288 | + $average = array_sum($durations) / count($durations); |
|
289 | + $result[$type][$day] = (int)$average; |
|
290 | + } |
|
291 | + |
|
292 | + foreach ($days as $day) { |
|
293 | + if (isset($result[$type][$day])) { |
|
294 | + continue; |
|
295 | + } |
|
296 | + |
|
297 | + $result[$type][$day] = 0; |
|
298 | + } |
|
299 | + |
|
300 | + ksort($result[$type]); |
|
301 | + } |
|
302 | + |
|
303 | + return $result; |
|
304 | + } |
|
305 | + |
|
306 | + /** |
|
307 | + * Look for a new job that can be processed with the current abilities and |
|
308 | + * from the specified group (or any if null). |
|
309 | + * |
|
310 | + * @param array $capabilities Available QueueWorkerTasks. |
|
311 | + * @param array $types Request a job from these types (or exclude certain types), or any otherwise. |
|
312 | + * @return \Queue\Model\Entity\QueuedTask|null |
|
313 | + */ |
|
314 | + public function requestJob(array $capabilities, array $types = []) |
|
315 | + { |
|
316 | + $now = $this->getDateTime(); |
|
317 | + $nowStr = $now->toDateTimeString(); |
|
318 | + $driverName = $this->_getDriverName(); |
|
319 | + |
|
320 | + $query = $this->find(); |
|
321 | + $age = $query->newExpr()->add('IFNULL(TIMESTAMPDIFF(SECOND, "' . $nowStr . '", not_before), 0)'); |
|
322 | + switch ($driverName) { |
|
323 | + case static::DRIVER_SQLSERVER: |
|
324 | + $age = $query->newExpr()->add('ISNULL(DATEDIFF(SECOND, GETDATE(), not_before), 0)'); |
|
325 | + break; |
|
326 | + case static::DRIVER_POSTGRES: |
|
327 | + $age = $query->newExpr()->add('COALESCE((EXTRACT(EPOCH FROM now()) - EXTRACT(EPOCH FROM not_before)), 0)'); |
|
328 | + break; |
|
329 | + } |
|
330 | + $options = [ |
|
331 | + 'conditions' => [ |
|
332 | + 'completed IS' => null, |
|
333 | + 'OR' => [] |
|
334 | + ], |
|
335 | + 'fields' => [ |
|
336 | + 'age' => $age |
|
337 | + ], |
|
338 | + 'order' => [ |
|
339 | + 'age' => 'ASC', |
|
340 | + 'id' => 'ASC' |
|
341 | + ] |
|
342 | + ]; |
|
343 | + |
|
344 | + if ($types) { |
|
345 | + $options['conditions'] = $this->addFilter($options['conditions'], 'task', $types); |
|
346 | + } |
|
347 | + |
|
348 | + // Generate the task specific conditions. |
|
349 | + foreach ($capabilities as $task) { |
|
350 | + list ($plugin, $name) = pluginSplit($task['name']); |
|
351 | + $timeoutAt = $now->copy(); |
|
352 | + $tmp = [ |
|
353 | + 'task' => $name, |
|
354 | + 'AND' => [ |
|
355 | + [ |
|
356 | + 'OR' => [ |
|
357 | + 'not_before <' => $nowStr, |
|
358 | + 'not_before IS' => null |
|
359 | + ] |
|
360 | + ], |
|
361 | + [ |
|
362 | + 'OR' => [ |
|
363 | + 'fetched <' => $timeoutAt->subSeconds($task['timeout']), |
|
364 | + 'fetched IS' => null |
|
365 | + ] |
|
366 | + ] |
|
367 | + ], |
|
368 | + 'failed_count <' => ($task['retries'] + 1) |
|
369 | + ]; |
|
370 | + $options['conditions']['OR'][] = $tmp; |
|
371 | + } |
|
372 | + |
|
373 | + /** @var \Queue\Model\Entity\QueuedTask|null $task */ |
|
374 | + $task = $this->getConnection()->transactional(function () use ($query, $options, $now) { |
|
375 | + $task = $query->find('all', $options) |
|
376 | + ->enableAutoFields(true) |
|
377 | + ->epilog('FOR UPDATE') |
|
378 | + ->first(); |
|
379 | + |
|
380 | + if (!$task) { |
|
381 | + return null; |
|
382 | + } |
|
383 | + |
|
384 | + $key = $this->key(); |
|
385 | + $task = $this->patchEntity($task, [ |
|
386 | + 'worker_key' => $key, |
|
387 | + 'fetched' => $now |
|
388 | + ]); |
|
389 | + |
|
390 | + return $this->saveOrFail($task); |
|
391 | + }); |
|
392 | + |
|
393 | + if (!$task) { |
|
394 | + return null; |
|
395 | + } |
|
396 | + |
|
397 | + return $task; |
|
398 | + } |
|
399 | + |
|
400 | + /** |
|
401 | + * Mark a task as Completed, removing it from the queue. |
|
402 | + * |
|
403 | + * @param \Queue\Model\Entity\QueuedTask $task Task |
|
404 | + * @return bool Success |
|
405 | + */ |
|
406 | + public function markJobDone(QueuedTask $task) |
|
407 | + { |
|
408 | + $fields = [ |
|
409 | + 'completed' => $this->getDateTime() |
|
410 | + ]; |
|
411 | + $task = $this->patchEntity($task, $fields); |
|
412 | + |
|
413 | + return (bool)$this->save($task); |
|
414 | + } |
|
415 | + |
|
416 | + /** |
|
417 | + * Mark a job as Failed, incrementing the failed-counter and Requeueing it. |
|
418 | + * |
|
419 | + * @param \Queue\Model\Entity\QueuedTask $task Task |
|
420 | + * @param string|null $failureMessage Optional message to append to the failure_message field. |
|
421 | + * @return bool Success |
|
422 | + */ |
|
423 | + public function markJobFailed(QueuedTask $task, $failureMessage = null) |
|
424 | + { |
|
425 | + $fields = [ |
|
426 | + 'failed_count' => $task->failed_count + 1, |
|
427 | + 'failure_message' => $failureMessage |
|
428 | + ]; |
|
429 | + $task = $this->patchEntity($task, $fields); |
|
430 | + |
|
431 | + return (bool)$this->save($task); |
|
432 | + } |
|
433 | + |
|
434 | + /** |
|
435 | + * Reset current jobs |
|
436 | + * |
|
437 | + * @param int|null $id ID |
|
438 | + * |
|
439 | + * @return int Success |
|
440 | + */ |
|
441 | + public function reset($id = null) |
|
442 | + { |
|
443 | + $fields = [ |
|
444 | + 'completed' => null, |
|
445 | + 'fetched' => null, |
|
446 | + 'failed_count' => 0, |
|
447 | + 'worker_key' => null, |
|
448 | + 'failure_message' => null |
|
449 | + ]; |
|
450 | + $conditions = [ |
|
451 | + 'completed IS' => null |
|
452 | + ]; |
|
453 | + if ($id) { |
|
454 | + $conditions['id'] = $id; |
|
455 | + } |
|
456 | + |
|
457 | + return $this->updateAll($fields, $conditions); |
|
458 | + } |
|
459 | + |
|
460 | + /** |
|
461 | + * |
|
462 | + * @param string $taskName Task name |
|
463 | + * |
|
464 | + * @return int |
|
465 | + */ |
|
466 | + public function rerun($taskName) |
|
467 | + { |
|
468 | + $fields = [ |
|
469 | + 'completed' => null, |
|
470 | + 'fetched' => null, |
|
471 | + 'failed_count' => 0, |
|
472 | + 'worker_key' => null, |
|
473 | + 'failure_message' => null |
|
474 | + ]; |
|
475 | + $conditions = [ |
|
476 | + 'completed IS NOT' => null, |
|
477 | + 'task' => $taskName |
|
478 | + ]; |
|
479 | + |
|
480 | + return $this->updateAll($fields, $conditions); |
|
481 | + } |
|
482 | + |
|
483 | + /** |
|
484 | + * Cleanup/Delete Completed Tasks. |
|
485 | + * |
|
486 | + * @return void |
|
487 | + */ |
|
488 | + public function cleanOldJobs() |
|
489 | + { |
|
490 | + if (!Configure::read('Queue.cleanuptimeout')) { |
|
491 | + return; |
|
492 | + } |
|
493 | + |
|
494 | + $this->deleteAll([ |
|
495 | + 'completed <' => time() - (int)Configure::read('Queue.cleanuptimeout') |
|
496 | + ]); |
|
497 | + } |
|
498 | + |
|
499 | + /** |
|
500 | + * |
|
501 | + * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task |
|
502 | + * @param array $taskConfiguration Task configuration |
|
503 | + * @return string |
|
504 | + */ |
|
505 | + public function getFailedStatus($queuedTask, array $taskConfiguration) |
|
506 | + { |
|
507 | + $failureMessageRequeued = 'requeued'; |
|
508 | + |
|
509 | + $queuedTaskName = 'Queue' . $queuedTask->task; |
|
510 | + if (empty($taskConfiguration[$queuedTaskName])) { |
|
511 | + return $failureMessageRequeued; |
|
512 | + } |
|
513 | + $retries = $taskConfiguration[$queuedTaskName]['retries']; |
|
514 | + if ($queuedTask->failed_count <= $retries) { |
|
515 | + return $failureMessageRequeued; |
|
516 | + } |
|
517 | + |
|
518 | + return 'aborted'; |
|
519 | + } |
|
520 | + |
|
521 | + /** |
|
522 | + * Generates a unique Identifier for the current worker thread. |
|
523 | + * |
|
524 | + * Useful to identify the currently running processes for this thread. |
|
525 | + * |
|
526 | + * @return string Identifier |
|
527 | + */ |
|
528 | + public function key() |
|
529 | + { |
|
530 | + if ($this->_key !== null) { |
|
531 | + return $this->_key; |
|
532 | + } |
|
533 | + $this->_key = sha1(microtime()); |
|
534 | + if (!$this->_key) { |
|
535 | + throw new RuntimeException('Invalid key generated'); |
|
536 | + } |
|
537 | + |
|
538 | + return $this->_key; |
|
539 | + } |
|
540 | + |
|
541 | + /** |
|
542 | + * Resets worker Identifier |
|
543 | + * |
|
544 | + * @return void |
|
545 | + */ |
|
546 | + public function clearKey() |
|
547 | + { |
|
548 | + $this->_key = null; |
|
549 | + } |
|
550 | + |
|
551 | + /** |
|
552 | + * truncate() |
|
553 | + * |
|
554 | + * @return void |
|
555 | + */ |
|
556 | + public function truncate() |
|
557 | + { |
|
558 | + $sql = $this->getSchema()->truncateSql($this->_connection); |
|
559 | + foreach ($sql as $snippet) { |
|
560 | + $this->_connection->execute($snippet); |
|
561 | + } |
|
562 | + } |
|
563 | + |
|
564 | + /** |
|
565 | + * get the name of the driver |
|
566 | + * |
|
567 | + * @return string |
|
568 | + */ |
|
569 | + protected function _getDriverName() |
|
570 | + { |
|
571 | + $className = explode('\\', $this->getConnection()->config()['driver']); |
|
572 | + $name = end($className); |
|
573 | + |
|
574 | + return $name; |
|
575 | + } |
|
576 | + |
|
577 | + /** |
|
578 | + * |
|
579 | + * @param array $conditions Conditions |
|
580 | + * @param string $key Key |
|
581 | + * @param array $values Values |
|
582 | + * @return array |
|
583 | + */ |
|
584 | + protected function addFilter(array $conditions, $key, array $values) |
|
585 | + { |
|
586 | + $include = []; |
|
587 | + $exclude = []; |
|
588 | + foreach ($values as $value) { |
|
589 | + if (substr($value, 0, 1) === '-') { |
|
590 | + $exclude[] = substr($value, 1); |
|
591 | + } else { |
|
592 | + $include[] = $value; |
|
593 | + } |
|
594 | + } |
|
595 | + |
|
596 | + if ($include) { |
|
597 | + $conditions[$key . ' IN'] = $include; |
|
598 | + } |
|
599 | + if ($exclude) { |
|
600 | + $conditions[$key . ' NOT IN'] = $exclude; |
|
601 | + } |
|
602 | + |
|
603 | + return $conditions; |
|
604 | + } |
|
605 | + |
|
606 | + /** |
|
607 | + * Returns a DateTime object from different input. |
|
608 | + * |
|
609 | + * Without argument this will be "now". |
|
610 | + * |
|
611 | + * @param int|string|\Cake\I18n\FrozenTime|\Cake\I18n\Time|null $notBefore Not before time |
|
612 | + * |
|
613 | + * @return \Cake\I18n\FrozenTime|\Cake\I18n\Time |
|
614 | + */ |
|
615 | + protected function getDateTime($notBefore = null) |
|
616 | + { |
|
617 | + if (is_object($notBefore)) { |
|
618 | + return $notBefore; |
|
619 | + } |
|
620 | + |
|
621 | + return new FrozenTime($notBefore); |
|
622 | + } |
|
623 | 623 | } |