Completed
Pull Request — master (#35)
by
unknown
03:01
created
src/Queue/Config.php 1 patch
Indentation   +56 added lines, -56 removed lines patch added patch discarded remove patch
@@ -6,66 +6,66 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.
src/Shell/Task/AddInterface.php 1 patch
Indentation   +8 added lines, -8 removed lines patch added patch discarded remove patch
@@ -10,12 +10,12 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.
src/Model/Entity/QueuedTask.php 1 patch
Indentation   +17 added lines, -17 removed lines patch added patch discarded remove patch
@@ -20,21 +20,21 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.
tests/config/app_queue.php 1 patch
Indentation   +10 added lines, -10 removed lines patch added patch discarded remove patch
@@ -1,18 +1,18 @@
 block discarded – undo
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
 ];
Please login to merge, or discard this patch.
src/Queue/TaskFinder.php 1 patch
Indentation   +68 added lines, -68 removed lines patch added patch discarded remove patch
@@ -8,80 +8,80 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.
src/Shell/Task/QueueTask.php 1 patch
Indentation   +45 added lines, -45 removed lines patch added patch discarded remove patch
@@ -18,57 +18,57 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.
src/Shell/Task/QueueExampleTask.php 1 patch
Indentation   +72 added lines, -72 removed lines patch added patch discarded remove patch
@@ -7,82 +7,82 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.
src/Shell/Task/QueueTaskInterface.php 1 patch
Indentation   +8 added lines, -8 removed lines patch added patch discarded remove patch
@@ -13,12 +13,12 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.
src/Model/Table/QueuedTasksTable.php 2 patches
Spacing   +3 added lines, -3 removed lines patch added patch discarded remove patch
@@ -189,7 +189,7 @@  discard block
 block discarded – undo
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
 block discarded – undo
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
 block discarded – undo
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')
Please login to merge, or discard this patch.
Indentation   +592 added lines, -592 removed lines patch added patch discarded remove patch
@@ -28,596 +28,596 @@
 block discarded – undo
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
 }
Please login to merge, or discard this patch.