Oefenweb / cakephp-queue

tests/TestCase/Model/Table/QueuedTasksTableTest.php

Indentation +414 added lines, -414 removed lines

@@ -19,418 +19,418 @@
 class QueuedTasksTableTest extends TestCase
 {
 
-    /**
-     *
-     * @var \Queue\Model\Table\QueuedTasksTable
-     */
-    protected $QueuedTasks;
-
-    /**
-     * Fixtures
-     *
-     * @var array
-     */
-    public $fixtures = [
-        'plugin.Queue.QueuedTasks'
-    ];
-
-    /**
-     * setUp method
-     *
-     * @return void
-     */
-    public function setUp()
-    {
-        parent::setUp();
-        $config = TableRegistry::getTableLocator()->exists('QueuedTasks') ? [] : [
-            'className' => QueuedTasksTable::class
-        ];
-        $this->QueuedTasks = TableRegistry::getTableLocator()->get('QueuedTasks', $config);
-    }
-
-    /**
-     * Basic Instance test
-     *
-     * @return void
-     */
-    public function testQueueInstance()
-    {
-        $this->assertInstanceOf(QueuedTasksTable::class, $this->QueuedTasks);
-    }
-
-    /**
-     * Test the basic create and length evaluation functions.
-     *
-     * @return void
-     */
-    public function testCreateAndCount()
-    {
-        // at first, the queue should contain 0 items.
-        $this->assertSame(0, $this->QueuedTasks->getLength());
-
-        // create a job
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('test1', [
-            'some' => 'random',
-            'test' => 'data'
-        ]));
-
-        // test if queue Length is 1 now.
-        $this->assertSame(1, $this->QueuedTasks->getLength());
-
-        // create some more jobs
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('test2', [
-            'some' => 'random',
-            'test' => 'data2'
-        ]));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('test2', [
-            'some' => 'random',
-            'test' => 'data3'
-        ]));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('test3', [
-            'some' => 'random',
-            'test' => 'data4'
-        ]));
-
-        // overall queueLength shpould now be 4
-        $this->assertSame(4, $this->QueuedTasks->getLength());
-
-        // there should be 1 task of type 'test1', one of type 'test3' and 2 of type 'test2'
-        $this->assertSame(1, $this->QueuedTasks->getLength('test1'));
-        $this->assertSame(2, $this->QueuedTasks->getLength('test2'));
-        $this->assertSame(1, $this->QueuedTasks->getLength('test3'));
-    }
-
-    /**
-     * Test the basic create and fetch functions.
-     *
-     * @return void
-     */
-    public function testCreateAndFetch()
-    {
-        $this->_needsConnection();
-
-        // $capabilities is a list of tasks the worker can run.
-        $capabilities = [
-            'task1' => [
-                'name' => 'task1',
-                'timeout' => 100,
-                'retries' => 2
-            ]
-        ];
-        $testData = [
-            'x1' => 'y1',
-            'x2' => 'y2',
-            'x3' => 'y3',
-            'x4' => 'y4'
-        ];
-
-        // start off empty.
-        $this->assertSame([], $this->QueuedTasks->find()
-            ->toArray());
-        // at first, the queue should contain 0 items.
-        $this->assertSame(0, $this->QueuedTasks->getLength());
-        // there are no jobs, so we cant fetch any.
-        $this->assertNull($this->QueuedTasks->requestJob($capabilities));
-        // insert one job.
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $testData));
-
-        // fetch and check the first job.
-        $job = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame(1, $job->id);
-        $this->assertSame('task1', $job->task);
-        $this->assertSame(0, $job->failed);
-        $this->assertNull($job->completed);
-        $this->assertSame($testData, unserialize($job->data));
-
-        // after this job has been fetched, it may not be reassigned.
-        $result = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertNull($result);
-
-        // queue length is still 1 since the first job did not finish.
-        $this->assertSame(1, $this->QueuedTasks->getLength());
-
-        // Now mark Task1 as done
-        $this->assertTrue($this->QueuedTasks->markJobDone($job));
-
-        // Should be 0 again.
-        $this->assertSame(0, $this->QueuedTasks->getLength());
-    }
-
-    /**
-     * Test the delivery of jobs in sequence, skipping fetched but not completed tasks.
-     *
-     * @return void
-     */
-    public function testSequence()
-    {
-        $this->_needsConnection();
-
-        // $capabilities is a list of tasks the worker can run.
-        $capabilities = [
-            'task1' => [
-                'name' => 'task1',
-                'timeout' => 100,
-                'retries' => 2
-            ]
-        ];
-        // at first, the queue should contain 0 items.
-        $this->assertSame(0, $this->QueuedTasks->getLength());
-        // create some more jobs
-        foreach (range(0, 9) as $num) {
-            $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
-                'tasknum' => $num
-            ]));
-        }
-        // 10 jobs in the queue.
-        $this->assertSame(10, $this->QueuedTasks->getLength());
-
-        // jobs should be fetched in the original sequence.
-        $array = [];
-        foreach (range(0, 4) as $num) {
-            $this->QueuedTasks->clearKey();
-            $array[$num] = $this->QueuedTasks->requestJob($capabilities);
-            $jobData = unserialize($array[$num]['data']);
-            $this->assertSame($num, $jobData['tasknum']);
-        }
-        // now mark them as done
-        foreach (range(0, 4) as $num) {
-            $this->assertTrue($this->QueuedTasks->markJobDone($array[$num]));
-            $this->assertSame(9 - $num, $this->QueuedTasks->getLength());
-        }
-
-        // jobs should be fetched in the original sequence.
-        foreach (range(5, 9) as $num) {
-            $job = $this->QueuedTasks->requestJob($capabilities);
-            $jobData = unserialize($job->data);
-            $this->assertSame($num, $jobData['tasknum']);
-            $this->assertTrue($this->QueuedTasks->markJobDone($job));
-            $this->assertSame(9 - $num, $this->QueuedTasks->getLength());
-        }
-    }
-
-    /**
-     * Test creating Jobs to run close to a specified time, and strtotime parsing.
-     * Using toUnixString() function to convert Time object to timestamp, instead of strtotime
-     *
-     * @return null
-     */
-    public function testNotBefore()
-    {
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', null, '+ 1 Min'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', null, '+ 1 Day'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', null, '2009-07-01 12:00:00'));
-        $data = $this->QueuedTasks->find('all')->toArray();
-        $this->assertWithinRange((new Time('+ 1 Min'))->toUnixString(), $data[0]['not_before']->toUnixString(), 60);
-        $this->assertWithinRange((new Time('+ 1 Day'))->toUnixString(), $data[1]['not_before']->toUnixString(), 60);
-        $this->assertWithinRange((new Time('2009-07-01 12:00:00'))->toUnixString(), $data[2]['not_before']->toUnixString(), 60);
-    }
-
-    /**
-     * Test Job reordering depending on 'notBefore' field.
-     * Jobs with an expired not_before field should be executed before any other job without specific timing info.
-     *
-     * @return void
-     */
-    public function testNotBeforeOrder()
-    {
-        $this->_needsConnection();
-
-        $capabilities = [
-            'task1' => [
-                'name' => 'task1',
-                'timeout' => 100,
-                'retries' => 2
-            ],
-            'dummytask' => [
-                'name' => 'dummytask',
-                'timeout' => 100,
-                'retries' => 2
-            ]
-        ];
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
-        // create a task with it's execution target some seconds in the past, so it should jump to the top of the testCreateAndFetchlist.
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
-            'three'
-        ], '- 3 Seconds'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
-            'two'
-        ], '- 5 Seconds'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
-            'one'
-        ], '- 7 Seconds'));
-
-        // when using requestJob, the jobs we just created should be delivered in this order, NOT the order in which they where created.
-        $expected = [
-            [
-                'name' => 'task1',
-                'data' => [
-                    'one'
-                ]
-            ],
-            [
-                'name' => 'task1',
-                'data' => [
-                    'two'
-                ]
-            ],
-            [
-                'name' => 'task1',
-                'data' => [
-                    'three'
-                ]
-            ],
-            [
-                'name' => 'dummytask',
-                'data' => null
-            ],
-            [
-                'name' => 'dummytask',
-                'data' => null
-            ]
-        ];
-
-        foreach ($expected as $item) {
-            $this->QueuedTasks->clearKey();
-            $tmp = $this->QueuedTasks->requestJob($capabilities);
-
-            $this->assertSame($item['name'], $tmp['task']);
-            $this->assertEquals($item['data'], unserialize($tmp['data']));
-        }
-    }
-
-    /**
-     * Job Rate limiting.
-     * Do not execute jobs of a certain type more often than once every X seconds.
-     *
-     * @return void
-     */
-    public function testRateLimit()
-    {
-        $this->_needsConnection();
-
-        $capabilities = [
-            'task1' => [
-                'name' => 'task1',
-                'timeout' => 101,
-                'retries' => 2,
-                'rate' => 2
-            ],
-            'dummytask' => [
-                'name' => 'dummytask',
-                'timeout' => 101,
-                'retries' => 2
-            ]
-        ];
-
-        // clear out the rate history
-        $this->QueuedTasks->rateHistory = [];
-
-        $data1 = [
-            'key' => 1
-        ];
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $data1));
-        $data2 = [
-            'key' => 2
-        ];
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $data2));
-        $data3 = [
-            'key' => 3
-        ];
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $data3));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
-        $this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
-
-        // At first we get task1-1.
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame('task1', $tmp['task']);
-        $this->assertSame($data1, unserialize($tmp['data']));
-
-        // The rate limit should now skip over task1-2 and fetch a dummytask.
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame('dummytask', $tmp['task']);
-        $this->assertFalse(unserialize($tmp['data']));
-
-        usleep(100000);
-        // and again.
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame('dummytask', $tmp['task']);
-        $this->assertFalse(unserialize($tmp['data']));
-
-        // Then some time passes
-        sleep(2);
-
-        // Now we should get task1-2
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame('task1', $tmp['task']);
-        $this->assertSame($data2, unserialize($tmp['data']));
-
-        // and again rate limit to dummytask.
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame('dummytask', $tmp['task']);
-        $this->assertFalse(unserialize($tmp['data']));
-
-        // Then some more time passes
-        sleep(2);
-
-        // Now we should get task1-3
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame('task1', $tmp['task']);
-        $this->assertSame($data3, unserialize($tmp['data']));
-
-        // and again rate limit to dummytask.
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertSame('dummytask', $tmp['task']);
-        $this->assertFalse(unserialize($tmp['data']));
-
-        // and now the queue is empty
-        $this->QueuedTasks->clearKey();
-        $tmp = $this->QueuedTasks->requestJob($capabilities);
-        $this->assertNull($tmp);
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testIsQueued()
-    {
-        $result = $this->QueuedTasks->isQueued('foo-bar');
-        $this->assertFalse($result);
-
-        $queuedJob = $this->QueuedTasks->newEntity([
-            'key' => 'key',
-            'task' => 'FooBar'
-        ]);
-        $this->QueuedTasks->saveOrFail($queuedJob);
-
-        $result = $this->QueuedTasks->isQueued('foo-bar');
-        $this->assertTrue($result);
-
-        $queuedJob->completed = new FrozenTime();
-        $this->QueuedTasks->saveOrFail($queuedJob);
-
-        $result = $this->QueuedTasks->isQueued('foo-bar');
-        $this->assertFalse($result);
-    }
-
-    /**
-     * Helper method for skipping tests that need a real connection.
-     *
-     * @return void
-     */
-    protected function _needsConnection()
-    {
-        $config = ConnectionManager::getConfig('test');
-        $this->skipIf(strpos($config['driver'], 'Mysql') === false, 'Only Mysql is working yet for this.');
-    }
+	/**
+	 *
+	 * @var \Queue\Model\Table\QueuedTasksTable
+	 */
+	protected $QueuedTasks;
+
+	/**
+	 * Fixtures
+	 *
+	 * @var array
+	 */
+	public $fixtures = [
+		'plugin.Queue.QueuedTasks'
+	];
+
+	/**
+	 * setUp method
+	 *
+	 * @return void
+	 */
+	public function setUp()
+	{
+		parent::setUp();
+		$config = TableRegistry::getTableLocator()->exists('QueuedTasks') ? [] : [
+			'className' => QueuedTasksTable::class
+		];
+		$this->QueuedTasks = TableRegistry::getTableLocator()->get('QueuedTasks', $config);
+	}
+
+	/**
+	 * Basic Instance test
+	 *
+	 * @return void
+	 */
+	public function testQueueInstance()
+	{
+		$this->assertInstanceOf(QueuedTasksTable::class, $this->QueuedTasks);
+	}
+
+	/**
+	 * Test the basic create and length evaluation functions.
+	 *
+	 * @return void
+	 */
+	public function testCreateAndCount()
+	{
+		// at first, the queue should contain 0 items.
+		$this->assertSame(0, $this->QueuedTasks->getLength());
+
+		// create a job
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('test1', [
+			'some' => 'random',
+			'test' => 'data'
+		]));
+
+		// test if queue Length is 1 now.
+		$this->assertSame(1, $this->QueuedTasks->getLength());
+
+		// create some more jobs
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('test2', [
+			'some' => 'random',
+			'test' => 'data2'
+		]));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('test2', [
+			'some' => 'random',
+			'test' => 'data3'
+		]));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('test3', [
+			'some' => 'random',
+			'test' => 'data4'
+		]));
+
+		// overall queueLength shpould now be 4
+		$this->assertSame(4, $this->QueuedTasks->getLength());
+
+		// there should be 1 task of type 'test1', one of type 'test3' and 2 of type 'test2'
+		$this->assertSame(1, $this->QueuedTasks->getLength('test1'));
+		$this->assertSame(2, $this->QueuedTasks->getLength('test2'));
+		$this->assertSame(1, $this->QueuedTasks->getLength('test3'));
+	}
+
+	/**
+	 * Test the basic create and fetch functions.
+	 *
+	 * @return void
+	 */
+	public function testCreateAndFetch()
+	{
+		$this->_needsConnection();
+
+		// $capabilities is a list of tasks the worker can run.
+		$capabilities = [
+			'task1' => [
+				'name' => 'task1',
+				'timeout' => 100,
+				'retries' => 2
+			]
+		];
+		$testData = [
+			'x1' => 'y1',
+			'x2' => 'y2',
+			'x3' => 'y3',
+			'x4' => 'y4'
+		];
+
+		// start off empty.
+		$this->assertSame([], $this->QueuedTasks->find()
+			->toArray());
+		// at first, the queue should contain 0 items.
+		$this->assertSame(0, $this->QueuedTasks->getLength());
+		// there are no jobs, so we cant fetch any.
+		$this->assertNull($this->QueuedTasks->requestJob($capabilities));
+		// insert one job.
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $testData));
+
+		// fetch and check the first job.
+		$job = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame(1, $job->id);
+		$this->assertSame('task1', $job->task);
+		$this->assertSame(0, $job->failed);
+		$this->assertNull($job->completed);
+		$this->assertSame($testData, unserialize($job->data));
+
+		// after this job has been fetched, it may not be reassigned.
+		$result = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertNull($result);
+
+		// queue length is still 1 since the first job did not finish.
+		$this->assertSame(1, $this->QueuedTasks->getLength());
+
+		// Now mark Task1 as done
+		$this->assertTrue($this->QueuedTasks->markJobDone($job));
+
+		// Should be 0 again.
+		$this->assertSame(0, $this->QueuedTasks->getLength());
+	}
+
+	/**
+	 * Test the delivery of jobs in sequence, skipping fetched but not completed tasks.
+	 *
+	 * @return void
+	 */
+	public function testSequence()
+	{
+		$this->_needsConnection();
+
+		// $capabilities is a list of tasks the worker can run.
+		$capabilities = [
+			'task1' => [
+				'name' => 'task1',
+				'timeout' => 100,
+				'retries' => 2
+			]
+		];
+		// at first, the queue should contain 0 items.
+		$this->assertSame(0, $this->QueuedTasks->getLength());
+		// create some more jobs
+		foreach (range(0, 9) as $num) {
+			$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
+				'tasknum' => $num
+			]));
+		}
+		// 10 jobs in the queue.
+		$this->assertSame(10, $this->QueuedTasks->getLength());
+
+		// jobs should be fetched in the original sequence.
+		$array = [];
+		foreach (range(0, 4) as $num) {
+			$this->QueuedTasks->clearKey();
+			$array[$num] = $this->QueuedTasks->requestJob($capabilities);
+			$jobData = unserialize($array[$num]['data']);
+			$this->assertSame($num, $jobData['tasknum']);
+		}
+		// now mark them as done
+		foreach (range(0, 4) as $num) {
+			$this->assertTrue($this->QueuedTasks->markJobDone($array[$num]));
+			$this->assertSame(9 - $num, $this->QueuedTasks->getLength());
+		}
+
+		// jobs should be fetched in the original sequence.
+		foreach (range(5, 9) as $num) {
+			$job = $this->QueuedTasks->requestJob($capabilities);
+			$jobData = unserialize($job->data);
+			$this->assertSame($num, $jobData['tasknum']);
+			$this->assertTrue($this->QueuedTasks->markJobDone($job));
+			$this->assertSame(9 - $num, $this->QueuedTasks->getLength());
+		}
+	}
+
+	/**
+	 * Test creating Jobs to run close to a specified time, and strtotime parsing.
+	 * Using toUnixString() function to convert Time object to timestamp, instead of strtotime
+	 *
+	 * @return null
+	 */
+	public function testNotBefore()
+	{
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', null, '+ 1 Min'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', null, '+ 1 Day'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', null, '2009-07-01 12:00:00'));
+		$data = $this->QueuedTasks->find('all')->toArray();
+		$this->assertWithinRange((new Time('+ 1 Min'))->toUnixString(), $data[0]['not_before']->toUnixString(), 60);
+		$this->assertWithinRange((new Time('+ 1 Day'))->toUnixString(), $data[1]['not_before']->toUnixString(), 60);
+		$this->assertWithinRange((new Time('2009-07-01 12:00:00'))->toUnixString(), $data[2]['not_before']->toUnixString(), 60);
+	}
+
+	/**
+	 * Test Job reordering depending on 'notBefore' field.
+	 * Jobs with an expired not_before field should be executed before any other job without specific timing info.
+	 *
+	 * @return void
+	 */
+	public function testNotBeforeOrder()
+	{
+		$this->_needsConnection();
+
+		$capabilities = [
+			'task1' => [
+				'name' => 'task1',
+				'timeout' => 100,
+				'retries' => 2
+			],
+			'dummytask' => [
+				'name' => 'dummytask',
+				'timeout' => 100,
+				'retries' => 2
+			]
+		];
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
+		// create a task with it's execution target some seconds in the past, so it should jump to the top of the testCreateAndFetchlist.
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
+			'three'
+		], '- 3 Seconds'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
+			'two'
+		], '- 5 Seconds'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', [
+			'one'
+		], '- 7 Seconds'));
+
+		// when using requestJob, the jobs we just created should be delivered in this order, NOT the order in which they where created.
+		$expected = [
+			[
+				'name' => 'task1',
+				'data' => [
+					'one'
+				]
+			],
+			[
+				'name' => 'task1',
+				'data' => [
+					'two'
+				]
+			],
+			[
+				'name' => 'task1',
+				'data' => [
+					'three'
+				]
+			],
+			[
+				'name' => 'dummytask',
+				'data' => null
+			],
+			[
+				'name' => 'dummytask',
+				'data' => null
+			]
+		];
+
+		foreach ($expected as $item) {
+			$this->QueuedTasks->clearKey();
+			$tmp = $this->QueuedTasks->requestJob($capabilities);
+
+			$this->assertSame($item['name'], $tmp['task']);
+			$this->assertEquals($item['data'], unserialize($tmp['data']));
+		}
+	}
+
+	/**
+	 * Job Rate limiting.
+	 * Do not execute jobs of a certain type more often than once every X seconds.
+	 *
+	 * @return void
+	 */
+	public function testRateLimit()
+	{
+		$this->_needsConnection();
+
+		$capabilities = [
+			'task1' => [
+				'name' => 'task1',
+				'timeout' => 101,
+				'retries' => 2,
+				'rate' => 2
+			],
+			'dummytask' => [
+				'name' => 'dummytask',
+				'timeout' => 101,
+				'retries' => 2
+			]
+		];
+
+		// clear out the rate history
+		$this->QueuedTasks->rateHistory = [];
+
+		$data1 = [
+			'key' => 1
+		];
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $data1));
+		$data2 = [
+			'key' => 2
+		];
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $data2));
+		$data3 = [
+			'key' => 3
+		];
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('task1', $data3));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
+		$this->assertTrue((bool)$this->QueuedTasks->createJob('dummytask'));
+
+		// At first we get task1-1.
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame('task1', $tmp['task']);
+		$this->assertSame($data1, unserialize($tmp['data']));
+
+		// The rate limit should now skip over task1-2 and fetch a dummytask.
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame('dummytask', $tmp['task']);
+		$this->assertFalse(unserialize($tmp['data']));
+
+		usleep(100000);
+		// and again.
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame('dummytask', $tmp['task']);
+		$this->assertFalse(unserialize($tmp['data']));
+
+		// Then some time passes
+		sleep(2);
+
+		// Now we should get task1-2
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame('task1', $tmp['task']);
+		$this->assertSame($data2, unserialize($tmp['data']));
+
+		// and again rate limit to dummytask.
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame('dummytask', $tmp['task']);
+		$this->assertFalse(unserialize($tmp['data']));
+
+		// Then some more time passes
+		sleep(2);
+
+		// Now we should get task1-3
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame('task1', $tmp['task']);
+		$this->assertSame($data3, unserialize($tmp['data']));
+
+		// and again rate limit to dummytask.
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertSame('dummytask', $tmp['task']);
+		$this->assertFalse(unserialize($tmp['data']));
+
+		// and now the queue is empty
+		$this->QueuedTasks->clearKey();
+		$tmp = $this->QueuedTasks->requestJob($capabilities);
+		$this->assertNull($tmp);
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testIsQueued()
+	{
+		$result = $this->QueuedTasks->isQueued('foo-bar');
+		$this->assertFalse($result);
+
+		$queuedJob = $this->QueuedTasks->newEntity([
+			'key' => 'key',
+			'task' => 'FooBar'
+		]);
+		$this->QueuedTasks->saveOrFail($queuedJob);
+
+		$result = $this->QueuedTasks->isQueued('foo-bar');
+		$this->assertTrue($result);
+
+		$queuedJob->completed = new FrozenTime();
+		$this->QueuedTasks->saveOrFail($queuedJob);
+
+		$result = $this->QueuedTasks->isQueued('foo-bar');
+		$this->assertFalse($result);
+	}
+
+	/**
+	 * Helper method for skipping tests that need a real connection.
+	 *
+	 * @return void
+	 */
+	protected function _needsConnection()
+	{
+		$config = ConnectionManager::getConfig('test');
+		$this->skipIf(strpos($config['driver'], 'Mysql') === false, 'Only Mysql is working yet for this.');
+	}
 }

