Completed
Pull Request — master (#35)
by
unknown
03:38
created
tests/Fixture/QueuedTasksFixture.php 1 patch
Indentation   +41 added lines, -41 removed lines patch added patch discarded remove patch
@@ -9,46 +9,46 @@
 block discarded – undo
9 9
 class QueuedTasksFixture extends TestFixture
10 10
 {
11 11
 
12
-    /**
13
-     * Fields
14
-     *
15
-     * @var array
16
-     */
17
-    // @codingStandardsIgnoreStart
18
-    public $fields = [
19
-        'id' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => null, 'comment' => '', 'autoIncrement' => true, 'precision' => null],
20
-        'task' => ['type' => 'string', 'length' => 255, 'null' => false, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null, 'fixed' => null],
21
-        'data' => ['type' => 'text', 'length' => 16777215, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
22
-        'not_before' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
23
-        'fetched' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
24
-        'completed' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
25
-        'failed_count' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => '0', 'comment' => '', 'precision' => null, 'autoIncrement' => null],
26
-        'failure_message' => ['type' => 'text', 'length' => null, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
27
-        'worker_key' => ['type' => 'string', 'fixed' => true, 'length' => 40, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
28
-        'created' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
29
-        '_indexes' => [
30
-            'completed' => ['type' => 'index', 'columns' => ['completed'], 'length' => []],
31
-            'worker_key' => ['type' => 'index', 'columns' => ['worker_key'], 'length' => []],
32
-            'task' => ['type' => 'index', 'columns' => ['task'], 'length' => []],
33
-        ],
34
-        '_constraints' => [
35
-            'primary' => ['type' => 'primary', 'columns' => ['id'], 'length' => []],
36
-        ],
37
-        '_options' => [
38
-            'engine' => 'InnoDB',
39
-            'collation' => 'utf8_general_ci'
40
-        ],
41
-    ];
42
-    // @codingStandardsIgnoreEnd
12
+	/**
13
+	 * Fields
14
+	 *
15
+	 * @var array
16
+	 */
17
+	// @codingStandardsIgnoreStart
18
+	public $fields = [
19
+		'id' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => null, 'comment' => '', 'autoIncrement' => true, 'precision' => null],
20
+		'task' => ['type' => 'string', 'length' => 255, 'null' => false, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null, 'fixed' => null],
21
+		'data' => ['type' => 'text', 'length' => 16777215, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
22
+		'not_before' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
23
+		'fetched' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
24
+		'completed' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
25
+		'failed_count' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => '0', 'comment' => '', 'precision' => null, 'autoIncrement' => null],
26
+		'failure_message' => ['type' => 'text', 'length' => null, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
27
+		'worker_key' => ['type' => 'string', 'fixed' => true, 'length' => 40, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
28
+		'created' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
29
+		'_indexes' => [
30
+			'completed' => ['type' => 'index', 'columns' => ['completed'], 'length' => []],
31
+			'worker_key' => ['type' => 'index', 'columns' => ['worker_key'], 'length' => []],
32
+			'task' => ['type' => 'index', 'columns' => ['task'], 'length' => []],
33
+		],
34
+		'_constraints' => [
35
+			'primary' => ['type' => 'primary', 'columns' => ['id'], 'length' => []],
36
+		],
37
+		'_options' => [
38
+			'engine' => 'InnoDB',
39
+			'collation' => 'utf8_general_ci'
40
+		],
41
+	];
42
+	// @codingStandardsIgnoreEnd
43 43
 
44
-    /**
45
-     * Init method
46
-     *
47
-     * @return void
48
-     */
49
-    public function init()
50
-    {
51
-        $this->records = [];
52
-        parent::init();
53
-    }
44
+	/**
45
+	 * Init method
46
+	 *
47
+	 * @return void
48
+	 */
49
+	public function init()
50
+	{
51
+		$this->records = [];
52
+		parent::init();
53
+	}
54 54
 }
Please login to merge, or discard this patch.
src/Shell/QueueShell.php 1 patch
Indentation   +562 added lines, -562 removed lines patch added patch discarded remove patch
@@ -31,578 +31,578 @@
 block discarded – undo
31 31
 class QueueShell extends Shell
32 32
 {
33 33
 
34
-    /**
35
-     *
36
-     * @var string
37
-     */
38
-    public $modelClass = 'Queue.QueuedTasks';
39
-
40
-    /**
41
-     *
42
-     * @var array|null
43
-     */
44
-    protected $_taskConf;
45
-
46
-    /**
47
-     *
48
-     * @var int
49
-     */
50
-    protected $_time = 0;
51
-
52
-    /**
53
-     *
54
-     * @var bool
55
-     */
56
-    protected $_exit = false;
57
-
58
-    /**
59
-     * Overwrite shell initialize to dynamically load all Queue Related Tasks.
60
-     *
61
-     * @return void
62
-     */
63
-    public function initialize()
64
-    {
65
-        $taskFinder = new TaskFinder();
66
-        $this->tasks = $taskFinder->allAppAndPluginTasks();
67
-
68
-        parent::initialize();
69
-    }
70
-
71
-    /**
72
-     *
73
-     * @return void
74
-     */
75
-    public function startup()
76
-    {
77
-        if ($this->param('quiet')) {
78
-            $this->interactive = false;
79
-        }
80
-
81
-        parent::startup();
82
-    }
83
-
84
-    /**
85
-     *
86
-     * @return string
87
-     */
88
-    public function getDescription()
89
-    {
90
-        $tasks = [];
91
-        foreach ($this->taskNames as $loadedTask) {
92
-            $tasks[] = "\t" . '* ' . $this->_taskName($loadedTask);
93
-        }
94
-        $tasks = implode(PHP_EOL, $tasks);
95
-
96
-        $text = <<<TEXT
34
+	/**
35
+	 *
36
+	 * @var string
37
+	 */
38
+	public $modelClass = 'Queue.QueuedTasks';
39
+
40
+	/**
41
+	 *
42
+	 * @var array|null
43
+	 */
44
+	protected $_taskConf;
45
+
46
+	/**
47
+	 *
48
+	 * @var int
49
+	 */
50
+	protected $_time = 0;
51
+
52
+	/**
53
+	 *
54
+	 * @var bool
55
+	 */
56
+	protected $_exit = false;
57
+
58
+	/**
59
+	 * Overwrite shell initialize to dynamically load all Queue Related Tasks.
60
+	 *
61
+	 * @return void
62
+	 */
63
+	public function initialize()
64
+	{
65
+		$taskFinder = new TaskFinder();
66
+		$this->tasks = $taskFinder->allAppAndPluginTasks();
67
+
68
+		parent::initialize();
69
+	}
70
+
71
+	/**
72
+	 *
73
+	 * @return void
74
+	 */
75
+	public function startup()
76
+	{
77
+		if ($this->param('quiet')) {
78
+			$this->interactive = false;
79
+		}
80
+
81
+		parent::startup();
82
+	}
83
+
84
+	/**
85
+	 *
86
+	 * @return string
87
+	 */
88
+	public function getDescription()
89
+	{
90
+		$tasks = [];
91
+		foreach ($this->taskNames as $loadedTask) {
92
+			$tasks[] = "\t" . '* ' . $this->_taskName($loadedTask);
93
+		}
94
+		$tasks = implode(PHP_EOL, $tasks);
95
+
96
+		$text = <<<TEXT
97 97
 Simple and minimalistic job queue (or deferred-task) system.
98 98
 
99 99
 Available Tasks:
100 100
 $tasks
101 101
 TEXT;
102 102
 
103
-        return $text;
104
-    }
105
-
106
-    /**
107
-     * Look for a Queue Task of hte passed name and try to call add() on it.
108
-     * A QueueTask may provide an add function to enable the user to create new jobs via commandline.
109
-     *
110
-     * @return void
111
-     */
112
-    public function add()
113
-    {
114
-        if (count($this->args) < 1) {
115
-            $this->out('Please call like this:');
116
-            $this->out('       bin/cake queue add <taskname>');
117
-            $this->_displayAvailableTasks();
118
-
119
-            return;
120
-        }
121
-
122
-        $name = Inflector::camelize($this->args[0]);
123
-        if (in_array('Queue' . $name, $this->taskNames, true)) {
124
-            /** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */
125
-            $task = $this->{'Queue' . $name};
126
-            if (!($task instanceof AddInterface)) {
127
-                $this->abort('This task does not support adding via CLI call');
128
-            }
129
-            $task->add();
130
-        } else {
131
-            $this->out('Error: Task not found: ' . $name);
132
-            $this->_displayAvailableTasks();
133
-        }
134
-    }
135
-
136
-    /**
137
-     * Output the task without Queue or Task
138
-     * example: QueueImageTask becomes Image on display
139
-     *
140
-     * @param string $task Task name
141
-     * @return string Cleaned task name
142
-     */
143
-    protected function _taskName($task)
144
-    {
145
-        if (strpos($task, 'Queue') === 0) {
146
-            return substr($task, 5);
147
-        }
148
-
149
-        return $task;
150
-    }
151
-
152
-    /**
153
-     * Run a QueueWorker loop.
154
-     * Runs a Queue Worker process which will try to find unassigned jobs in the queue
155
-     * which it may run and try to fetch and execute them.
156
-     *
157
-     * @return int|null
158
-     */
159
-    public function runworker()
160
-    {
161
-        try {
162
-            $pid = $this->_initPid();
163
-        } catch (PersistenceFailedException $exception) {
164
-            $this->err($exception->getMessage());
165
-            $limit = (int)Configure::read('Queue.maxWorkers');
166
-            if ($limit) {
167
-                $this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')');
168
-            }
169
-
170
-            return static::CODE_ERROR;
171
-        }
172
-
173
-        // Enable Garbage Collector (PHP >= 5.3)
174
-        if (function_exists('gc_enable')) {
175
-            gc_enable();
176
-        }
177
-        if (function_exists('pcntl_signal')) {
178
-            pcntl_signal(SIGTERM, [
179
-                &$this,
180
-                '_exit'
181
-            ]);
182
-            pcntl_signal(SIGINT, [
183
-                &$this,
184
-                '_exit'
185
-            ]);
186
-            pcntl_signal(SIGTSTP, [
187
-                &$this,
188
-                '_exit'
189
-            ]);
190
-            pcntl_signal(SIGQUIT, [
191
-                &$this,
192
-                '_exit'
193
-            ]);
194
-        }
195
-        $this->_exit = false;
196
-
197
-        $startTime = time();
198
-        $types = $this->_stringToArray($this->param('type'));
199
-
200
-        while (!$this->_exit) {
201
-            $this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE);
202
-
203
-            $QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types);
204
-
205
-            if ($QueuedTask) {
206
-                $this->runJob($QueuedTask, $pid);
207
-            } elseif (Configure::read('Queue.exitWhenNothingToDo')) {
208
-                $this->out('nothing to do, exiting.');
209
-                $this->_exit = true;
210
-            } else {
211
-                $this->out('nothing to do, sleeping.');
212
-                sleep(Config::sleepTime());
213
-            }
214
-
215
-            // check if we are over the maximum runtime and end processing if so.
216
-            if (Configure::readOrFail('Queue.workerMaxRuntime') && (time() - $startTime) >= Configure::readOrFail('Queue.workerMaxRuntime')) {
217
-                $this->_exit = true;
218
-                $this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.');
219
-            }
220
-            if ($this->_exit || mt_rand(0, 100) > (100 - (int)Config::gcprob())) {
221
-                $this->out(__d('queue', 'Performing old job cleanup.'));
222
-                $this->QueuedTasks->cleanOldJobs();
223
-            }
224
-            $this->hr();
225
-        }
226
-
227
-        $this->_deletePid($pid);
228
-
229
-        if ($this->param('verbose')) {
230
-            $this->_log('endworker', $pid);
231
-        }
232
-    }
233
-
234
-    /**
235
-     *
236
-     * @param \Queue\Model\Entity\QueuedTask $QueuedTask Queued task
237
-     * @param string $pid PID of the process
238
-     * @return void
239
-     */
240
-    protected function runJob(QueuedTask $QueuedTask, $pid)
241
-    {
242
-        $this->out('Running Job of type "' . $QueuedTask->task . '"');
243
-        $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id, $pid, false);
244
-        $taskName = 'Queue' . $QueuedTask->task;
245
-
246
-        try {
247
-            $this->_time = time();
248
-
249
-            $data = unserialize($QueuedTask->data);
250
-            /** @var \Queue\Shell\Task\QueueTask $task */
251
-            $task = $this->{$taskName};
252
-            if (!$task instanceof QueueTaskInterface) {
253
-                throw new RuntimeException('Task must implement ' . QueueTaskInterface::class);
254
-            }
255
-
256
-            $return = $task->run((array)$data, $QueuedTask->id);
257
-            if ($return !== null) {
258
-                trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED);
259
-            }
260
-            $failureMessage = $taskName . ' failed';
261
-        } catch (Throwable $e) {
262
-            $return = false;
263
-
264
-            $failureMessage = get_class($e) . ': ' . $e->getMessage();
265
-            if (!($e instanceof QueueException)) {
266
-                $failureMessage .= "\n" . $e->getTraceAsString();
267
-            }
268
-
269
-            $this->_logError($taskName . ' (job ' . $QueuedTask->id . ')' . "\n" . $failureMessage, $pid);
270
-        } catch (Exception $e) {
271
-            $return = false;
272
-
273
-            $failureMessage = get_class($e) . ': ' . $e->getMessage();
274
-            $this->_logError($taskName . "\n" . $failureMessage, $pid);
275
-        }
276
-
277
-        if ($return === false) {
278
-            $this->QueuedTasks->markJobFailed($QueuedTask, $failureMessage);
279
-            $failedStatus = $this->QueuedTasks->getFailedStatus($QueuedTask, $this->_getTaskConf());
280
-            $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id . ' failed and ' . $failedStatus, $pid);
281
-            $this->out('Job did not finish, ' . $failedStatus . ' after try ' . $QueuedTask->failed . '.');
282
-
283
-            return;
284
-        }
285
-
286
-        $this->QueuedTasks->markJobDone($QueuedTask);
287
-        $this->out('Job Finished.');
288
-    }
289
-
290
-    /**
291
-     * Manually trigger a Finished job cleanup.
292
-     *
293
-     * @return void
294
-     */
295
-    public function clean()
296
-    {
297
-        if (!Configure::read('Queue.cleanupTimeout')) {
298
-            $this->abort('You disabled cleanuptimout in config. Aborting.');
299
-        }
300
-
301
-        $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int)Configure::read('Queue.cleanupTimeout')));
302
-        $this->QueuedTasks->cleanOldJobs();
303
-    }
304
-
305
-    /**
306
-     * Display current settings
307
-     *
308
-     * @return void
309
-     */
310
-    public function settings()
311
-    {
312
-        $this->out('Current Settings:');
313
-        $conf = (array)Configure::read('Queue');
314
-        foreach ($conf as $key => $val) {
315
-            if ($val === false) {
316
-                $val = 'no';
317
-            }
318
-            if ($val === true) {
319
-                $val = 'yes';
320
-            }
321
-            $this->out('* ' . $key . ': ' . print_r($val, true));
322
-        }
323
-
324
-        $this->out();
325
-    }
326
-
327
-    /**
328
-     * Display some statistics about Finished Jobs.
329
-     *
330
-     * @return void
331
-     */
332
-    public function stats()
333
-    {
334
-        $this->out('Jobs currently in the queue:');
335
-
336
-        $types = $this->QueuedTasks->getTypes()->toArray();
337
-        foreach ($types as $type) {
338
-            $this->out('      ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedTasks->getLength($type));
339
-        }
340
-        $this->hr();
341
-        $this->out('Total unfinished jobs: ' . $this->QueuedTasks->getLength());
342
-        $this->hr();
343
-        $this->out('Finished job statistics:');
344
-        $data = $this->QueuedTasks->getStats();
345
-        foreach ($data as $item) {
346
-            $this->out(' ' . $item['task'] . ': ');
347
-            $this->out('   Finished Jobs in Database: ' . $item['num']);
348
-            $this->out('   Average Job existence    : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's');
349
-            $this->out('   Average Execution delay  : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's');
350
-            $this->out('   Average Execution time   : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's');
351
-        }
352
-    }
353
-
354
-    /**
355
-     * Get option parser method to parse commandline options
356
-     *
357
-     * @return \Cake\Console\ConsoleOptionParser
358
-     */
359
-    public function getOptionParser()
360
-    {
361
-        $subcommandParser = [
362
-            'options' => [
363
-                /*
103
+		return $text;
104
+	}
105
+
106
+	/**
107
+	 * Look for a Queue Task of hte passed name and try to call add() on it.
108
+	 * A QueueTask may provide an add function to enable the user to create new jobs via commandline.
109
+	 *
110
+	 * @return void
111
+	 */
112
+	public function add()
113
+	{
114
+		if (count($this->args) < 1) {
115
+			$this->out('Please call like this:');
116
+			$this->out('       bin/cake queue add <taskname>');
117
+			$this->_displayAvailableTasks();
118
+
119
+			return;
120
+		}
121
+
122
+		$name = Inflector::camelize($this->args[0]);
123
+		if (in_array('Queue' . $name, $this->taskNames, true)) {
124
+			/** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */
125
+			$task = $this->{'Queue' . $name};
126
+			if (!($task instanceof AddInterface)) {
127
+				$this->abort('This task does not support adding via CLI call');
128
+			}
129
+			$task->add();
130
+		} else {
131
+			$this->out('Error: Task not found: ' . $name);
132
+			$this->_displayAvailableTasks();
133
+		}
134
+	}
135
+
136
+	/**
137
+	 * Output the task without Queue or Task
138
+	 * example: QueueImageTask becomes Image on display
139
+	 *
140
+	 * @param string $task Task name
141
+	 * @return string Cleaned task name
142
+	 */
143
+	protected function _taskName($task)
144
+	{
145
+		if (strpos($task, 'Queue') === 0) {
146
+			return substr($task, 5);
147
+		}
148
+
149
+		return $task;
150
+	}
151
+
152
+	/**
153
+	 * Run a QueueWorker loop.
154
+	 * Runs a Queue Worker process which will try to find unassigned jobs in the queue
155
+	 * which it may run and try to fetch and execute them.
156
+	 *
157
+	 * @return int|null
158
+	 */
159
+	public function runworker()
160
+	{
161
+		try {
162
+			$pid = $this->_initPid();
163
+		} catch (PersistenceFailedException $exception) {
164
+			$this->err($exception->getMessage());
165
+			$limit = (int)Configure::read('Queue.maxWorkers');
166
+			if ($limit) {
167
+				$this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')');
168
+			}
169
+
170
+			return static::CODE_ERROR;
171
+		}
172
+
173
+		// Enable Garbage Collector (PHP >= 5.3)
174
+		if (function_exists('gc_enable')) {
175
+			gc_enable();
176
+		}
177
+		if (function_exists('pcntl_signal')) {
178
+			pcntl_signal(SIGTERM, [
179
+				&$this,
180
+				'_exit'
181
+			]);
182
+			pcntl_signal(SIGINT, [
183
+				&$this,
184
+				'_exit'
185
+			]);
186
+			pcntl_signal(SIGTSTP, [
187
+				&$this,
188
+				'_exit'
189
+			]);
190
+			pcntl_signal(SIGQUIT, [
191
+				&$this,
192
+				'_exit'
193
+			]);
194
+		}
195
+		$this->_exit = false;
196
+
197
+		$startTime = time();
198
+		$types = $this->_stringToArray($this->param('type'));
199
+
200
+		while (!$this->_exit) {
201
+			$this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE);
202
+
203
+			$QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types);
204
+
205
+			if ($QueuedTask) {
206
+				$this->runJob($QueuedTask, $pid);
207
+			} elseif (Configure::read('Queue.exitWhenNothingToDo')) {
208
+				$this->out('nothing to do, exiting.');
209
+				$this->_exit = true;
210
+			} else {
211
+				$this->out('nothing to do, sleeping.');
212
+				sleep(Config::sleepTime());
213
+			}
214
+
215
+			// check if we are over the maximum runtime and end processing if so.
216
+			if (Configure::readOrFail('Queue.workerMaxRuntime') && (time() - $startTime) >= Configure::readOrFail('Queue.workerMaxRuntime')) {
217
+				$this->_exit = true;
218
+				$this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.');
219
+			}
220
+			if ($this->_exit || mt_rand(0, 100) > (100 - (int)Config::gcprob())) {
221
+				$this->out(__d('queue', 'Performing old job cleanup.'));
222
+				$this->QueuedTasks->cleanOldJobs();
223
+			}
224
+			$this->hr();
225
+		}
226
+
227
+		$this->_deletePid($pid);
228
+
229
+		if ($this->param('verbose')) {
230
+			$this->_log('endworker', $pid);
231
+		}
232
+	}
233
+
234
+	/**
235
+	 *
236
+	 * @param \Queue\Model\Entity\QueuedTask $QueuedTask Queued task
237
+	 * @param string $pid PID of the process
238
+	 * @return void
239
+	 */
240
+	protected function runJob(QueuedTask $QueuedTask, $pid)
241
+	{
242
+		$this->out('Running Job of type "' . $QueuedTask->task . '"');
243
+		$this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id, $pid, false);
244
+		$taskName = 'Queue' . $QueuedTask->task;
245
+
246
+		try {
247
+			$this->_time = time();
248
+
249
+			$data = unserialize($QueuedTask->data);
250
+			/** @var \Queue\Shell\Task\QueueTask $task */
251
+			$task = $this->{$taskName};
252
+			if (!$task instanceof QueueTaskInterface) {
253
+				throw new RuntimeException('Task must implement ' . QueueTaskInterface::class);
254
+			}
255
+
256
+			$return = $task->run((array)$data, $QueuedTask->id);
257
+			if ($return !== null) {
258
+				trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED);
259
+			}
260
+			$failureMessage = $taskName . ' failed';
261
+		} catch (Throwable $e) {
262
+			$return = false;
263
+
264
+			$failureMessage = get_class($e) . ': ' . $e->getMessage();
265
+			if (!($e instanceof QueueException)) {
266
+				$failureMessage .= "\n" . $e->getTraceAsString();
267
+			}
268
+
269
+			$this->_logError($taskName . ' (job ' . $QueuedTask->id . ')' . "\n" . $failureMessage, $pid);
270
+		} catch (Exception $e) {
271
+			$return = false;
272
+
273
+			$failureMessage = get_class($e) . ': ' . $e->getMessage();
274
+			$this->_logError($taskName . "\n" . $failureMessage, $pid);
275
+		}
276
+
277
+		if ($return === false) {
278
+			$this->QueuedTasks->markJobFailed($QueuedTask, $failureMessage);
279
+			$failedStatus = $this->QueuedTasks->getFailedStatus($QueuedTask, $this->_getTaskConf());
280
+			$this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id . ' failed and ' . $failedStatus, $pid);
281
+			$this->out('Job did not finish, ' . $failedStatus . ' after try ' . $QueuedTask->failed . '.');
282
+
283
+			return;
284
+		}
285
+
286
+		$this->QueuedTasks->markJobDone($QueuedTask);
287
+		$this->out('Job Finished.');
288
+	}
289
+
290
+	/**
291
+	 * Manually trigger a Finished job cleanup.
292
+	 *
293
+	 * @return void
294
+	 */
295
+	public function clean()
296
+	{
297
+		if (!Configure::read('Queue.cleanupTimeout')) {
298
+			$this->abort('You disabled cleanuptimout in config. Aborting.');
299
+		}
300
+
301
+		$this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int)Configure::read('Queue.cleanupTimeout')));
302
+		$this->QueuedTasks->cleanOldJobs();
303
+	}
304
+
305
+	/**
306
+	 * Display current settings
307
+	 *
308
+	 * @return void
309
+	 */
310
+	public function settings()
311
+	{
312
+		$this->out('Current Settings:');
313
+		$conf = (array)Configure::read('Queue');
314
+		foreach ($conf as $key => $val) {
315
+			if ($val === false) {
316
+				$val = 'no';
317
+			}
318
+			if ($val === true) {
319
+				$val = 'yes';
320
+			}
321
+			$this->out('* ' . $key . ': ' . print_r($val, true));
322
+		}
323
+
324
+		$this->out();
325
+	}
326
+
327
+	/**
328
+	 * Display some statistics about Finished Jobs.
329
+	 *
330
+	 * @return void
331
+	 */
332
+	public function stats()
333
+	{
334
+		$this->out('Jobs currently in the queue:');
335
+
336
+		$types = $this->QueuedTasks->getTypes()->toArray();
337
+		foreach ($types as $type) {
338
+			$this->out('      ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedTasks->getLength($type));
339
+		}
340
+		$this->hr();
341
+		$this->out('Total unfinished jobs: ' . $this->QueuedTasks->getLength());
342
+		$this->hr();
343
+		$this->out('Finished job statistics:');
344
+		$data = $this->QueuedTasks->getStats();
345
+		foreach ($data as $item) {
346
+			$this->out(' ' . $item['task'] . ': ');
347
+			$this->out('   Finished Jobs in Database: ' . $item['num']);
348
+			$this->out('   Average Job existence    : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's');
349
+			$this->out('   Average Execution delay  : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's');
350
+			$this->out('   Average Execution time   : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's');
351
+		}
352
+	}
353
+
354
+	/**
355
+	 * Get option parser method to parse commandline options
356
+	 *
357
+	 * @return \Cake\Console\ConsoleOptionParser
358
+	 */
359
+	public function getOptionParser()
360
+	{
361
+		$subcommandParser = [
362
+			'options' => [
363
+				/*
364 364
                  * 'dry-run'=> array(
365 365
                  * 'short' => 'd',
366 366
                  * 'help' => 'Dry run the update, no jobs will actually be added.',
367 367
                  * 'boolean' => true
368 368
                  * ),
369 369
                  */
370
-            ]
371
-        ];
372
-        $subcommandParserFull = $subcommandParser;
373
-        $subcommandParserFull['options']['type'] = [
374
-            'short' => 't',
375
-            'help' => 'Type (comma separated list possible)',
376
-            'default' => null
377
-        ];
378
-
379
-        return parent::getOptionParser()->setDescription($this->getDescription())
380
-            ->addSubcommand('clean', [
381
-                'help' => 'Remove old jobs (cleanup)',
382
-                'parser' => $subcommandParser
383
-            ])
384
-            ->addSubcommand('add', [
385
-                'help' => 'Add Job',
386
-                'parser' => $subcommandParser
387
-            ])
388
-            ->addSubcommand('stats', [
389
-                'help' => 'Stats',
390
-                'parser' => $subcommandParserFull
391
-            ])
392
-            ->addSubcommand('settings', [
393
-                'help' => 'Settings',
394
-                'parser' => $subcommandParserFull
395
-            ])
396
-            ->addSubcommand('runworker', [
397
-                'help' => 'Run Worker',
398
-                'parser' => $subcommandParserFull
399
-            ]);
400
-    }
401
-
402
-    /**
403
-     * Timestamped log.
404
-     *
405
-     * @param string $message Log type
406
-     * @param string|null $pid PID of the process
407
-     * @param bool $addDetails
408
-     * @return void
409
-     */
410
-    protected function _log($message, $pid = null, $addDetails = true)
411
-    {
412
-        if (!Configure::read('Queue.log')) {
413
-            return;
414
-        }
415
-
416
-        if ($addDetails) {
417
-            $timeNeeded = $this->_timeNeeded();
418
-            $memoryUsage = $this->_memoryUsage();
419
-            $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
420
-        }
421
-
422
-        if ($pid) {
423
-            $message .= ' (pid ' . $pid . ')';
424
-        }
425
-        Log::write('info', $message, [
426
-            'scope' => 'queue'
427
-        ]);
428
-    }
429
-
430
-    /**
431
-     *
432
-     * @param string $message Message
433
-     * @param string|null $pid PID of the process
434
-     * @return void
435
-     */
436
-    protected function _logError($message, $pid = null)
437
-    {
438
-        $timeNeeded = $this->_timeNeeded();
439
-        $memoryUsage = $this->_memoryUsage();
440
-        $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
441
-
442
-        if ($pid) {
443
-            $message .= ' (pid ' . $pid . ')';
444
-        }
445
-
446
-        Log::write('error', $message);
447
-    }
448
-
449
-    /**
450
-     * Returns a List of available QueueTasks and their individual configurations.
451
-     *
452
-     * @return array
453
-     */
454
-    protected function _getTaskConf()
455
-    {
456
-        if (!is_array($this->_taskConf)) {
457
-            $this->_taskConf = [];
458
-            foreach ($this->tasks as $task) {
459
-                list ($pluginName, $taskName) = pluginSplit($task);
460
-
461
-                $this->_taskConf[$taskName]['name'] = substr($taskName, 5);
462
-                $this->_taskConf[$taskName]['plugin'] = $pluginName;
463
-                if (property_exists($this->{$taskName}, 'timeout')) {
464
-                    $this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout;
465
-                } else {
466
-                    $this->_taskConf[$taskName]['timeout'] = Config::defaultWorkerTimeout();
467
-                }
468
-                if (property_exists($this->{$taskName}, 'retries')) {
469
-                    $this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries;
470
-                } else {
471
-                    $this->_taskConf[$taskName]['retries'] = Config::defaultWorkerRetries();
472
-                }
473
-                if (property_exists($this->{$taskName}, 'cleanupTimeout')) {
474
-                    $this->_taskConf[$taskName]['cleanupTimeout'] = $this->{$taskName}->cleanupTimeout;
475
-                } else {
476
-                    $this->_taskConf[$taskName]['cleanupTimeout'] = Config::cleanupTimeout();
477
-                }
478
-            }
479
-        }
480
-
481
-        return $this->_taskConf;
482
-    }
483
-
484
-    /**
485
-     * Signal handling to queue worker for clean shutdown
486
-     *
487
-     * @param int $signal The signal
488
-     * @return void
489
-     */
490
-    protected function _exit($signal)
491
-    {
492
-        $this->out(__d('queue', 'Caught %d signal, exiting.', $signal));
493
-        $this->_exit = true;
494
-    }
495
-
496
-    /**
497
-     *
498
-     * @return void
499
-     */
500
-    protected function _displayAvailableTasks()
501
-    {
502
-        $this->out('Available Tasks:');
503
-        foreach ($this->taskNames as $loadedTask) {
504
-            $this->out("\t" . '* ' . $this->_taskName($loadedTask));
505
-        }
506
-    }
507
-
508
-    /**
509
-     *
510
-     * @return string
511
-     */
512
-    protected function _initPid()
513
-    {
514
-        $this->_pid = $this->_retrievePid();
515
-
516
-        return $this->_pid;
517
-    }
518
-
519
-    /**
520
-     *
521
-     * @return string
522
-     */
523
-    protected function _retrievePid()
524
-    {
525
-        if (function_exists('posix_getpid')) {
526
-            $pid = (string)posix_getpid();
527
-        } else {
528
-            $pid = $this->QueuedTasks->key();
529
-        }
530
-
531
-        return $pid;
532
-    }
533
-
534
-    /**
535
-     *
536
-     * @return string Memory usage in MB.
537
-     */
538
-    protected function _memoryUsage()
539
-    {
540
-        $limit = ini_get('memory_limit');
541
-
542
-        $used = number_format(memory_get_peak_usage(true) / (1024 * 1024), 0) . 'MB';
543
-        if ($limit !== '-1') {
544
-            $used .= '/' . $limit;
545
-        }
546
-
547
-        return $used;
548
-    }
549
-
550
-    /**
551
-     *
552
-     * @param string|null $pid PID of the process
553
-     *
554
-     * @return void
555
-     */
556
-    protected function _deletePid($pid)
557
-    {
558
-        if (!$pid) {
559
-            $pid = $this->_pid;
560
-        }
561
-        if (!$pid) {
562
-            return;
563
-        }
564
-    }
565
-
566
-    /**
567
-     *
568
-     * @return string
569
-     */
570
-    protected function _timeNeeded()
571
-    {
572
-        $diff = $this->_time() - $this->_time($this->_time);
573
-        $seconds = max($diff, 1);
574
-
575
-        return $seconds . 's';
576
-    }
577
-
578
-    /**
579
-     *
580
-     * @param int|null $providedTime Provided time
581
-     *
582
-     * @return int
583
-     */
584
-    protected function _time($providedTime = null)
585
-    {
586
-        if ($providedTime) {
587
-            return $providedTime;
588
-        }
589
-
590
-        return time();
591
-    }
592
-
593
-    /**
594
-     *
595
-     * @param string|null $param String to convert
596
-     * @return array
597
-     */
598
-    protected function _stringToArray($param)
599
-    {
600
-        if (!$param) {
601
-            return [];
602
-        }
603
-
604
-        $array = Text::tokenize($param);
605
-
606
-        return array_filter($array);
607
-    }
370
+			]
371
+		];
372
+		$subcommandParserFull = $subcommandParser;
373
+		$subcommandParserFull['options']['type'] = [
374
+			'short' => 't',
375
+			'help' => 'Type (comma separated list possible)',
376
+			'default' => null
377
+		];
378
+
379
+		return parent::getOptionParser()->setDescription($this->getDescription())
380
+			->addSubcommand('clean', [
381
+				'help' => 'Remove old jobs (cleanup)',
382
+				'parser' => $subcommandParser
383
+			])
384
+			->addSubcommand('add', [
385
+				'help' => 'Add Job',
386
+				'parser' => $subcommandParser
387
+			])
388
+			->addSubcommand('stats', [
389
+				'help' => 'Stats',
390
+				'parser' => $subcommandParserFull
391
+			])
392
+			->addSubcommand('settings', [
393
+				'help' => 'Settings',
394
+				'parser' => $subcommandParserFull
395
+			])
396
+			->addSubcommand('runworker', [
397
+				'help' => 'Run Worker',
398
+				'parser' => $subcommandParserFull
399
+			]);
400
+	}
401
+
402
+	/**
403
+	 * Timestamped log.
404
+	 *
405
+	 * @param string $message Log type
406
+	 * @param string|null $pid PID of the process
407
+	 * @param bool $addDetails
408
+	 * @return void
409
+	 */
410
+	protected function _log($message, $pid = null, $addDetails = true)
411
+	{
412
+		if (!Configure::read('Queue.log')) {
413
+			return;
414
+		}
415
+
416
+		if ($addDetails) {
417
+			$timeNeeded = $this->_timeNeeded();
418
+			$memoryUsage = $this->_memoryUsage();
419
+			$message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
420
+		}
421
+
422
+		if ($pid) {
423
+			$message .= ' (pid ' . $pid . ')';
424
+		}
425
+		Log::write('info', $message, [
426
+			'scope' => 'queue'
427
+		]);
428
+	}
429
+
430
+	/**
431
+	 *
432
+	 * @param string $message Message
433
+	 * @param string|null $pid PID of the process
434
+	 * @return void
435
+	 */
436
+	protected function _logError($message, $pid = null)
437
+	{
438
+		$timeNeeded = $this->_timeNeeded();
439
+		$memoryUsage = $this->_memoryUsage();
440
+		$message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
441
+
442
+		if ($pid) {
443
+			$message .= ' (pid ' . $pid . ')';
444
+		}
445
+
446
+		Log::write('error', $message);
447
+	}
448
+
449
+	/**
450
+	 * Returns a List of available QueueTasks and their individual configurations.
451
+	 *
452
+	 * @return array
453
+	 */
454
+	protected function _getTaskConf()
455
+	{
456
+		if (!is_array($this->_taskConf)) {
457
+			$this->_taskConf = [];
458
+			foreach ($this->tasks as $task) {
459
+				list ($pluginName, $taskName) = pluginSplit($task);
460
+
461
+				$this->_taskConf[$taskName]['name'] = substr($taskName, 5);
462
+				$this->_taskConf[$taskName]['plugin'] = $pluginName;
463
+				if (property_exists($this->{$taskName}, 'timeout')) {
464
+					$this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout;
465
+				} else {
466
+					$this->_taskConf[$taskName]['timeout'] = Config::defaultWorkerTimeout();
467
+				}
468
+				if (property_exists($this->{$taskName}, 'retries')) {
469
+					$this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries;
470
+				} else {
471
+					$this->_taskConf[$taskName]['retries'] = Config::defaultWorkerRetries();
472
+				}
473
+				if (property_exists($this->{$taskName}, 'cleanupTimeout')) {
474
+					$this->_taskConf[$taskName]['cleanupTimeout'] = $this->{$taskName}->cleanupTimeout;
475
+				} else {
476
+					$this->_taskConf[$taskName]['cleanupTimeout'] = Config::cleanupTimeout();
477
+				}
478
+			}
479
+		}
480
+
481
+		return $this->_taskConf;
482
+	}
483
+
484
+	/**
485
+	 * Signal handling to queue worker for clean shutdown
486
+	 *
487
+	 * @param int $signal The signal
488
+	 * @return void
489
+	 */
490
+	protected function _exit($signal)
491
+	{
492
+		$this->out(__d('queue', 'Caught %d signal, exiting.', $signal));
493
+		$this->_exit = true;
494
+	}
495
+
496
+	/**
497
+	 *
498
+	 * @return void
499
+	 */
500
+	protected function _displayAvailableTasks()
501
+	{
502
+		$this->out('Available Tasks:');
503
+		foreach ($this->taskNames as $loadedTask) {
504
+			$this->out("\t" . '* ' . $this->_taskName($loadedTask));
505
+		}
506
+	}
507
+
508
+	/**
509
+	 *
510
+	 * @return string
511
+	 */
512
+	protected function _initPid()
513
+	{
514
+		$this->_pid = $this->_retrievePid();
515
+
516
+		return $this->_pid;
517
+	}
518
+
519
+	/**
520
+	 *
521
+	 * @return string
522
+	 */
523
+	protected function _retrievePid()
524
+	{
525
+		if (function_exists('posix_getpid')) {
526
+			$pid = (string)posix_getpid();
527
+		} else {
528
+			$pid = $this->QueuedTasks->key();
529
+		}
530
+
531
+		return $pid;
532
+	}
533
+
534
+	/**
535
+	 *
536
+	 * @return string Memory usage in MB.
537
+	 */
538
+	protected function _memoryUsage()
539
+	{
540
+		$limit = ini_get('memory_limit');
541
+
542
+		$used = number_format(memory_get_peak_usage(true) / (1024 * 1024), 0) . 'MB';
543
+		if ($limit !== '-1') {
544
+			$used .= '/' . $limit;
545
+		}
546
+
547
+		return $used;
548
+	}
549
+
550
+	/**
551
+	 *
552
+	 * @param string|null $pid PID of the process
553
+	 *
554
+	 * @return void
555
+	 */
556
+	protected function _deletePid($pid)
557
+	{
558
+		if (!$pid) {
559
+			$pid = $this->_pid;
560
+		}
561
+		if (!$pid) {
562
+			return;
563
+		}
564
+	}
565
+
566
+	/**
567
+	 *
568
+	 * @return string
569
+	 */
570
+	protected function _timeNeeded()
571
+	{
572
+		$diff = $this->_time() - $this->_time($this->_time);
573
+		$seconds = max($diff, 1);
574
+
575
+		return $seconds . 's';
576
+	}
577
+
578
+	/**
579
+	 *
580
+	 * @param int|null $providedTime Provided time
581
+	 *
582
+	 * @return int
583
+	 */
584
+	protected function _time($providedTime = null)
585
+	{
586
+		if ($providedTime) {
587
+			return $providedTime;
588
+		}
589
+
590
+		return time();
591
+	}
592
+
593
+	/**
594
+	 *
595
+	 * @param string|null $param String to convert
596
+	 * @return array
597
+	 */
598
+	protected function _stringToArray($param)
599
+	{
600
+		if (!$param) {
601
+			return [];
602
+		}
603
+
604
+		$array = Text::tokenize($param);
605
+
606
+		return array_filter($array);
607
+	}
608 608
 }
Please login to merge, or discard this patch.
src/Model/Table/QueuedTasksTable.php 1 patch
Indentation   +593 added lines, -593 removed lines patch added patch discarded remove patch
@@ -28,597 +28,597 @@
 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
-
113
-        return $this->saveOrFail($queuedTask);
114
-    }
115
-
116
-    /**
117
-     *
118
-     * @param string|null $taskName
119
-     *
120
-     * @return bool
121
-     *
122
-     * @throws \InvalidArgumentException
123
-     */
124
-    public function isQueued($taskName = null)
125
-    {
126
-        $conditions = [
127
-            'completed IS' => null
128
-        ];
129
-        if ($taskName) {
130
-            $conditions['task'] = $taskName;
131
-        }
132
-
133
-        return (bool)$this->find()
134
-            ->where($conditions)
135
-            ->select([
136
-                'id'
137
-            ])
138
-            ->first();
139
-    }
140
-
141
-    /**
142
-     * Returns the number of items in the queue.
143
-     * Either returns the number of ALL pending jobs, or the number of pending jobs of the passed type.
144
-     *
145
-     * @param string|null $taskName Task type to Count
146
-     * @return int
147
-     */
148
-    public function getLength($taskName = null)
149
-    {
150
-        $findConf = [
151
-            'conditions' => [
152
-                'completed IS' => null
153
-            ]
154
-        ];
155
-        if ($taskName !== null) {
156
-            $findConf['conditions']['task'] = $taskName;
157
-        }
158
-
159
-        return $this->find('all', $findConf)->count();
160
-    }
161
-
162
-    /**
163
-     * Return a list of all task types in the Queue.
164
-     *
165
-     * @return \Cake\ORM\Query
166
-     */
167
-    public function getTypes()
168
-    {
169
-        $findCond = [
170
-            'fields' => [
171
-                'task'
172
-            ],
173
-            'group' => [
174
-                'task'
175
-            ],
176
-            'keyField' => 'task',
177
-            'valueField' => 'task'
178
-        ];
179
-
180
-        return $this->find('list', $findCond);
181
-    }
182
-
183
-    /**
184
-     * Return some statistics about finished jobs still in the Database.
185
-     * TO-DO: rewrite as virtual field
186
-     *
187
-     * @return \Cake\ORM\Query
188
-     */
189
-    public function getStats()
190
-    {
191
-        $driverName = $this->_getDriverName();
192
-        $options = [
193
-            'fields' => function (Query $query) use ($driverName) {
194
-                $alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)');
195
-                $runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
196
-                $fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(not_before is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(not_before))');
197
-                switch ($driverName) {
198
-                    case static::DRIVER_SQLSERVER:
199
-                        $alltime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', created)");
200
-                        $runtime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
201
-                        $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)");
202
-                        break;
203
-                }
204
-                /**
205
-                 *
206
-                 * @var \Cake\ORM\Query
207
-                 */
208
-                return [
209
-                    'task',
210
-                    'num' => $query->func()->count('*'),
211
-                    'alltime' => $alltime,
212
-                    'runtime' => $runtime,
213
-                    'fetchdelay' => $fetchdelay
214
-                ];
215
-            },
216
-            'conditions' => [
217
-                'completed IS NOT' => null
218
-            ],
219
-            'group' => [
220
-                'task'
221
-            ]
222
-        ];
223
-
224
-        return $this->find('all', $options);
225
-    }
226
-
227
-    /**
228
-     * Returns [
229
-     * 'Task' => [
230
-     * 'YYYY-MM-DD' => INT,
231
-     * ...
232
-     * ]
233
-     * ]
234
-     *
235
-     * @param string|null $taskName The task name
236
-     * @return array
237
-     */
238
-    public function getFullStats($taskName = null)
239
-    {
240
-        $driverName = $this->_getDriverName();
241
-        $fields = function (Query $query) use ($driverName) {
242
-            $runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
243
-            switch ($driverName) {
244
-                case static::DRIVER_SQLSERVER:
245
-                    $runtime = $query->newExpr("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
246
-                    break;
247
-            }
248
-
249
-            return [
250
-                'task',
251
-                'created',
252
-                'duration' => $runtime
253
-            ];
254
-        };
255
-
256
-        $conditions = [
257
-            'completed IS NOT' => null
258
-        ];
259
-        if ($taskName) {
260
-            $conditions['task'] = $taskName;
261
-        }
262
-
263
-        $tasks = $this->find()
264
-            ->select($fields)
265
-            ->where($conditions)
266
-            ->enableHydration(false)
267
-            ->orderDesc('id')
268
-            ->limit(static::STATS_LIMIT)
269
-            ->all()
270
-            ->toArray();
271
-
272
-        $result = [];
273
-
274
-        $days = [];
275
-
276
-        foreach ($tasks as $task) {
277
-            /** @var \DateTime $created */
278
-            $created = $task['created'];
279
-            $day = $created->format('Y-m-d');
280
-            if (!isset($days[$day])) {
281
-                $days[$day] = $day;
282
-            }
283
-
284
-            $result[$task['task']][$day][] = $task['duration'];
285
-        }
286
-
287
-        foreach ($result as $type => $tasks) {
288
-            foreach ($tasks as $day => $durations) {
289
-                $average = array_sum($durations) / count($durations);
290
-                $result[$type][$day] = (int)$average;
291
-            }
292
-
293
-            foreach ($days as $day) {
294
-                if (isset($result[$type][$day])) {
295
-                    continue;
296
-                }
297
-
298
-                $result[$type][$day] = 0;
299
-            }
300
-
301
-            ksort($result[$type]);
302
-        }
303
-
304
-        return $result;
305
-    }
306
-
307
-    /**
308
-     * Look for a new job that can be processed with the current abilities and
309
-     * from the specified group (or any if null).
310
-     *
311
-     * @param array $capabilities Available QueueWorkerTasks.
312
-     * @param array $types Request a job from these types (or exclude certain types), or any otherwise.
313
-     * @return \Queue\Model\Entity\QueuedTask|null
314
-     */
315
-    public function requestJob(array $capabilities, array $types = [])
316
-    {
317
-        $now = $this->getDateTime();
318
-        $nowStr = $now->toDateTimeString();
319
-        $driverName = $this->_getDriverName();
320
-
321
-        $query = $this->find();
322
-        $age = $query->newExpr()->add('IFNULL(TIMESTAMPDIFF(SECOND, "' . $nowStr . '", not_before), 0)');
323
-        switch ($driverName) {
324
-            case static::DRIVER_SQLSERVER:
325
-                $age = $query->newExpr()->add('ISNULL(DATEDIFF(SECOND, GETDATE(), not_before), 0)');
326
-                break;
327
-            case static::DRIVER_POSTGRES:
328
-                $age = $query->newExpr()->add('COALESCE((EXTRACT(EPOCH FROM now()) - EXTRACT(EPOCH FROM not_before)), 0)');
329
-                break;
330
-        }
331
-        $options = [
332
-            'conditions' => [
333
-                'completed IS' => null,
334
-                'OR' => []
335
-            ],
336
-            'fields' => [
337
-                'age' => $age
338
-            ],
339
-            'order' => [
340
-                'age' => 'ASC',
341
-                'id' => 'ASC'
342
-            ]
343
-        ];
344
-
345
-        if ($types) {
346
-            $options['conditions'] = $this->addFilter($options['conditions'], 'task', $types);
347
-        }
348
-
349
-        // Generate the task specific conditions.
350
-        foreach ($capabilities as $task) {
351
-            list ($plugin, $name) = pluginSplit($task['name']);
352
-            $timeoutAt = $now->copy();
353
-            $tmp = [
354
-                'task' => $name,
355
-                'AND' => [
356
-                    [
357
-                        'OR' => [
358
-                            'not_before <' => $nowStr,
359
-                            'not_before IS' => null
360
-                        ]
361
-                    ],
362
-                    [
363
-                        'OR' => [
364
-                            'fetched <' => $timeoutAt->subSeconds($task['timeout']),
365
-                            'fetched IS' => null
366
-                        ]
367
-                    ]
368
-                ],
369
-                'failed_count <' => ($task['retries'] + 1)
370
-            ];
371
-            $options['conditions']['OR'][] = $tmp;
372
-        }
373
-
374
-        /** @var \Queue\Model\Entity\QueuedTask|null $task */
375
-        $task = $this->getConnection()->transactional(function () use ($query, $options, $now) {
376
-            $task = $query->find('all', $options)
377
-                ->enableAutoFields(true)
378
-                ->epilog('FOR UPDATE')
379
-                ->first();
380
-
381
-            if (!$task) {
382
-                return null;
383
-            }
384
-
385
-            $key = $this->key();
386
-            $task = $this->patchEntity($task, [
387
-                'worker_key' => $key,
388
-                'fetched' => $now
389
-            ]);
390
-
391
-            return $this->saveOrFail($task);
392
-        });
393
-
394
-        if (!$task) {
395
-            return null;
396
-        }
397
-
398
-        return $task;
399
-    }
400
-
401
-    /**
402
-     * Mark a task as Completed, removing it from the queue.
403
-     *
404
-     * @param \Queue\Model\Entity\QueuedTask $task Task
405
-     * @return bool Success
406
-     */
407
-    public function markJobDone(QueuedTask $task)
408
-    {
409
-        $fields = [
410
-            'completed' => $this->getDateTime()
411
-        ];
412
-        $task = $this->patchEntity($task, $fields);
413
-
414
-        return (bool)$this->save($task);
415
-    }
416
-
417
-    /**
418
-     * Mark a job as Failed, incrementing the failed-counter and Requeueing it.
419
-     *
420
-     * @param \Queue\Model\Entity\QueuedTask $task Task
421
-     * @param string|null $failureMessage Optional message to append to the failure_message field.
422
-     * @return bool Success
423
-     */
424
-    public function markJobFailed(QueuedTask $task, $failureMessage = null)
425
-    {
426
-        $fields = [
427
-            'failed_count' => $task->failed_count + 1,
428
-            'failure_message' => $failureMessage
429
-        ];
430
-        $task = $this->patchEntity($task, $fields);
431
-
432
-        return (bool)$this->save($task);
433
-    }
434
-
435
-    /**
436
-     * Reset current jobs
437
-     *
438
-     * @param int|null $id ID
439
-     *
440
-     * @return int Success
441
-     */
442
-    public function reset($id = null)
443
-    {
444
-        $fields = [
445
-            'completed' => null,
446
-            'fetched' => null,
447
-            'failed_count' => 0,
448
-            'worker_key' => null,
449
-            'failure_message' => null
450
-        ];
451
-        $conditions = [
452
-            'completed IS' => null
453
-        ];
454
-        if ($id) {
455
-            $conditions['id'] = $id;
456
-        }
457
-
458
-        return $this->updateAll($fields, $conditions);
459
-    }
460
-
461
-    /**
462
-     *
463
-     * @param string $taskName Task name
464
-     *
465
-     * @return int
466
-     */
467
-    public function rerun($taskName)
468
-    {
469
-        $fields = [
470
-            'completed' => null,
471
-            'fetched' => null,
472
-            'failed_count' => 0,
473
-            'worker_key' => null,
474
-            'failure_message' => null
475
-        ];
476
-        $conditions = [
477
-            'completed IS NOT' => null,
478
-            'task' => $taskName
479
-        ];
480
-
481
-        return $this->updateAll($fields, $conditions);
482
-    }
483
-
484
-    /**
485
-     * Cleanup/Delete Completed Tasks.
486
-     *
487
-     * @return void
488
-     */
489
-    public function cleanOldJobs()
490
-    {
491
-        if (!Configure::read('Queue.cleanuptimeout')) {
492
-            return;
493
-        }
494
-
495
-        $this->deleteAll([
496
-            'completed <' => time() - (int)Configure::read('Queue.cleanuptimeout')
497
-        ]);
498
-    }
499
-
500
-    /**
501
-     *
502
-     * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task
503
-     * @param array $taskConfiguration Task configuration
504
-     * @return string
505
-     */
506
-    public function getFailedStatus($queuedTask, array $taskConfiguration)
507
-    {
508
-        $failureMessageRequeued = 'requeued';
509
-
510
-        $queuedTaskName = 'Queue' . $queuedTask->task;
511
-        if (empty($taskConfiguration[$queuedTaskName])) {
512
-            return $failureMessageRequeued;
513
-        }
514
-        $retries = $taskConfiguration[$queuedTaskName]['retries'];
515
-        if ($queuedTask->failed_count <= $retries) {
516
-            return $failureMessageRequeued;
517
-        }
518
-
519
-        return 'aborted';
520
-    }
521
-
522
-    /**
523
-     * Generates a unique Identifier for the current worker thread.
524
-     *
525
-     * Useful to identify the currently running processes for this thread.
526
-     *
527
-     * @return string Identifier
528
-     */
529
-    public function key()
530
-    {
531
-        if ($this->_key !== null) {
532
-            return $this->_key;
533
-        }
534
-        $this->_key = sha1(microtime());
535
-        if (!$this->_key) {
536
-            throw new RuntimeException('Invalid key generated');
537
-        }
538
-
539
-        return $this->_key;
540
-    }
541
-
542
-    /**
543
-     * Resets worker Identifier
544
-     *
545
-     * @return void
546
-     */
547
-    public function clearKey()
548
-    {
549
-        $this->_key = null;
550
-    }
551
-
552
-    /**
553
-     * truncate()
554
-     *
555
-     * @return void
556
-     */
557
-    public function truncate()
558
-    {
559
-        $sql = $this->getSchema()->truncateSql($this->_connection);
560
-        foreach ($sql as $snippet) {
561
-            $this->_connection->execute($snippet);
562
-        }
563
-    }
564
-
565
-    /**
566
-     * get the name of the driver
567
-     *
568
-     * @return string
569
-     */
570
-    protected function _getDriverName()
571
-    {
572
-        $className = explode('\\', $this->getConnection()->config()['driver']);
573
-        $name = end($className);
574
-
575
-        return $name;
576
-    }
577
-
578
-    /**
579
-     *
580
-     * @param array $conditions Conditions
581
-     * @param string $key Key
582
-     * @param array $values Values
583
-     * @return array
584
-     */
585
-    protected function addFilter(array $conditions, $key, array $values)
586
-    {
587
-        $include = [];
588
-        $exclude = [];
589
-        foreach ($values as $value) {
590
-            if (substr($value, 0, 1) === '-') {
591
-                $exclude[] = substr($value, 1);
592
-            } else {
593
-                $include[] = $value;
594
-            }
595
-        }
596
-
597
-        if ($include) {
598
-            $conditions[$key . ' IN'] = $include;
599
-        }
600
-        if ($exclude) {
601
-            $conditions[$key . ' NOT IN'] = $exclude;
602
-        }
603
-
604
-        return $conditions;
605
-    }
606
-
607
-    /**
608
-     * Returns a DateTime object from different input.
609
-     *
610
-     * Without argument this will be "now".
611
-     *
612
-     * @param int|string|\Cake\I18n\FrozenTime|\Cake\I18n\Time|null $notBefore Not before time
613
-     *
614
-     * @return \Cake\I18n\FrozenTime|\Cake\I18n\Time
615
-     */
616
-    protected function getDateTime($notBefore = null)
617
-    {
618
-        if (is_object($notBefore)) {
619
-            return $notBefore;
620
-        }
621
-
622
-        return new FrozenTime($notBefore);
623
-    }
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
+
113
+		return $this->saveOrFail($queuedTask);
114
+	}
115
+
116
+	/**
117
+	 *
118
+	 * @param string|null $taskName
119
+	 *
120
+	 * @return bool
121
+	 *
122
+	 * @throws \InvalidArgumentException
123
+	 */
124
+	public function isQueued($taskName = null)
125
+	{
126
+		$conditions = [
127
+			'completed IS' => null
128
+		];
129
+		if ($taskName) {
130
+			$conditions['task'] = $taskName;
131
+		}
132
+
133
+		return (bool)$this->find()
134
+			->where($conditions)
135
+			->select([
136
+				'id'
137
+			])
138
+			->first();
139
+	}
140
+
141
+	/**
142
+	 * Returns the number of items in the queue.
143
+	 * Either returns the number of ALL pending jobs, or the number of pending jobs of the passed type.
144
+	 *
145
+	 * @param string|null $taskName Task type to Count
146
+	 * @return int
147
+	 */
148
+	public function getLength($taskName = null)
149
+	{
150
+		$findConf = [
151
+			'conditions' => [
152
+				'completed IS' => null
153
+			]
154
+		];
155
+		if ($taskName !== null) {
156
+			$findConf['conditions']['task'] = $taskName;
157
+		}
158
+
159
+		return $this->find('all', $findConf)->count();
160
+	}
161
+
162
+	/**
163
+	 * Return a list of all task types in the Queue.
164
+	 *
165
+	 * @return \Cake\ORM\Query
166
+	 */
167
+	public function getTypes()
168
+	{
169
+		$findCond = [
170
+			'fields' => [
171
+				'task'
172
+			],
173
+			'group' => [
174
+				'task'
175
+			],
176
+			'keyField' => 'task',
177
+			'valueField' => 'task'
178
+		];
179
+
180
+		return $this->find('list', $findCond);
181
+	}
182
+
183
+	/**
184
+	 * Return some statistics about finished jobs still in the Database.
185
+	 * TO-DO: rewrite as virtual field
186
+	 *
187
+	 * @return \Cake\ORM\Query
188
+	 */
189
+	public function getStats()
190
+	{
191
+		$driverName = $this->_getDriverName();
192
+		$options = [
193
+			'fields' => function (Query $query) use ($driverName) {
194
+				$alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)');
195
+				$runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
196
+				$fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(not_before is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(not_before))');
197
+				switch ($driverName) {
198
+					case static::DRIVER_SQLSERVER:
199
+						$alltime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', created)");
200
+						$runtime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
201
+						$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)");
202
+						break;
203
+				}
204
+				/**
205
+				 *
206
+				 * @var \Cake\ORM\Query
207
+				 */
208
+				return [
209
+					'task',
210
+					'num' => $query->func()->count('*'),
211
+					'alltime' => $alltime,
212
+					'runtime' => $runtime,
213
+					'fetchdelay' => $fetchdelay
214
+				];
215
+			},
216
+			'conditions' => [
217
+				'completed IS NOT' => null
218
+			],
219
+			'group' => [
220
+				'task'
221
+			]
222
+		];
223
+
224
+		return $this->find('all', $options);
225
+	}
226
+
227
+	/**
228
+	 * Returns [
229
+	 * 'Task' => [
230
+	 * 'YYYY-MM-DD' => INT,
231
+	 * ...
232
+	 * ]
233
+	 * ]
234
+	 *
235
+	 * @param string|null $taskName The task name
236
+	 * @return array
237
+	 */
238
+	public function getFullStats($taskName = null)
239
+	{
240
+		$driverName = $this->_getDriverName();
241
+		$fields = function (Query $query) use ($driverName) {
242
+			$runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
243
+			switch ($driverName) {
244
+				case static::DRIVER_SQLSERVER:
245
+					$runtime = $query->newExpr("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
246
+					break;
247
+			}
248
+
249
+			return [
250
+				'task',
251
+				'created',
252
+				'duration' => $runtime
253
+			];
254
+		};
255
+
256
+		$conditions = [
257
+			'completed IS NOT' => null
258
+		];
259
+		if ($taskName) {
260
+			$conditions['task'] = $taskName;
261
+		}
262
+
263
+		$tasks = $this->find()
264
+			->select($fields)
265
+			->where($conditions)
266
+			->enableHydration(false)
267
+			->orderDesc('id')
268
+			->limit(static::STATS_LIMIT)
269
+			->all()
270
+			->toArray();
271
+
272
+		$result = [];
273
+
274
+		$days = [];
275
+
276
+		foreach ($tasks as $task) {
277
+			/** @var \DateTime $created */
278
+			$created = $task['created'];
279
+			$day = $created->format('Y-m-d');
280
+			if (!isset($days[$day])) {
281
+				$days[$day] = $day;
282
+			}
283
+
284
+			$result[$task['task']][$day][] = $task['duration'];
285
+		}
286
+
287
+		foreach ($result as $type => $tasks) {
288
+			foreach ($tasks as $day => $durations) {
289
+				$average = array_sum($durations) / count($durations);
290
+				$result[$type][$day] = (int)$average;
291
+			}
292
+
293
+			foreach ($days as $day) {
294
+				if (isset($result[$type][$day])) {
295
+					continue;
296
+				}
297
+
298
+				$result[$type][$day] = 0;
299
+			}
300
+
301
+			ksort($result[$type]);
302
+		}
303
+
304
+		return $result;
305
+	}
306
+
307
+	/**
308
+	 * Look for a new job that can be processed with the current abilities and
309
+	 * from the specified group (or any if null).
310
+	 *
311
+	 * @param array $capabilities Available QueueWorkerTasks.
312
+	 * @param array $types Request a job from these types (or exclude certain types), or any otherwise.
313
+	 * @return \Queue\Model\Entity\QueuedTask|null
314
+	 */
315
+	public function requestJob(array $capabilities, array $types = [])
316
+	{
317
+		$now = $this->getDateTime();
318
+		$nowStr = $now->toDateTimeString();
319
+		$driverName = $this->_getDriverName();
320
+
321
+		$query = $this->find();
322
+		$age = $query->newExpr()->add('IFNULL(TIMESTAMPDIFF(SECOND, "' . $nowStr . '", not_before), 0)');
323
+		switch ($driverName) {
324
+			case static::DRIVER_SQLSERVER:
325
+				$age = $query->newExpr()->add('ISNULL(DATEDIFF(SECOND, GETDATE(), not_before), 0)');
326
+				break;
327
+			case static::DRIVER_POSTGRES:
328
+				$age = $query->newExpr()->add('COALESCE((EXTRACT(EPOCH FROM now()) - EXTRACT(EPOCH FROM not_before)), 0)');
329
+				break;
330
+		}
331
+		$options = [
332
+			'conditions' => [
333
+				'completed IS' => null,
334
+				'OR' => []
335
+			],
336
+			'fields' => [
337
+				'age' => $age
338
+			],
339
+			'order' => [
340
+				'age' => 'ASC',
341
+				'id' => 'ASC'
342
+			]
343
+		];
344
+
345
+		if ($types) {
346
+			$options['conditions'] = $this->addFilter($options['conditions'], 'task', $types);
347
+		}
348
+
349
+		// Generate the task specific conditions.
350
+		foreach ($capabilities as $task) {
351
+			list ($plugin, $name) = pluginSplit($task['name']);
352
+			$timeoutAt = $now->copy();
353
+			$tmp = [
354
+				'task' => $name,
355
+				'AND' => [
356
+					[
357
+						'OR' => [
358
+							'not_before <' => $nowStr,
359
+							'not_before IS' => null
360
+						]
361
+					],
362
+					[
363
+						'OR' => [
364
+							'fetched <' => $timeoutAt->subSeconds($task['timeout']),
365
+							'fetched IS' => null
366
+						]
367
+					]
368
+				],
369
+				'failed_count <' => ($task['retries'] + 1)
370
+			];
371
+			$options['conditions']['OR'][] = $tmp;
372
+		}
373
+
374
+		/** @var \Queue\Model\Entity\QueuedTask|null $task */
375
+		$task = $this->getConnection()->transactional(function () use ($query, $options, $now) {
376
+			$task = $query->find('all', $options)
377
+				->enableAutoFields(true)
378
+				->epilog('FOR UPDATE')
379
+				->first();
380
+
381
+			if (!$task) {
382
+				return null;
383
+			}
384
+
385
+			$key = $this->key();
386
+			$task = $this->patchEntity($task, [
387
+				'worker_key' => $key,
388
+				'fetched' => $now
389
+			]);
390
+
391
+			return $this->saveOrFail($task);
392
+		});
393
+
394
+		if (!$task) {
395
+			return null;
396
+		}
397
+
398
+		return $task;
399
+	}
400
+
401
+	/**
402
+	 * Mark a task as Completed, removing it from the queue.
403
+	 *
404
+	 * @param \Queue\Model\Entity\QueuedTask $task Task
405
+	 * @return bool Success
406
+	 */
407
+	public function markJobDone(QueuedTask $task)
408
+	{
409
+		$fields = [
410
+			'completed' => $this->getDateTime()
411
+		];
412
+		$task = $this->patchEntity($task, $fields);
413
+
414
+		return (bool)$this->save($task);
415
+	}
416
+
417
+	/**
418
+	 * Mark a job as Failed, incrementing the failed-counter and Requeueing it.
419
+	 *
420
+	 * @param \Queue\Model\Entity\QueuedTask $task Task
421
+	 * @param string|null $failureMessage Optional message to append to the failure_message field.
422
+	 * @return bool Success
423
+	 */
424
+	public function markJobFailed(QueuedTask $task, $failureMessage = null)
425
+	{
426
+		$fields = [
427
+			'failed_count' => $task->failed_count + 1,
428
+			'failure_message' => $failureMessage
429
+		];
430
+		$task = $this->patchEntity($task, $fields);
431
+
432
+		return (bool)$this->save($task);
433
+	}
434
+
435
+	/**
436
+	 * Reset current jobs
437
+	 *
438
+	 * @param int|null $id ID
439
+	 *
440
+	 * @return int Success
441
+	 */
442
+	public function reset($id = null)
443
+	{
444
+		$fields = [
445
+			'completed' => null,
446
+			'fetched' => null,
447
+			'failed_count' => 0,
448
+			'worker_key' => null,
449
+			'failure_message' => null
450
+		];
451
+		$conditions = [
452
+			'completed IS' => null
453
+		];
454
+		if ($id) {
455
+			$conditions['id'] = $id;
456
+		}
457
+
458
+		return $this->updateAll($fields, $conditions);
459
+	}
460
+
461
+	/**
462
+	 *
463
+	 * @param string $taskName Task name
464
+	 *
465
+	 * @return int
466
+	 */
467
+	public function rerun($taskName)
468
+	{
469
+		$fields = [
470
+			'completed' => null,
471
+			'fetched' => null,
472
+			'failed_count' => 0,
473
+			'worker_key' => null,
474
+			'failure_message' => null
475
+		];
476
+		$conditions = [
477
+			'completed IS NOT' => null,
478
+			'task' => $taskName
479
+		];
480
+
481
+		return $this->updateAll($fields, $conditions);
482
+	}
483
+
484
+	/**
485
+	 * Cleanup/Delete Completed Tasks.
486
+	 *
487
+	 * @return void
488
+	 */
489
+	public function cleanOldJobs()
490
+	{
491
+		if (!Configure::read('Queue.cleanuptimeout')) {
492
+			return;
493
+		}
494
+
495
+		$this->deleteAll([
496
+			'completed <' => time() - (int)Configure::read('Queue.cleanuptimeout')
497
+		]);
498
+	}
499
+
500
+	/**
501
+	 *
502
+	 * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task
503
+	 * @param array $taskConfiguration Task configuration
504
+	 * @return string
505
+	 */
506
+	public function getFailedStatus($queuedTask, array $taskConfiguration)
507
+	{
508
+		$failureMessageRequeued = 'requeued';
509
+
510
+		$queuedTaskName = 'Queue' . $queuedTask->task;
511
+		if (empty($taskConfiguration[$queuedTaskName])) {
512
+			return $failureMessageRequeued;
513
+		}
514
+		$retries = $taskConfiguration[$queuedTaskName]['retries'];
515
+		if ($queuedTask->failed_count <= $retries) {
516
+			return $failureMessageRequeued;
517
+		}
518
+
519
+		return 'aborted';
520
+	}
521
+
522
+	/**
523
+	 * Generates a unique Identifier for the current worker thread.
524
+	 *
525
+	 * Useful to identify the currently running processes for this thread.
526
+	 *
527
+	 * @return string Identifier
528
+	 */
529
+	public function key()
530
+	{
531
+		if ($this->_key !== null) {
532
+			return $this->_key;
533
+		}
534
+		$this->_key = sha1(microtime());
535
+		if (!$this->_key) {
536
+			throw new RuntimeException('Invalid key generated');
537
+		}
538
+
539
+		return $this->_key;
540
+	}
541
+
542
+	/**
543
+	 * Resets worker Identifier
544
+	 *
545
+	 * @return void
546
+	 */
547
+	public function clearKey()
548
+	{
549
+		$this->_key = null;
550
+	}
551
+
552
+	/**
553
+	 * truncate()
554
+	 *
555
+	 * @return void
556
+	 */
557
+	public function truncate()
558
+	{
559
+		$sql = $this->getSchema()->truncateSql($this->_connection);
560
+		foreach ($sql as $snippet) {
561
+			$this->_connection->execute($snippet);
562
+		}
563
+	}
564
+
565
+	/**
566
+	 * get the name of the driver
567
+	 *
568
+	 * @return string
569
+	 */
570
+	protected function _getDriverName()
571
+	{
572
+		$className = explode('\\', $this->getConnection()->config()['driver']);
573
+		$name = end($className);
574
+
575
+		return $name;
576
+	}
577
+
578
+	/**
579
+	 *
580
+	 * @param array $conditions Conditions
581
+	 * @param string $key Key
582
+	 * @param array $values Values
583
+	 * @return array
584
+	 */
585
+	protected function addFilter(array $conditions, $key, array $values)
586
+	{
587
+		$include = [];
588
+		$exclude = [];
589
+		foreach ($values as $value) {
590
+			if (substr($value, 0, 1) === '-') {
591
+				$exclude[] = substr($value, 1);
592
+			} else {
593
+				$include[] = $value;
594
+			}
595
+		}
596
+
597
+		if ($include) {
598
+			$conditions[$key . ' IN'] = $include;
599
+		}
600
+		if ($exclude) {
601
+			$conditions[$key . ' NOT IN'] = $exclude;
602
+		}
603
+
604
+		return $conditions;
605
+	}
606
+
607
+	/**
608
+	 * Returns a DateTime object from different input.
609
+	 *
610
+	 * Without argument this will be "now".
611
+	 *
612
+	 * @param int|string|\Cake\I18n\FrozenTime|\Cake\I18n\Time|null $notBefore Not before time
613
+	 *
614
+	 * @return \Cake\I18n\FrozenTime|\Cake\I18n\Time
615
+	 */
616
+	protected function getDateTime($notBefore = null)
617
+	{
618
+		if (is_object($notBefore)) {
619
+			return $notBefore;
620
+		}
621
+
622
+		return new FrozenTime($notBefore);
623
+	}
624 624
 }
Please login to merge, or discard this patch.