tests/Fixture/QueuedTasksFixture.php

Indentation +41 added lines, -41 removed lines

@@ -9,46 +9,46 @@
 class QueuedTasksFixture extends TestFixture
 {
 
-    /**
-     * Fields
-     *
-     * @var array
-     */
-    // @codingStandardsIgnoreStart
-    public $fields = [
-        'id' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => null, 'comment' => '', 'autoIncrement' => true, 'precision' => null],
-        'task' => ['type' => 'string', 'length' => 255, 'null' => false, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null, 'fixed' => null],
-        'data' => ['type' => 'text', 'length' => 16777215, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
-        'not_before' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
-        'fetched' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
-        'completed' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
-        'failed_count' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => '0', 'comment' => '', 'precision' => null, 'autoIncrement' => null],
-        'failure_message' => ['type' => 'text', 'length' => null, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
-        'worker_key' => ['type' => 'string', 'fixed' => true, 'length' => 40, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
-        'created' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
-        '_indexes' => [
-            'completed' => ['type' => 'index', 'columns' => ['completed'], 'length' => []],
-            'worker_key' => ['type' => 'index', 'columns' => ['worker_key'], 'length' => []],
-            'task' => ['type' => 'index', 'columns' => ['task'], 'length' => []],
-        ],
-        '_constraints' => [
-            'primary' => ['type' => 'primary', 'columns' => ['id'], 'length' => []],
-        ],
-        '_options' => [
-            'engine' => 'InnoDB',
-            'collation' => 'utf8_general_ci'
-        ],
-    ];
-    // @codingStandardsIgnoreEnd
+	/**
+	 * Fields
+	 *
+	 * @var array
+	 */
+	// @codingStandardsIgnoreStart
+	public $fields = [
+		'id' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => null, 'comment' => '', 'autoIncrement' => true, 'precision' => null],
+		'task' => ['type' => 'string', 'length' => 255, 'null' => false, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null, 'fixed' => null],
+		'data' => ['type' => 'text', 'length' => 16777215, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
+		'not_before' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
+		'fetched' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
+		'completed' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
+		'failed_count' => ['type' => 'integer', 'length' => 10, 'unsigned' => true, 'null' => false, 'default' => '0', 'comment' => '', 'precision' => null, 'autoIncrement' => null],
+		'failure_message' => ['type' => 'text', 'length' => null, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
+		'worker_key' => ['type' => 'string', 'fixed' => true, 'length' => 40, 'null' => true, 'default' => null, 'collate' => 'utf8_general_ci', 'comment' => '', 'precision' => null],
+		'created' => ['type' => 'timestamp', 'length' => null, 'null' => true, 'default' => null, 'comment' => '', 'precision' => null],
+		'_indexes' => [
+			'completed' => ['type' => 'index', 'columns' => ['completed'], 'length' => []],
+			'worker_key' => ['type' => 'index', 'columns' => ['worker_key'], 'length' => []],
+			'task' => ['type' => 'index', 'columns' => ['task'], 'length' => []],
+		],
+		'_constraints' => [
+			'primary' => ['type' => 'primary', 'columns' => ['id'], 'length' => []],
+		],
+		'_options' => [
+			'engine' => 'InnoDB',
+			'collation' => 'utf8_general_ci'
+		],
+	];
+	// @codingStandardsIgnoreEnd
 
-    /**
-     * Init method
-     *
-     * @return void
-     */
-    public function init()
-    {
-        $this->records = [];
-        parent::init();
-    }
+	/**
+	 * Init method
+	 *
+	 * @return void
+	 */
+	public function init()
+	{
+		$this->records = [];
+		parent::init();
+	}
 }

src/Shell/QueueShell.php

Spacing +1 added lines, -1 removed lines

@@ -18,7 +18,7 @@
 use RuntimeException;
 use Throwable;
 
-declare(ticks = 1);
+declare(ticks=1);
 
 /**
  * Main shell to init and run queue workers.

Indentation +562 added lines, -562 removed lines

@@ -31,578 +31,578 @@
 class QueueShell extends Shell
 {
 
-    /**
-     *
-     * @var string
-     */
-    public $modelClass = 'Queue.QueuedTasks';
-
-    /**
-     *
-     * @var array|null
-     */
-    protected $_taskConf;
-
-    /**
-     *
-     * @var int
-     */
-    protected $_time = 0;
-
-    /**
-     *
-     * @var bool
-     */
-    protected $_exit = false;
-
-    /**
-     * Overwrite shell initialize to dynamically load all Queue Related Tasks.
-     *
-     * @return void
-     */
-    public function initialize()
-    {
-        $taskFinder = new TaskFinder();
-        $this->tasks = $taskFinder->allAppAndPluginTasks();
-
-        parent::initialize();
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function startup()
-    {
-        if ($this->param('quiet')) {
-            $this->interactive = false;
-        }
-
-        parent::startup();
-    }
-
-    /**
-     *
-     * @return string
-     */
-    public function getDescription()
-    {
-        $tasks = [];
-        foreach ($this->taskNames as $loadedTask) {
-            $tasks[] = "\t" . '* ' . $this->_taskName($loadedTask);
-        }
-        $tasks = implode(PHP_EOL, $tasks);
-
-        $text = <<<TEXT
+	/**
+	 *
+	 * @var string
+	 */
+	public $modelClass = 'Queue.QueuedTasks';
+
+	/**
+	 *
+	 * @var array|null
+	 */
+	protected $_taskConf;
+
+	/**
+	 *
+	 * @var int
+	 */
+	protected $_time = 0;
+
+	/**
+	 *
+	 * @var bool
+	 */
+	protected $_exit = false;
+
+	/**
+	 * Overwrite shell initialize to dynamically load all Queue Related Tasks.
+	 *
+	 * @return void
+	 */
+	public function initialize()
+	{
+		$taskFinder = new TaskFinder();
+		$this->tasks = $taskFinder->allAppAndPluginTasks();
+
+		parent::initialize();
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function startup()
+	{
+		if ($this->param('quiet')) {
+			$this->interactive = false;
+		}
+
+		parent::startup();
+	}
+
+	/**
+	 *
+	 * @return string
+	 */
+	public function getDescription()
+	{
+		$tasks = [];
+		foreach ($this->taskNames as $loadedTask) {
+			$tasks[] = "\t" . '* ' . $this->_taskName($loadedTask);
+		}
+		$tasks = implode(PHP_EOL, $tasks);
+
+		$text = <<<TEXT
 Simple and minimalistic job queue (or deferred-task) system.
 
 Available Tasks:
 $tasks
 TEXT;
 
-        return $text;
-    }
-
-    /**
-     * Look for a Queue Task of hte passed name and try to call add() on it.
-     * A QueueTask may provide an add function to enable the user to create new jobs via commandline.
-     *
-     * @return void
-     */
-    public function add()
-    {
-        if (count($this->args) < 1) {
-            $this->out('Please call like this:');
-            $this->out('       bin/cake queue add <taskname>');
-            $this->_displayAvailableTasks();
-
-            return;
-        }
-
-        $name = Inflector::camelize($this->args[0]);
-        if (in_array('Queue' . $name, $this->taskNames, true)) {
-            /** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */
-            $task = $this->{'Queue' . $name};
-            if (!($task instanceof AddInterface)) {
-                $this->abort('This task does not support adding via CLI call');
-            }
-            $task->add();
-        } else {
-            $this->out('Error: Task not found: ' . $name);
-            $this->_displayAvailableTasks();
-        }
-    }
-
-    /**
-     * Output the task without Queue or Task
-     * example: QueueImageTask becomes Image on display
-     *
-     * @param string $task Task name
-     * @return string Cleaned task name
-     */
-    protected function _taskName($task)
-    {
-        if (strpos($task, 'Queue') === 0) {
-            return substr($task, 5);
-        }
-
-        return $task;
-    }
-
-    /**
-     * Run a QueueWorker loop.
-     * Runs a Queue Worker process which will try to find unassigned jobs in the queue
-     * which it may run and try to fetch and execute them.
-     *
-     * @return int|null
-     */
-    public function runworker()
-    {
-        try {
-            $pid = $this->_initPid();
-        } catch (PersistenceFailedException $exception) {
-            $this->err($exception->getMessage());
-            $limit = (int)Configure::read('Queue.maxWorkers');
-            if ($limit) {
-                $this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')');
-            }
-
-            return static::CODE_ERROR;
-        }
-
-        // Enable Garbage Collector (PHP >= 5.3)
-        if (function_exists('gc_enable')) {
-            gc_enable();
-        }
-        if (function_exists('pcntl_signal')) {
-            pcntl_signal(SIGTERM, [
-                &$this,
-                '_exit'
-            ]);
-            pcntl_signal(SIGINT, [
-                &$this,
-                '_exit'
-            ]);
-            pcntl_signal(SIGTSTP, [
-                &$this,
-                '_exit'
-            ]);
-            pcntl_signal(SIGQUIT, [
-                &$this,
-                '_exit'
-            ]);
-        }
-        $this->_exit = false;
-
-        $startTime = time();
-        $types = $this->_stringToArray($this->param('type'));
-
-        while (!$this->_exit) {
-            $this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE);
-
-            $QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types);
-
-            if ($QueuedTask) {
-                $this->runJob($QueuedTask, $pid);
-            } elseif (Configure::read('Queue.exitWhenNothingToDo')) {
-                $this->out('nothing to do, exiting.');
-                $this->_exit = true;
-            } else {
-                $this->out('nothing to do, sleeping.');
-                sleep(Config::sleepTime());
-            }
-
-            // check if we are over the maximum runtime and end processing if so.
-            if (Configure::readOrFail('Queue.workerMaxRuntime') && (time() - $startTime) >= Configure::readOrFail('Queue.workerMaxRuntime')) {
-                $this->_exit = true;
-                $this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.');
-            }
-            if ($this->_exit || mt_rand(0, 100) > (100 - (int)Config::gcprob())) {
-                $this->out(__d('queue', 'Performing old job cleanup.'));
-                $this->QueuedTasks->cleanOldJobs();
-            }
-            $this->hr();
-        }
-
-        $this->_deletePid($pid);
-
-        if ($this->param('verbose')) {
-            $this->_log('endworker', $pid);
-        }
-    }
-
-    /**
-     *
-     * @param \Queue\Model\Entity\QueuedTask $QueuedTask Queued task
-     * @param string $pid PID of the process
-     * @return void
-     */
-    protected function runJob(QueuedTask $QueuedTask, $pid)
-    {
-        $this->out('Running Job of type "' . $QueuedTask->task . '"');
-        $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id, $pid, false);
-        $taskName = 'Queue' . $QueuedTask->task;
-
-        try {
-            $this->_time = time();
-
-            $data = unserialize($QueuedTask->data);
-            /** @var \Queue\Shell\Task\QueueTask $task */
-            $task = $this->{$taskName};
-            if (!$task instanceof QueueTaskInterface) {
-                throw new RuntimeException('Task must implement ' . QueueTaskInterface::class);
-            }
-
-            $return = $task->run((array)$data, $QueuedTask->id);
-            if ($return !== null) {
-                trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED);
-            }
-            $failureMessage = $taskName . ' failed';
-        } catch (Throwable $e) {
-            $return = false;
-
-            $failureMessage = get_class($e) . ': ' . $e->getMessage();
-            if (!($e instanceof QueueException)) {
-                $failureMessage .= "\n" . $e->getTraceAsString();
-            }
-
-            $this->_logError($taskName . ' (job ' . $QueuedTask->id . ')' . "\n" . $failureMessage, $pid);
-        } catch (Exception $e) {
-            $return = false;
-
-            $failureMessage = get_class($e) . ': ' . $e->getMessage();
-            $this->_logError($taskName . "\n" . $failureMessage, $pid);
-        }
-
-        if ($return === false) {
-            $this->QueuedTasks->markJobFailed($QueuedTask, $failureMessage);
-            $failedStatus = $this->QueuedTasks->getFailedStatus($QueuedTask, $this->_getTaskConf());
-            $this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id . ' failed and ' . $failedStatus, $pid);
-            $this->out('Job did not finish, ' . $failedStatus . ' after try ' . $QueuedTask->failed . '.');
-
-            return;
-        }
-
-        $this->QueuedTasks->markJobDone($QueuedTask);
-        $this->out('Job Finished.');
-    }
-
-    /**
-     * Manually trigger a Finished job cleanup.
-     *
-     * @return void
-     */
-    public function clean()
-    {
-        if (!Configure::read('Queue.cleanupTimeout')) {
-            $this->abort('You disabled cleanuptimout in config. Aborting.');
-        }
-
-        $this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int)Configure::read('Queue.cleanupTimeout')));
-        $this->QueuedTasks->cleanOldJobs();
-    }
-
-    /**
-     * Display current settings
-     *
-     * @return void
-     */
-    public function settings()
-    {
-        $this->out('Current Settings:');
-        $conf = (array)Configure::read('Queue');
-        foreach ($conf as $key => $val) {
-            if ($val === false) {
-                $val = 'no';
-            }
-            if ($val === true) {
-                $val = 'yes';
-            }
-            $this->out('* ' . $key . ': ' . print_r($val, true));
-        }
-
-        $this->out();
-    }
-
-    /**
-     * Display some statistics about Finished Jobs.
-     *
-     * @return void
-     */
-    public function stats()
-    {
-        $this->out('Jobs currently in the queue:');
-
-        $types = $this->QueuedTasks->getTypes()->toArray();
-        foreach ($types as $type) {
-            $this->out('      ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedTasks->getLength($type));
-        }
-        $this->hr();
-        $this->out('Total unfinished jobs: ' . $this->QueuedTasks->getLength());
-        $this->hr();
-        $this->out('Finished job statistics:');
-        $data = $this->QueuedTasks->getStats();
-        foreach ($data as $item) {
-            $this->out(' ' . $item['task'] . ': ');
-            $this->out('   Finished Jobs in Database: ' . $item['num']);
-            $this->out('   Average Job existence    : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's');
-            $this->out('   Average Execution delay  : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's');
-            $this->out('   Average Execution time   : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's');
-        }
-    }
-
-    /**
-     * Get option parser method to parse commandline options
-     *
-     * @return \Cake\Console\ConsoleOptionParser
-     */
-    public function getOptionParser()
-    {
-        $subcommandParser = [
-            'options' => [
-                /*
+		return $text;
+	}
+
+	/**
+	 * Look for a Queue Task of hte passed name and try to call add() on it.
+	 * A QueueTask may provide an add function to enable the user to create new jobs via commandline.
+	 *
+	 * @return void
+	 */
+	public function add()
+	{
+		if (count($this->args) < 1) {
+			$this->out('Please call like this:');
+			$this->out('       bin/cake queue add <taskname>');
+			$this->_displayAvailableTasks();
+
+			return;
+		}
+
+		$name = Inflector::camelize($this->args[0]);
+		if (in_array('Queue' . $name, $this->taskNames, true)) {
+			/** @var \Queue\Shell\Task\QueueTask|\Queue\Shell\Task\AddInterface $task */
+			$task = $this->{'Queue' . $name};
+			if (!($task instanceof AddInterface)) {
+				$this->abort('This task does not support adding via CLI call');
+			}
+			$task->add();
+		} else {
+			$this->out('Error: Task not found: ' . $name);
+			$this->_displayAvailableTasks();
+		}
+	}
+
+	/**
+	 * Output the task without Queue or Task
+	 * example: QueueImageTask becomes Image on display
+	 *
+	 * @param string $task Task name
+	 * @return string Cleaned task name
+	 */
+	protected function _taskName($task)
+	{
+		if (strpos($task, 'Queue') === 0) {
+			return substr($task, 5);
+		}
+
+		return $task;
+	}
+
+	/**
+	 * Run a QueueWorker loop.
+	 * Runs a Queue Worker process which will try to find unassigned jobs in the queue
+	 * which it may run and try to fetch and execute them.
+	 *
+	 * @return int|null
+	 */
+	public function runworker()
+	{
+		try {
+			$pid = $this->_initPid();
+		} catch (PersistenceFailedException $exception) {
+			$this->err($exception->getMessage());
+			$limit = (int)Configure::read('Queue.maxWorkers');
+			if ($limit) {
+				$this->out('Cannot start worker: Too many workers already/still running on this server (' . $limit . '/' . $limit . ')');
+			}
+
+			return static::CODE_ERROR;
+		}
+
+		// Enable Garbage Collector (PHP >= 5.3)
+		if (function_exists('gc_enable')) {
+			gc_enable();
+		}
+		if (function_exists('pcntl_signal')) {
+			pcntl_signal(SIGTERM, [
+				&$this,
+				'_exit'
+			]);
+			pcntl_signal(SIGINT, [
+				&$this,
+				'_exit'
+			]);
+			pcntl_signal(SIGTSTP, [
+				&$this,
+				'_exit'
+			]);
+			pcntl_signal(SIGQUIT, [
+				&$this,
+				'_exit'
+			]);
+		}
+		$this->_exit = false;
+
+		$startTime = time();
+		$types = $this->_stringToArray($this->param('type'));
+
+		while (!$this->_exit) {
+			$this->out(__d('queue', 'Looking for a job.'), 1, Shell::VERBOSE);
+
+			$QueuedTask = $this->QueuedTasks->requestJob($this->_getTaskConf(), $types);
+
+			if ($QueuedTask) {
+				$this->runJob($QueuedTask, $pid);
+			} elseif (Configure::read('Queue.exitWhenNothingToDo')) {
+				$this->out('nothing to do, exiting.');
+				$this->_exit = true;
+			} else {
+				$this->out('nothing to do, sleeping.');
+				sleep(Config::sleepTime());
+			}
+
+			// check if we are over the maximum runtime and end processing if so.
+			if (Configure::readOrFail('Queue.workerMaxRuntime') && (time() - $startTime) >= Configure::readOrFail('Queue.workerMaxRuntime')) {
+				$this->_exit = true;
+				$this->out('queue', 'Reached runtime of ' . (time() - $startTime) . ' Seconds (Max ' . Configure::readOrFail('Queue.workerMaxRuntime') . '), terminating.');
+			}
+			if ($this->_exit || mt_rand(0, 100) > (100 - (int)Config::gcprob())) {
+				$this->out(__d('queue', 'Performing old job cleanup.'));
+				$this->QueuedTasks->cleanOldJobs();
+			}
+			$this->hr();
+		}
+
+		$this->_deletePid($pid);
+
+		if ($this->param('verbose')) {
+			$this->_log('endworker', $pid);
+		}
+	}
+
+	/**
+	 *
+	 * @param \Queue\Model\Entity\QueuedTask $QueuedTask Queued task
+	 * @param string $pid PID of the process
+	 * @return void
+	 */
+	protected function runJob(QueuedTask $QueuedTask, $pid)
+	{
+		$this->out('Running Job of type "' . $QueuedTask->task . '"');
+		$this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id, $pid, false);
+		$taskName = 'Queue' . $QueuedTask->task;
+
+		try {
+			$this->_time = time();
+
+			$data = unserialize($QueuedTask->data);
+			/** @var \Queue\Shell\Task\QueueTask $task */
+			$task = $this->{$taskName};
+			if (!$task instanceof QueueTaskInterface) {
+				throw new RuntimeException('Task must implement ' . QueueTaskInterface::class);
+			}
+
+			$return = $task->run((array)$data, $QueuedTask->id);
+			if ($return !== null) {
+				trigger_error('run() should be void and throw exception in error case now.', E_USER_DEPRECATED);
+			}
+			$failureMessage = $taskName . ' failed';
+		} catch (Throwable $e) {
+			$return = false;
+
+			$failureMessage = get_class($e) . ': ' . $e->getMessage();
+			if (!($e instanceof QueueException)) {
+				$failureMessage .= "\n" . $e->getTraceAsString();
+			}
+
+			$this->_logError($taskName . ' (job ' . $QueuedTask->id . ')' . "\n" . $failureMessage, $pid);
+		} catch (Exception $e) {
+			$return = false;
+
+			$failureMessage = get_class($e) . ': ' . $e->getMessage();
+			$this->_logError($taskName . "\n" . $failureMessage, $pid);
+		}
+
+		if ($return === false) {
+			$this->QueuedTasks->markJobFailed($QueuedTask, $failureMessage);
+			$failedStatus = $this->QueuedTasks->getFailedStatus($QueuedTask, $this->_getTaskConf());
+			$this->_log('job ' . $QueuedTask->task . ', id ' . $QueuedTask->id . ' failed and ' . $failedStatus, $pid);
+			$this->out('Job did not finish, ' . $failedStatus . ' after try ' . $QueuedTask->failed . '.');
+
+			return;
+		}
+
+		$this->QueuedTasks->markJobDone($QueuedTask);
+		$this->out('Job Finished.');
+	}
+
+	/**
+	 * Manually trigger a Finished job cleanup.
+	 *
+	 * @return void
+	 */
+	public function clean()
+	{
+		if (!Configure::read('Queue.cleanupTimeout')) {
+			$this->abort('You disabled cleanuptimout in config. Aborting.');
+		}
+
+		$this->out('Deleting old jobs, that have finished before ' . date('Y-m-d H:i:s', time() - (int)Configure::read('Queue.cleanupTimeout')));
+		$this->QueuedTasks->cleanOldJobs();
+	}
+
+	/**
+	 * Display current settings
+	 *
+	 * @return void
+	 */
+	public function settings()
+	{
+		$this->out('Current Settings:');
+		$conf = (array)Configure::read('Queue');
+		foreach ($conf as $key => $val) {
+			if ($val === false) {
+				$val = 'no';
+			}
+			if ($val === true) {
+				$val = 'yes';
+			}
+			$this->out('* ' . $key . ': ' . print_r($val, true));
+		}
+
+		$this->out();
+	}
+
+	/**
+	 * Display some statistics about Finished Jobs.
+	 *
+	 * @return void
+	 */
+	public function stats()
+	{
+		$this->out('Jobs currently in the queue:');
+
+		$types = $this->QueuedTasks->getTypes()->toArray();
+		foreach ($types as $type) {
+			$this->out('      ' . str_pad($type, 20, ' ', STR_PAD_RIGHT) . ': ' . $this->QueuedTasks->getLength($type));
+		}
+		$this->hr();
+		$this->out('Total unfinished jobs: ' . $this->QueuedTasks->getLength());
+		$this->hr();
+		$this->out('Finished job statistics:');
+		$data = $this->QueuedTasks->getStats();
+		foreach ($data as $item) {
+			$this->out(' ' . $item['task'] . ': ');
+			$this->out('   Finished Jobs in Database: ' . $item['num']);
+			$this->out('   Average Job existence    : ' . str_pad(Number::precision($item['alltime']), 8, ' ', STR_PAD_LEFT) . 's');
+			$this->out('   Average Execution delay  : ' . str_pad(Number::precision($item['fetchdelay']), 8, ' ', STR_PAD_LEFT) . 's');
+			$this->out('   Average Execution time   : ' . str_pad(Number::precision($item['runtime']), 8, ' ', STR_PAD_LEFT) . 's');
+		}
+	}
+
+	/**
+	 * Get option parser method to parse commandline options
+	 *
+	 * @return \Cake\Console\ConsoleOptionParser
+	 */
+	public function getOptionParser()
+	{
+		$subcommandParser = [
+			'options' => [
+				/*
                  * 'dry-run'=> array(
                  * 'short' => 'd',
                  * 'help' => 'Dry run the update, no jobs will actually be added.',
                  * 'boolean' => true
                  * ),
                  */
-            ]
-        ];
-        $subcommandParserFull = $subcommandParser;
-        $subcommandParserFull['options']['type'] = [
-            'short' => 't',
-            'help' => 'Type (comma separated list possible)',
-            'default' => null
-        ];
-
-        return parent::getOptionParser()->setDescription($this->getDescription())
-            ->addSubcommand('clean', [
-                'help' => 'Remove old jobs (cleanup)',
-                'parser' => $subcommandParser
-            ])
-            ->addSubcommand('add', [
-                'help' => 'Add Job',
-                'parser' => $subcommandParser
-            ])
-            ->addSubcommand('stats', [
-                'help' => 'Stats',
-                'parser' => $subcommandParserFull
-            ])
-            ->addSubcommand('settings', [
-                'help' => 'Settings',
-                'parser' => $subcommandParserFull
-            ])
-            ->addSubcommand('runworker', [
-                'help' => 'Run Worker',
-                'parser' => $subcommandParserFull
-            ]);
-    }
-
-    /**
-     * Timestamped log.
-     *
-     * @param string $message Log type
-     * @param string|null $pid PID of the process
-     * @param bool $addDetails Details
-     * @return void
-     */
-    protected function _log($message, $pid = null, $addDetails = true)
-    {
-        if (!Configure::read('Queue.log')) {
-            return;
-        }
-
-        if ($addDetails) {
-            $timeNeeded = $this->_timeNeeded();
-            $memoryUsage = $this->_memoryUsage();
-            $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
-        }
-
-        if ($pid) {
-            $message .= ' (pid ' . $pid . ')';
-        }
-        Log::write('info', $message, [
-            'scope' => 'queue'
-        ]);
-    }
-
-    /**
-     *
-     * @param string $message Message
-     * @param string|null $pid PID of the process
-     * @return void
-     */
-    protected function _logError($message, $pid = null)
-    {
-        $timeNeeded = $this->_timeNeeded();
-        $memoryUsage = $this->_memoryUsage();
-        $message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
-
-        if ($pid) {
-            $message .= ' (pid ' . $pid . ')';
-        }
-
-        Log::write('error', $message);
-    }
-
-    /**
-     * Returns a List of available QueueTasks and their individual configurations.
-     *
-     * @return array
-     */
-    protected function _getTaskConf()
-    {
-        if (!is_array($this->_taskConf)) {
-            $this->_taskConf = [];
-            foreach ($this->tasks as $task) {
-                list ($pluginName, $taskName) = pluginSplit($task);
-
-                $this->_taskConf[$taskName]['name'] = substr($taskName, 5);
-                $this->_taskConf[$taskName]['plugin'] = $pluginName;
-                if (property_exists($this->{$taskName}, 'timeout')) {
-                    $this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout;
-                } else {
-                    $this->_taskConf[$taskName]['timeout'] = Config::defaultWorkerTimeout();
-                }
-                if (property_exists($this->{$taskName}, 'retries')) {
-                    $this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries;
-                } else {
-                    $this->_taskConf[$taskName]['retries'] = Config::defaultWorkerRetries();
-                }
-                if (property_exists($this->{$taskName}, 'cleanupTimeout')) {
-                    $this->_taskConf[$taskName]['cleanupTimeout'] = $this->{$taskName}->cleanupTimeout;
-                } else {
-                    $this->_taskConf[$taskName]['cleanupTimeout'] = Config::cleanupTimeout();
-                }
-            }
-        }
-
-        return $this->_taskConf;
-    }
-
-    /**
-     * Signal handling to queue worker for clean shutdown
-     *
-     * @param int $signal The signal
-     * @return void
-     */
-    protected function _exit($signal)
-    {
-        $this->out(__d('queue', 'Caught %d signal, exiting.', $signal));
-        $this->_exit = true;
-    }
-
-    /**
-     *
-     * @return void
-     */
-    protected function _displayAvailableTasks()
-    {
-        $this->out('Available Tasks:');
-        foreach ($this->taskNames as $loadedTask) {
-            $this->out("\t" . '* ' . $this->_taskName($loadedTask));
-        }
-    }
-
-    /**
-     *
-     * @return string
-     */
-    protected function _initPid()
-    {
-        $this->_pid = $this->_retrievePid();
-
-        return $this->_pid;
-    }
-
-    /**
-     *
-     * @return string
-     */
-    protected function _retrievePid()
-    {
-        if (function_exists('posix_getpid')) {
-            $pid = (string)posix_getpid();
-        } else {
-            $pid = $this->QueuedTasks->key();
-        }
-
-        return $pid;
-    }
-
-    /**
-     *
-     * @return string Memory usage in MB.
-     */
-    protected function _memoryUsage()
-    {
-        $limit = ini_get('memory_limit');
-
-        $used = number_format(memory_get_peak_usage(true) / (1024 * 1024), 0) . 'MB';
-        if ($limit !== '-1') {
-            $used .= '/' . $limit;
-        }
-
-        return $used;
-    }
-
-    /**
-     *
-     * @param string|null $pid PID of the process
-     *
-     * @return void
-     */
-    protected function _deletePid($pid)
-    {
-        if (!$pid) {
-            $pid = $this->_pid;
-        }
-        if (!$pid) {
-            return;
-        }
-    }
-
-    /**
-     *
-     * @return string
-     */
-    protected function _timeNeeded()
-    {
-        $diff = $this->_time() - $this->_time($this->_time);
-        $seconds = max($diff, 1);
-
-        return $seconds . 's';
-    }
-
-    /**
-     *
-     * @param int|null $providedTime Provided time
-     *
-     * @return int
-     */
-    protected function _time($providedTime = null)
-    {
-        if ($providedTime) {
-            return $providedTime;
-        }
-
-        return time();
-    }
-
-    /**
-     *
-     * @param string|null $param String to convert
-     * @return array
-     */
-    protected function _stringToArray($param)
-    {
-        if (!$param) {
-            return [];
-        }
-
-        $array = Text::tokenize($param);
-
-        return array_filter($array);
-    }
+			]
+		];
+		$subcommandParserFull = $subcommandParser;
+		$subcommandParserFull['options']['type'] = [
+			'short' => 't',
+			'help' => 'Type (comma separated list possible)',
+			'default' => null
+		];
+
+		return parent::getOptionParser()->setDescription($this->getDescription())
+			->addSubcommand('clean', [
+				'help' => 'Remove old jobs (cleanup)',
+				'parser' => $subcommandParser
+			])
+			->addSubcommand('add', [
+				'help' => 'Add Job',
+				'parser' => $subcommandParser
+			])
+			->addSubcommand('stats', [
+				'help' => 'Stats',
+				'parser' => $subcommandParserFull
+			])
+			->addSubcommand('settings', [
+				'help' => 'Settings',
+				'parser' => $subcommandParserFull
+			])
+			->addSubcommand('runworker', [
+				'help' => 'Run Worker',
+				'parser' => $subcommandParserFull
+			]);
+	}
+
+	/**
+	 * Timestamped log.
+	 *
+	 * @param string $message Log type
+	 * @param string|null $pid PID of the process
+	 * @param bool $addDetails Details
+	 * @return void
+	 */
+	protected function _log($message, $pid = null, $addDetails = true)
+	{
+		if (!Configure::read('Queue.log')) {
+			return;
+		}
+
+		if ($addDetails) {
+			$timeNeeded = $this->_timeNeeded();
+			$memoryUsage = $this->_memoryUsage();
+			$message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
+		}
+
+		if ($pid) {
+			$message .= ' (pid ' . $pid . ')';
+		}
+		Log::write('info', $message, [
+			'scope' => 'queue'
+		]);
+	}
+
+	/**
+	 *
+	 * @param string $message Message
+	 * @param string|null $pid PID of the process
+	 * @return void
+	 */
+	protected function _logError($message, $pid = null)
+	{
+		$timeNeeded = $this->_timeNeeded();
+		$memoryUsage = $this->_memoryUsage();
+		$message .= ' [' . $timeNeeded . ', ' . $memoryUsage . ']';
+
+		if ($pid) {
+			$message .= ' (pid ' . $pid . ')';
+		}
+
+		Log::write('error', $message);
+	}
+
+	/**
+	 * Returns a List of available QueueTasks and their individual configurations.
+	 *
+	 * @return array
+	 */
+	protected function _getTaskConf()
+	{
+		if (!is_array($this->_taskConf)) {
+			$this->_taskConf = [];
+			foreach ($this->tasks as $task) {
+				list ($pluginName, $taskName) = pluginSplit($task);
+
+				$this->_taskConf[$taskName]['name'] = substr($taskName, 5);
+				$this->_taskConf[$taskName]['plugin'] = $pluginName;
+				if (property_exists($this->{$taskName}, 'timeout')) {
+					$this->_taskConf[$taskName]['timeout'] = $this->{$taskName}->timeout;
+				} else {
+					$this->_taskConf[$taskName]['timeout'] = Config::defaultWorkerTimeout();
+				}
+				if (property_exists($this->{$taskName}, 'retries')) {
+					$this->_taskConf[$taskName]['retries'] = $this->{$taskName}->retries;
+				} else {
+					$this->_taskConf[$taskName]['retries'] = Config::defaultWorkerRetries();
+				}
+				if (property_exists($this->{$taskName}, 'cleanupTimeout')) {
+					$this->_taskConf[$taskName]['cleanupTimeout'] = $this->{$taskName}->cleanupTimeout;
+				} else {
+					$this->_taskConf[$taskName]['cleanupTimeout'] = Config::cleanupTimeout();
+				}
+			}
+		}
+
+		return $this->_taskConf;
+	}
+
+	/**
+	 * Signal handling to queue worker for clean shutdown
+	 *
+	 * @param int $signal The signal
+	 * @return void
+	 */
+	protected function _exit($signal)
+	{
+		$this->out(__d('queue', 'Caught %d signal, exiting.', $signal));
+		$this->_exit = true;
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	protected function _displayAvailableTasks()
+	{
+		$this->out('Available Tasks:');
+		foreach ($this->taskNames as $loadedTask) {
+			$this->out("\t" . '* ' . $this->_taskName($loadedTask));
+		}
+	}
+
+	/**
+	 *
+	 * @return string
+	 */
+	protected function _initPid()
+	{
+		$this->_pid = $this->_retrievePid();
+
+		return $this->_pid;
+	}
+
+	/**
+	 *
+	 * @return string
+	 */
+	protected function _retrievePid()
+	{
+		if (function_exists('posix_getpid')) {
+			$pid = (string)posix_getpid();
+		} else {
+			$pid = $this->QueuedTasks->key();
+		}
+
+		return $pid;
+	}
+
+	/**
+	 *
+	 * @return string Memory usage in MB.
+	 */
+	protected function _memoryUsage()
+	{
+		$limit = ini_get('memory_limit');
+
+		$used = number_format(memory_get_peak_usage(true) / (1024 * 1024), 0) . 'MB';
+		if ($limit !== '-1') {
+			$used .= '/' . $limit;
+		}
+
+		return $used;
+	}
+
+	/**
+	 *
+	 * @param string|null $pid PID of the process
+	 *
+	 * @return void
+	 */
+	protected function _deletePid($pid)
+	{
+		if (!$pid) {
+			$pid = $this->_pid;
+		}
+		if (!$pid) {
+			return;
+		}
+	}
+
+	/**
+	 *
+	 * @return string
+	 */
+	protected function _timeNeeded()
+	{
+		$diff = $this->_time() - $this->_time($this->_time);
+		$seconds = max($diff, 1);
+
+		return $seconds . 's';
+	}
+
+	/**
+	 *
+	 * @param int|null $providedTime Provided time
+	 *
+	 * @return int
+	 */
+	protected function _time($providedTime = null)
+	{
+		if ($providedTime) {
+			return $providedTime;
+		}
+
+		return time();
+	}
+
+	/**
+	 *
+	 * @param string|null $param String to convert
+	 * @return array
+	 */
+	protected function _stringToArray($param)
+	{
+		if (!$param) {
+			return [];
+		}
+
+		$array = Text::tokenize($param);
+
+		return array_filter($array);
+	}
 }