Oefenweb / cakephp-queue

tests/TestCase/Shell/QueueShellTest.php

Indentation +183 added lines, -183 removed lines

@@ -11,187 +11,187 @@
 
 class QueueShellTest extends TestCase
 {
-    use ToolsTestTrait;
-
-    /**
-     *
-     * @var \Queue\Shell\QueueShell|\PHPUnit_Framework_MockObject_MockObject
-     */
-    public $QueueShell;
-
-    /**
-     *
-     * @var \Tools\TestSuite\ConsoleOutput
-     */
-    public $out;
-
-    /**
-     *
-     * @var \Tools\TestSuite\ConsoleOutput
-     */
-    public $err;
-
-    /**
-     * Fixtures to load
-     *
-     * @var array
-     */
-    public $fixtures = [
-        'plugin.Queue.QueuedTasks'
-    ];
-
-    /**
-     * Setup Defaults
-     *
-     * @return void
-     */
-    public function setUp()
-    {
-        parent::setUp();
-
-        $this->out = new ConsoleOutput();
-        $this->err = new ConsoleOutput();
-        $io = new ConsoleIo($this->out, $this->err);
-
-        $this->QueueShell = $this->getMockBuilder(QueueShell::class)
-            ->setMethods([
-                'in',
-                'err',
-                '_stop'
-            ])
-            ->setConstructorArgs([
-                $io
-            ])
-            ->getMock();
-        $this->QueueShell->initialize();
-
-        Configure::write('Queue', [
-            'sleepTime' => 2,
-            'defaultWorkerTimeout' => 3,
-            'workerMaxRuntime' => 5,
-            'cleanupTimeout' => 10,
-            'exitWhenNothingToDo' => false,
-            'log' => false
-        ]);
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testObject()
-    {
-        $this->assertTrue(is_object($this->QueueShell));
-        $this->assertInstanceOf(QueueShell::class, $this->QueueShell);
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testStats()
-    {
-        $this->_needsConnection();
-
-        $this->QueueShell->stats();
-        $this->assertContains('Total unfinished jobs: 0', $this->out->output());
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testSettings()
-    {
-        $this->QueueShell->settings();
-        $this->assertContains('* cleanupTimeout: 10', $this->out->output());
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testAddInexistent()
-    {
-        $this->QueueShell->args[] = 'FooBar';
-        $this->QueueShell->add();
-        $this->assertContains('Error: Task not found: FooBar', $this->out->output());
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testAdd()
-    {
-        $this->QueueShell->args[] = 'Example';
-        $this->QueueShell->add();
-
-        $this->assertContains('OK, job created, now run the worker', $this->out->output(), print_r($this->out->output(), true));
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testTimeNeeded()
-    {
-        $this->QueueShell = $this->getMockBuilder(QueueShell::class)
-            ->setMethods([
-                '_time'
-            ])
-            ->getMock();
-
-        $first = time();
-        $second = $first - HOUR + MINUTE;
-        $this->QueueShell->expects($this->at(0))
-            ->method('_time')
-            ->will($this->returnValue($first));
-        $this->QueueShell->expects($this->at(1))
-            ->method('_time')
-            ->will($this->returnValue($second));
-        $this->QueueShell->expects($this->exactly(2))
-            ->method('_time')
-            ->withAnyParameters();
-
-        $result = $this->invokeMethod($this->QueueShell, '_timeNeeded');
-        $this->assertSame('3540s', $result);
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testMemoryUsage()
-    {
-        $result = $this->invokeMethod($this->QueueShell, '_memoryUsage');
-        $this->assertRegExp('/^\d+MB/', $result, 'Should be e.g. `17MB` or `17MB/1GB` etc.');
-    }
-
-    /**
-     *
-     * @return void
-     */
-    public function testStringToArray()
-    {
-        $string = 'Foo,Bar,';
-        $result = $this->invokeMethod($this->QueueShell, '_stringToArray', [$string]);
-
-        $expected = [
-            'Foo',
-            'Bar'
-        ];
-        $this->assertSame($expected, $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.');
-    }
+	use ToolsTestTrait;
+
+	/**
+	 *
+	 * @var \Queue\Shell\QueueShell|\PHPUnit_Framework_MockObject_MockObject
+	 */
+	public $QueueShell;
+
+	/**
+	 *
+	 * @var \Tools\TestSuite\ConsoleOutput
+	 */
+	public $out;
+
+	/**
+	 *
+	 * @var \Tools\TestSuite\ConsoleOutput
+	 */
+	public $err;
+
+	/**
+	 * Fixtures to load
+	 *
+	 * @var array
+	 */
+	public $fixtures = [
+		'plugin.Queue.QueuedTasks'
+	];
+
+	/**
+	 * Setup Defaults
+	 *
+	 * @return void
+	 */
+	public function setUp()
+	{
+		parent::setUp();
+
+		$this->out = new ConsoleOutput();
+		$this->err = new ConsoleOutput();
+		$io = new ConsoleIo($this->out, $this->err);
+
+		$this->QueueShell = $this->getMockBuilder(QueueShell::class)
+			->setMethods([
+				'in',
+				'err',
+				'_stop'
+			])
+			->setConstructorArgs([
+				$io
+			])
+			->getMock();
+		$this->QueueShell->initialize();
+
+		Configure::write('Queue', [
+			'sleepTime' => 2,
+			'defaultWorkerTimeout' => 3,
+			'workerMaxRuntime' => 5,
+			'cleanupTimeout' => 10,
+			'exitWhenNothingToDo' => false,
+			'log' => false
+		]);
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testObject()
+	{
+		$this->assertTrue(is_object($this->QueueShell));
+		$this->assertInstanceOf(QueueShell::class, $this->QueueShell);
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testStats()
+	{
+		$this->_needsConnection();
+
+		$this->QueueShell->stats();
+		$this->assertContains('Total unfinished jobs: 0', $this->out->output());
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testSettings()
+	{
+		$this->QueueShell->settings();
+		$this->assertContains('* cleanupTimeout: 10', $this->out->output());
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testAddInexistent()
+	{
+		$this->QueueShell->args[] = 'FooBar';
+		$this->QueueShell->add();
+		$this->assertContains('Error: Task not found: FooBar', $this->out->output());
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testAdd()
+	{
+		$this->QueueShell->args[] = 'Example';
+		$this->QueueShell->add();
+
+		$this->assertContains('OK, job created, now run the worker', $this->out->output(), print_r($this->out->output(), true));
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testTimeNeeded()
+	{
+		$this->QueueShell = $this->getMockBuilder(QueueShell::class)
+			->setMethods([
+				'_time'
+			])
+			->getMock();
+
+		$first = time();
+		$second = $first - HOUR + MINUTE;
+		$this->QueueShell->expects($this->at(0))
+			->method('_time')
+			->will($this->returnValue($first));
+		$this->QueueShell->expects($this->at(1))
+			->method('_time')
+			->will($this->returnValue($second));
+		$this->QueueShell->expects($this->exactly(2))
+			->method('_time')
+			->withAnyParameters();
+
+		$result = $this->invokeMethod($this->QueueShell, '_timeNeeded');
+		$this->assertSame('3540s', $result);
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testMemoryUsage()
+	{
+		$result = $this->invokeMethod($this->QueueShell, '_memoryUsage');
+		$this->assertRegExp('/^\d+MB/', $result, 'Should be e.g. `17MB` or `17MB/1GB` etc.');
+	}
+
+	/**
+	 *
+	 * @return void
+	 */
+	public function testStringToArray()
+	{
+		$string = 'Foo,Bar,';
+		$result = $this->invokeMethod($this->QueueShell, '_stringToArray', [$string]);
+
+		$expected = [
+			'Foo',
+			'Bar'
+		];
+		$this->assertSame($expected, $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.');
+	}
 }

src/Model/Table/QueuedTasksTable.php

Indentation +532 added lines, -532 removed lines

@@ -28,536 +28,536 @@
 class QueuedTasksTable extends Table
 {
 
-    const DRIVER_MYSQL = 'Mysql';
-
-    const DRIVER_POSTGRES = 'Postgres';
-
-    const DRIVER_SQLSERVER = 'Sqlserver';
-
-    const STATS_LIMIT = 100000;
-
-    /**
-     * Initialize method
-     *
-     * @param array $config The configuration for the Table.
-     * @return void
-     */
-    public function initialize(array $config)
-    {
-        parent::initialize($config);
-
-        $this->setTable('queued_tasks');
-        $this->setDisplayField('id');
-        $this->setPrimaryKey('id');
-
-        $this->addBehavior('Timestamp');
-    }
-
-    /**
-     * set connection name
-     *
-     * @return string
-     */
-    public static function defaultConnectionName()
-    {
-        $connection = Configure::read('Queue.connection');
-        if (!empty($connection)) {
-            return $connection;
-        }
-
-        return parent::defaultConnectionName();
-    }
-
-    /**
-     *
-     * @param \Cake\Event\Event $event Model event
-     * @param \ArrayObject $data The data
-     * @param \ArrayObject $options The options
-     * @return void
-     */
-    public function beforeMarshal(Event $event, ArrayObject $data, ArrayObject $options)
-    {
-        if (isset($data['data']) && $data['data'] === '') {
-            $data['data'] = null;
-        }
-    }
-
-    /**
-     * Adds a new job to the queue.
-     *
-     * @param string $taskName Task name
-     * @param array|null $data Array of data
-     * @param string $notBefore A datetime which indicates when the job may be executed
-     * @return \Queue\Model\Entity\QueuedTask Saved job entity
-     */
-    public function createJob($taskName, array $data = null, string $notBefore = null)
-    {
-        $task = [
-            'task' => $taskName,
-            'data' => serialize($data),
-            'not_before' => $this->getDateTime()
-        ];
-
-        if (!empty($notBefore)) {
-            $task['not_before'] = $this->getDateTime(strtotime($notBefore));
-        }
-
-        $queuedTask = $this->newEntity($task);
-
-        return $this->saveOrFail($queuedTask);
-    }
-
-    /**
-     * Returns the number of items in the queue.
-     * Either returns the number of ALL pending jobs, or the number of pending jobs of the passed type.
-     *
-     * @param string|null $taskName Task type to Count
-     * @return int
-     */
-    public function getLength($taskName = null)
-    {
-        $findConf = [
-            'conditions' => [
-                'completed IS' => null
-            ]
-        ];
-        if ($taskName !== null) {
-            $findConf['conditions']['task'] = $taskName;
-        }
-
-        return $this->find('all', $findConf)->count();
-    }
-
-    /**
-     * Return a list of all task types in the Queue.
-     *
-     * @return \Cake\ORM\Query
-     */
-    public function getTypes()
-    {
-        $findCond = [
-            'fields' => [
-                'task'
-            ],
-            'group' => [
-                'task'
-            ],
-            'keyField' => 'task',
-            'valueField' => 'task'
-        ];
-
-        return $this->find('list', $findCond);
-    }
-
-    /**
-     * Return some statistics about finished jobs still in the Database.
-     * TO-DO: rewrite as virtual field
-     *
-     * @return \Cake\ORM\Query
-     */
-    public function getStats()
-    {
-        $driverName = $this->_getDriverName();
-        $options = [
-            'fields' => function (Query $query) use ($driverName) {
-                $alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)');
-                $runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
-                $fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(not_before is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(not_before))');
-                switch ($driverName) {
-                    case static::DRIVER_SQLSERVER:
-                        $alltime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', created)");
-                        $runtime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
-                        $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)");
-                        break;
-                }
-                /**
-                 *
-                 * @var \Cake\ORM\Query
-                 */
-                return [
-                    'task',
-                    'num' => $query->func()->count('*'),
-                    'alltime' => $alltime,
-                    'runtime' => $runtime,
-                    'fetchdelay' => $fetchdelay
-                ];
-            },
-            'conditions' => [
-                'completed IS NOT' => null
-            ],
-            'group' => [
-                'task'
-            ]
-        ];
-
-        return $this->find('all', $options);
-    }
-
-    /**
-     * Returns [
-     * 'Task' => [
-     * 'YYYY-MM-DD' => INT,
-     * ...
-     * ]
-     * ]
-     *
-     * @param string|null $taskName The task name
-     * @return array
-     */
-    public function getFullStats($taskName = null)
-    {
-        $driverName = $this->_getDriverName();
-        $fields = function (Query $query) use ($driverName) {
-            $runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
-            switch ($driverName) {
-                case static::DRIVER_SQLSERVER:
-                    $runtime = $query->newExpr("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
-                    break;
-            }
-
-            return [
-                'task',
-                'created',
-                'duration' => $runtime
-            ];
-        };
-
-        $conditions = [
-            'completed IS NOT' => null
-        ];
-        if ($taskName) {
-            $conditions['task'] = $taskName;
-        }
-
-        $tasks = $this->find()
-            ->select($fields)
-            ->where($conditions)
-            ->enableHydration(false)
-            ->orderDesc('id')
-            ->limit(static::STATS_LIMIT)
-            ->all()
-            ->toArray();
-
-        $result = [];
-
-        $days = [];
-
-        foreach ($tasks as $task) {
-            /** @var \DateTime $created */
-            $created = $task['created'];
-            $day = $created->format('Y-m-d');
-            if (!isset($days[$day])) {
-                $days[$day] = $day;
-            }
-
-            $result[$task['task']][$day][] = $task['duration'];
-        }
-
-        foreach ($result as $type => $tasks) {
-            foreach ($tasks as $day => $durations) {
-                $average = array_sum($durations) / count($durations);
-                $result[$type][$day] = (int)$average;
-            }
-
-            foreach ($days as $day) {
-                if (isset($result[$type][$day])) {
-                    continue;
-                }
-
-                $result[$type][$day] = 0;
-            }
-
-            ksort($result[$type]);
-        }
-
-        return $result;
-    }
-
-    /**
-     * Look for a new job that can be processed with the current abilities and
-     * from the specified group (or any if null).
-     *
-     * @param array $capabilities Available QueueWorkerTasks.
-     * @param array $types Request a job from these types (or exclude certain types), or any otherwise.
-     * @return \Queue\Model\Entity\QueuedTask|null
-     */
-    public function requestJob(array $capabilities, array $types = [])
-    {
-        $now = $this->getDateTime();
-        $nowStr = $now->toDateTimeString();
-        $driverName = $this->_getDriverName();
-
-        $query = $this->find();
-        $age = $query->newExpr()->add('IFNULL(TIMESTAMPDIFF(SECOND, "' . $nowStr . '", not_before), 0)');
-        switch ($driverName) {
-            case static::DRIVER_SQLSERVER:
-                $age = $query->newExpr()->add('ISNULL(DATEDIFF(SECOND, GETDATE(), not_before), 0)');
-                break;
-            case static::DRIVER_POSTGRES:
-                $age = $query->newExpr()->add('COALESCE((EXTRACT(EPOCH FROM now()) - EXTRACT(EPOCH FROM not_before)), 0)');
-                break;
-        }
-        $options = [
-            'conditions' => [
-                'completed IS' => null,
-                'OR' => []
-            ],
-            'fields' => [
-                'age' => $age
-            ],
-            'order' => [
-                'age' => 'ASC',
-                'id' => 'ASC'
-            ]
-        ];
-
-        if ($types) {
-            $options['conditions'] = $this->addFilter($options['conditions'], 'task', $types);
-        }
-
-        // Generate the task specific conditions.
-        foreach ($capabilities as $task) {
-            list ($plugin, $name) = pluginSplit($task['name']);
-            $timeoutAt = $now->copy();
-            $tmp = [
-                'task' => $name,
-                'AND' => [
-                    [
-                        'OR' => [
-                            'not_before <=' => $nowStr,
-                            'not_before IS' => null
-                        ]
-                    ],
-                    [
-                        'OR' => [
-                            'fetched <' => $timeoutAt->subSeconds($task['timeout']),
-                            'fetched IS' => null
-                        ]
-                    ]
-                ],
-                'failed_count <' => ($task['retries'] + 1)
-            ];
-            $options['conditions']['OR'][] = $tmp;
-        }
-
-        /** @var \Queue\Model\Entity\QueuedTask|null $task */
-        $task = $this->getConnection()->transactional(function () use ($query, $options, $now) {
-            $task = $query->find('all', $options)
-                ->enableAutoFields(true)
-                ->epilog('FOR UPDATE')
-                ->first();
-
-            if (!$task) {
-                return null;
-            }
-
-            $key = sha1(microtime());
-            $task = $this->patchEntity($task, [
-                'worker_key' => $key,
-                'fetched' => $now
-            ]);
-
-            return $this->saveOrFail($task);
-        });
-
-        if (!$task) {
-            return null;
-        }
-
-        return $task;
-    }
-
-    /**
-     * Mark a task as Completed, removing it from the queue.
-     *
-     * @param \Queue\Model\Entity\QueuedTask $task Task
-     * @return bool Success
-     */
-    public function markJobDone(QueuedTask $task)
-    {
-        $fields = [
-            'completed' => $this->getDateTime()
-        ];
-        $task = $this->patchEntity($task, $fields);
-
-        return (bool)$this->save($task);
-    }
-
-    /**
-     * Mark a job as Failed, incrementing the failed-counter and Requeueing it.
-     *
-     * @param \Queue\Model\Entity\QueuedTask $task Task
-     * @param string|null $failureMessage Optional message to append to the failure_message field.
-     * @return bool Success
-     */
-    public function markJobFailed(QueuedTask $task, $failureMessage = null)
-    {
-        $fields = [
-            'failed_count' => $task->failed_count + 1,
-            'failure_message' => $failureMessage
-        ];
-        $task = $this->patchEntity($task, $fields);
-
-        return (bool)$this->save($task);
-    }
-
-    /**
-     * Reset current jobs
-     *
-     * @param int|null $id ID
-     *
-     * @return int Success
-     */
-    public function reset($id = null)
-    {
-        $fields = [
-            'completed' => null,
-            'fetched' => null,
-            'failed_count' => 0,
-            'worker_key' => null,
-            'failure_message' => null
-        ];
-        $conditions = [
-            'completed IS' => null
-        ];
-        if ($id) {
-            $conditions['id'] = $id;
-        }
-
-        return $this->updateAll($fields, $conditions);
-    }
-
-    /**
-     *
-     * @param string $taskName Task name
-     *
-     * @return int
-     */
-    public function rerun($taskName)
-    {
-        $fields = [
-            'completed' => null,
-            'fetched' => null,
-            'failed_count' => 0,
-            'worker_key' => null,
-            'failure_message' => null
-        ];
-        $conditions = [
-            'completed IS NOT' => null,
-            'task' => $taskName
-        ];
-
-        return $this->updateAll($fields, $conditions);
-    }
-
-    /**
-     * Cleanup/Delete Completed Tasks.
-     *
-     * @return void
-     */
-    public function cleanOldJobs()
-    {
-        if (!Configure::read('Queue.cleanuptimeout')) {
-            return;
-        }
-
-        $this->deleteAll([
-            'completed <' => time() - (int)Configure::read('Queue.cleanuptimeout')
-        ]);
-    }
-
-    /**
-     *
-     * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task
-     * @param array $taskConfiguration Task configuration
-     * @return string
-     */
-    public function getFailedStatus($queuedTask, array $taskConfiguration)
-    {
-        $failureMessageRequeued = 'requeued';
-
-        $queuedTaskName = 'Queue' . $queuedTask->task;
-        if (empty($taskConfiguration[$queuedTaskName])) {
-            return $failureMessageRequeued;
-        }
-        $retries = $taskConfiguration[$queuedTaskName]['retries'];
-        if ($queuedTask->failed_count <= $retries) {
-            return $failureMessageRequeued;
-        }
-
-        return 'aborted';
-    }
-
-    /**
-     * truncate()
-     *
-     * @return void
-     */
-    public function truncate()
-    {
-        $sql = $this->getSchema()->truncateSql($this->_connection);
-        foreach ($sql as $snippet) {
-            $this->_connection->execute($snippet);
-        }
-    }
-
-    /**
-     * get the name of the driver
-     *
-     * @return string
-     */
-    protected function _getDriverName()
-    {
-        $className = explode('\\', $this->getConnection()->config()['driver']);
-        $name = end($className);
-
-        return $name;
-    }
-
-    /**
-     *
-     * @param array $conditions Conditions
-     * @param string $key Key
-     * @param array $values Values
-     * @return array
-     */
-    protected function addFilter(array $conditions, $key, array $values)
-    {
-        $include = [];
-        $exclude = [];
-        foreach ($values as $value) {
-            if (substr($value, 0, 1) === '-') {
-                $exclude[] = substr($value, 1);
-            } else {
-                $include[] = $value;
-            }
-        }
-
-        if ($include) {
-            $conditions[$key . ' IN'] = $include;
-        }
-        if ($exclude) {
-            $conditions[$key . ' NOT IN'] = $exclude;
-        }
-
-        return $conditions;
-    }
-
-    /**
-     * Returns a DateTime object from different input.
-     *
-     * Without argument this will be "now".
-     *
-     * @param int|string|\Cake\I18n\FrozenTime|\Cake\I18n\Time|null $notBefore Not before time
-     *
-     * @return \Cake\I18n\FrozenTime|\Cake\I18n\Time
-     */
-    protected function getDateTime($notBefore = null)
-    {
-        if (is_object($notBefore)) {
-            return $notBefore;
-        }
-
-        return new FrozenTime($notBefore);
-    }
+	const DRIVER_MYSQL = 'Mysql';
+
+	const DRIVER_POSTGRES = 'Postgres';
+
+	const DRIVER_SQLSERVER = 'Sqlserver';
+
+	const STATS_LIMIT = 100000;
+
+	/**
+	 * Initialize method
+	 *
+	 * @param array $config The configuration for the Table.
+	 * @return void
+	 */
+	public function initialize(array $config)
+	{
+		parent::initialize($config);
+
+		$this->setTable('queued_tasks');
+		$this->setDisplayField('id');
+		$this->setPrimaryKey('id');
+
+		$this->addBehavior('Timestamp');
+	}
+
+	/**
+	 * set connection name
+	 *
+	 * @return string
+	 */
+	public static function defaultConnectionName()
+	{
+		$connection = Configure::read('Queue.connection');
+		if (!empty($connection)) {
+			return $connection;
+		}
+
+		return parent::defaultConnectionName();
+	}
+
+	/**
+	 *
+	 * @param \Cake\Event\Event $event Model event
+	 * @param \ArrayObject $data The data
+	 * @param \ArrayObject $options The options
+	 * @return void
+	 */
+	public function beforeMarshal(Event $event, ArrayObject $data, ArrayObject $options)
+	{
+		if (isset($data['data']) && $data['data'] === '') {
+			$data['data'] = null;
+		}
+	}
+
+	/**
+	 * Adds a new job to the queue.
+	 *
+	 * @param string $taskName Task name
+	 * @param array|null $data Array of data
+	 * @param string $notBefore A datetime which indicates when the job may be executed
+	 * @return \Queue\Model\Entity\QueuedTask Saved job entity
+	 */
+	public function createJob($taskName, array $data = null, string $notBefore = null)
+	{
+		$task = [
+			'task' => $taskName,
+			'data' => serialize($data),
+			'not_before' => $this->getDateTime()
+		];
+
+		if (!empty($notBefore)) {
+			$task['not_before'] = $this->getDateTime(strtotime($notBefore));
+		}
+
+		$queuedTask = $this->newEntity($task);
+
+		return $this->saveOrFail($queuedTask);
+	}
+
+	/**
+	 * Returns the number of items in the queue.
+	 * Either returns the number of ALL pending jobs, or the number of pending jobs of the passed type.
+	 *
+	 * @param string|null $taskName Task type to Count
+	 * @return int
+	 */
+	public function getLength($taskName = null)
+	{
+		$findConf = [
+			'conditions' => [
+				'completed IS' => null
+			]
+		];
+		if ($taskName !== null) {
+			$findConf['conditions']['task'] = $taskName;
+		}
+
+		return $this->find('all', $findConf)->count();
+	}
+
+	/**
+	 * Return a list of all task types in the Queue.
+	 *
+	 * @return \Cake\ORM\Query
+	 */
+	public function getTypes()
+	{
+		$findCond = [
+			'fields' => [
+				'task'
+			],
+			'group' => [
+				'task'
+			],
+			'keyField' => 'task',
+			'valueField' => 'task'
+		];
+
+		return $this->find('list', $findCond);
+	}
+
+	/**
+	 * Return some statistics about finished jobs still in the Database.
+	 * TO-DO: rewrite as virtual field
+	 *
+	 * @return \Cake\ORM\Query
+	 */
+	public function getStats()
+	{
+		$driverName = $this->_getDriverName();
+		$options = [
+			'fields' => function (Query $query) use ($driverName) {
+				$alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)');
+				$runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
+				$fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(not_before is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(not_before))');
+				switch ($driverName) {
+					case static::DRIVER_SQLSERVER:
+						$alltime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', created)");
+						$runtime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
+						$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)");
+						break;
+				}
+				/**
+				 *
+				 * @var \Cake\ORM\Query
+				 */
+				return [
+					'task',
+					'num' => $query->func()->count('*'),
+					'alltime' => $alltime,
+					'runtime' => $runtime,
+					'fetchdelay' => $fetchdelay
+				];
+			},
+			'conditions' => [
+				'completed IS NOT' => null
+			],
+			'group' => [
+				'task'
+			]
+		];
+
+		return $this->find('all', $options);
+	}
+
+	/**
+	 * Returns [
+	 * 'Task' => [
+	 * 'YYYY-MM-DD' => INT,
+	 * ...
+	 * ]
+	 * ]
+	 *
+	 * @param string|null $taskName The task name
+	 * @return array
+	 */
+	public function getFullStats($taskName = null)
+	{
+		$driverName = $this->_getDriverName();
+		$fields = function (Query $query) use ($driverName) {
+			$runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
+			switch ($driverName) {
+				case static::DRIVER_SQLSERVER:
+					$runtime = $query->newExpr("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
+					break;
+			}
+
+			return [
+				'task',
+				'created',
+				'duration' => $runtime
+			];
+		};
+
+		$conditions = [
+			'completed IS NOT' => null
+		];
+		if ($taskName) {
+			$conditions['task'] = $taskName;
+		}
+
+		$tasks = $this->find()
+			->select($fields)
+			->where($conditions)
+			->enableHydration(false)
+			->orderDesc('id')
+			->limit(static::STATS_LIMIT)
+			->all()
+			->toArray();
+
+		$result = [];
+
+		$days = [];
+
+		foreach ($tasks as $task) {
+			/** @var \DateTime $created */
+			$created = $task['created'];
+			$day = $created->format('Y-m-d');
+			if (!isset($days[$day])) {
+				$days[$day] = $day;
+			}
+
+			$result[$task['task']][$day][] = $task['duration'];
+		}
+
+		foreach ($result as $type => $tasks) {
+			foreach ($tasks as $day => $durations) {
+				$average = array_sum($durations) / count($durations);
+				$result[$type][$day] = (int)$average;
+			}
+
+			foreach ($days as $day) {
+				if (isset($result[$type][$day])) {
+					continue;
+				}
+
+				$result[$type][$day] = 0;
+			}
+
+			ksort($result[$type]);
+		}
+
+		return $result;
+	}
+
+	/**
+	 * Look for a new job that can be processed with the current abilities and
+	 * from the specified group (or any if null).
+	 *
+	 * @param array $capabilities Available QueueWorkerTasks.
+	 * @param array $types Request a job from these types (or exclude certain types), or any otherwise.
+	 * @return \Queue\Model\Entity\QueuedTask|null
+	 */
+	public function requestJob(array $capabilities, array $types = [])
+	{
+		$now = $this->getDateTime();
+		$nowStr = $now->toDateTimeString();
+		$driverName = $this->_getDriverName();
+
+		$query = $this->find();
+		$age = $query->newExpr()->add('IFNULL(TIMESTAMPDIFF(SECOND, "' . $nowStr . '", not_before), 0)');
+		switch ($driverName) {
+			case static::DRIVER_SQLSERVER:
+				$age = $query->newExpr()->add('ISNULL(DATEDIFF(SECOND, GETDATE(), not_before), 0)');
+				break;
+			case static::DRIVER_POSTGRES:
+				$age = $query->newExpr()->add('COALESCE((EXTRACT(EPOCH FROM now()) - EXTRACT(EPOCH FROM not_before)), 0)');
+				break;
+		}
+		$options = [
+			'conditions' => [
+				'completed IS' => null,
+				'OR' => []
+			],
+			'fields' => [
+				'age' => $age
+			],
+			'order' => [
+				'age' => 'ASC',
+				'id' => 'ASC'
+			]
+		];
+
+		if ($types) {
+			$options['conditions'] = $this->addFilter($options['conditions'], 'task', $types);
+		}
+
+		// Generate the task specific conditions.
+		foreach ($capabilities as $task) {
+			list ($plugin, $name) = pluginSplit($task['name']);
+			$timeoutAt = $now->copy();
+			$tmp = [
+				'task' => $name,
+				'AND' => [
+					[
+						'OR' => [
+							'not_before <=' => $nowStr,
+							'not_before IS' => null
+						]
+					],
+					[
+						'OR' => [
+							'fetched <' => $timeoutAt->subSeconds($task['timeout']),
+							'fetched IS' => null
+						]
+					]
+				],
+				'failed_count <' => ($task['retries'] + 1)
+			];
+			$options['conditions']['OR'][] = $tmp;
+		}
+
+		/** @var \Queue\Model\Entity\QueuedTask|null $task */
+		$task = $this->getConnection()->transactional(function () use ($query, $options, $now) {
+			$task = $query->find('all', $options)
+				->enableAutoFields(true)
+				->epilog('FOR UPDATE')
+				->first();
+
+			if (!$task) {
+				return null;
+			}
+
+			$key = sha1(microtime());
+			$task = $this->patchEntity($task, [
+				'worker_key' => $key,
+				'fetched' => $now
+			]);
+
+			return $this->saveOrFail($task);
+		});
+
+		if (!$task) {
+			return null;
+		}
+
+		return $task;
+	}
+
+	/**
+	 * Mark a task as Completed, removing it from the queue.
+	 *
+	 * @param \Queue\Model\Entity\QueuedTask $task Task
+	 * @return bool Success
+	 */
+	public function markJobDone(QueuedTask $task)
+	{
+		$fields = [
+			'completed' => $this->getDateTime()
+		];
+		$task = $this->patchEntity($task, $fields);
+
+		return (bool)$this->save($task);
+	}
+
+	/**
+	 * Mark a job as Failed, incrementing the failed-counter and Requeueing it.
+	 *
+	 * @param \Queue\Model\Entity\QueuedTask $task Task
+	 * @param string|null $failureMessage Optional message to append to the failure_message field.
+	 * @return bool Success
+	 */
+	public function markJobFailed(QueuedTask $task, $failureMessage = null)
+	{
+		$fields = [
+			'failed_count' => $task->failed_count + 1,
+			'failure_message' => $failureMessage
+		];
+		$task = $this->patchEntity($task, $fields);
+
+		return (bool)$this->save($task);
+	}
+
+	/**
+	 * Reset current jobs
+	 *
+	 * @param int|null $id ID
+	 *
+	 * @return int Success
+	 */
+	public function reset($id = null)
+	{
+		$fields = [
+			'completed' => null,
+			'fetched' => null,
+			'failed_count' => 0,
+			'worker_key' => null,
+			'failure_message' => null
+		];
+		$conditions = [
+			'completed IS' => null
+		];
+		if ($id) {
+			$conditions['id'] = $id;
+		}
+
+		return $this->updateAll($fields, $conditions);
+	}
+
+	/**
+	 *
+	 * @param string $taskName Task name
+	 *
+	 * @return int
+	 */
+	public function rerun($taskName)
+	{
+		$fields = [
+			'completed' => null,
+			'fetched' => null,
+			'failed_count' => 0,
+			'worker_key' => null,
+			'failure_message' => null
+		];
+		$conditions = [
+			'completed IS NOT' => null,
+			'task' => $taskName
+		];
+
+		return $this->updateAll($fields, $conditions);
+	}
+
+	/**
+	 * Cleanup/Delete Completed Tasks.
+	 *
+	 * @return void
+	 */
+	public function cleanOldJobs()
+	{
+		if (!Configure::read('Queue.cleanuptimeout')) {
+			return;
+		}
+
+		$this->deleteAll([
+			'completed <' => time() - (int)Configure::read('Queue.cleanuptimeout')
+		]);
+	}
+
+	/**
+	 *
+	 * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task
+	 * @param array $taskConfiguration Task configuration
+	 * @return string
+	 */
+	public function getFailedStatus($queuedTask, array $taskConfiguration)
+	{
+		$failureMessageRequeued = 'requeued';
+
+		$queuedTaskName = 'Queue' . $queuedTask->task;
+		if (empty($taskConfiguration[$queuedTaskName])) {
+			return $failureMessageRequeued;
+		}
+		$retries = $taskConfiguration[$queuedTaskName]['retries'];
+		if ($queuedTask->failed_count <= $retries) {
+			return $failureMessageRequeued;
+		}
+
+		return 'aborted';
+	}
+
+	/**
+	 * truncate()
+	 *
+	 * @return void
+	 */
+	public function truncate()
+	{
+		$sql = $this->getSchema()->truncateSql($this->_connection);
+		foreach ($sql as $snippet) {
+			$this->_connection->execute($snippet);
+		}
+	}
+
+	/**
+	 * get the name of the driver
+	 *
+	 * @return string
+	 */
+	protected function _getDriverName()
+	{
+		$className = explode('\\', $this->getConnection()->config()['driver']);
+		$name = end($className);
+
+		return $name;
+	}
+
+	/**
+	 *
+	 * @param array $conditions Conditions
+	 * @param string $key Key
+	 * @param array $values Values
+	 * @return array
+	 */
+	protected function addFilter(array $conditions, $key, array $values)
+	{
+		$include = [];
+		$exclude = [];
+		foreach ($values as $value) {
+			if (substr($value, 0, 1) === '-') {
+				$exclude[] = substr($value, 1);
+			} else {
+				$include[] = $value;
+			}
+		}
+
+		if ($include) {
+			$conditions[$key . ' IN'] = $include;
+		}
+		if ($exclude) {
+			$conditions[$key . ' NOT IN'] = $exclude;
+		}
+
+		return $conditions;
+	}
+
+	/**
+	 * Returns a DateTime object from different input.
+	 *
+	 * Without argument this will be "now".
+	 *
+	 * @param int|string|\Cake\I18n\FrozenTime|\Cake\I18n\Time|null $notBefore Not before time
+	 *
+	 * @return \Cake\I18n\FrozenTime|\Cake\I18n\Time
+	 */
+	protected function getDateTime($notBefore = null)
+	{
+		if (is_object($notBefore)) {
+			return $notBefore;
+		}
+
+		return new FrozenTime($notBefore);
+	}
 }

src/Queue/Config.php

Indentation +53 added lines, -53 removed lines

@@ -6,57 +6,57 @@
 class Config
 {
 
-    /**
-     *
-     * @return int
-     */
-    public static function defaultWorkerTimeout()
-    {
-        return Configure::read('Queue.defaultWorkerTimeout', 600); // 10min
-    }
-
-    /**
-     *
-     * @return int
-     */
-    public static function workerMaxRuntime()
-    {
-        return Configure::read('Queue.workerMaxRuntime', 120);
-    }
-
-    /**
-     *
-     * @return int
-     */
-    public static function cleanupTimeout()
-    {
-        return Configure::read('Queue.cleanupTimeout', 2592000); // 30 days
-    }
-
-    /**
-     *
-     * @return int
-     */
-    public static function sleepTime()
-    {
-        return Configure::read('Queue.sleepTime', 10);
-    }
-
-    /**
-     *
-     * @return int
-     */
-    public static function gcprob()
-    {
-        return Configure::read('Queue.gcprob', 10);
-    }
-
-    /**
-     *
-     * @return int
-     */
-    public static function defaultWorkerRetries()
-    {
-        return Configure::read('Queue.defaultWorkerRetries', 1);
-    }
+	/**
+	 *
+	 * @return int
+	 */
+	public static function defaultWorkerTimeout()
+	{
+		return Configure::read('Queue.defaultWorkerTimeout', 600); // 10min
+	}
+
+	/**
+	 *
+	 * @return int
+	 */
+	public static function workerMaxRuntime()
+	{
+		return Configure::read('Queue.workerMaxRuntime', 120);
+	}
+
+	/**
+	 *
+	 * @return int
+	 */
+	public static function cleanupTimeout()
+	{
+		return Configure::read('Queue.cleanupTimeout', 2592000); // 30 days
+	}
+
+	/**
+	 *
+	 * @return int
+	 */
+	public static function sleepTime()
+	{
+		return Configure::read('Queue.sleepTime', 10);
+	}
+
+	/**
+	 *
+	 * @return int
+	 */
+	public static function gcprob()
+	{
+		return Configure::read('Queue.gcprob', 10);
+	}
+
+	/**
+	 *
+	 * @return int
+	 */
+	public static function defaultWorkerRetries()
+	{
+		return Configure::read('Queue.defaultWorkerRetries', 1);
+	}
 }

src/Mailer/Transport/SimpleQueueTransport.php

Indentation +105 added lines, -105 removed lines

@@ -15,115 +15,115 @@
 class SimpleQueueTransport extends AbstractTransport
 {
 
-    /**
-     * Send mail
-     *
-     * @param \Cake\Mailer\Email $email Email
-     * @return array
-     */
-    public function send(Email $email)
-    {
-        if (!empty($this->_config['queue'])) {
-            $this->_config = $this->_config['queue'] + $this->_config;
-            $email->setConfig((array)$this->_config['queue'] + [
-                'queue' => []
-            ]);
-            unset($this->_config['queue']);
-        }
+	/**
+	 * Send mail
+	 *
+	 * @param \Cake\Mailer\Email $email Email
+	 * @return array
+	 */
+	public function send(Email $email)
+	{
+		if (!empty($this->_config['queue'])) {
+			$this->_config = $this->_config['queue'] + $this->_config;
+			$email->setConfig((array)$this->_config['queue'] + [
+				'queue' => []
+			]);
+			unset($this->_config['queue']);
+		}
 
-        $settings = [
-            'from' => [
-                $email->getFrom()
-            ],
-            'to' => [
-                $email->getTo()
-            ],
-            'cc' => [
-                $email->getCc()
-            ],
-            'bcc' => [
-                $email->getBcc()
-            ],
-            'charset' => [
-                $email->getCharset()
-            ],
-            'replyTo' => [
-                $email->getReplyTo()
-            ],
-            'readReceipt' => [
-                $email->getReadReceipt()
-            ],
-            'returnPath' => [
-                $email->getReturnPath()
-            ],
-            'messageId' => [
-                $email->getMessageId()
-            ],
-            'domain' => [
-                $email->getDomain()
-            ],
-            'headers' => [
-                $email->getHeaders()
-            ],
-            'headerCharset' => [
-                $email->getHeaderCharset()
-            ],
-            'theme' => [
-                $email->viewBuilder()->getTheme()
-            ],
-            'profile' => [
-                $email->getProfile()
-            ],
-            'emailFormat' => [
-                $email->getEmailFormat()
-            ],
-            'subject' => method_exists($email, 'getOriginalSubject') ? [
-                $email->getOriginalSubject()
-            ] : [
-                $email->getSubject()
-            ],
-            'transport' => [
-                $this->_config['transport']
-            ],
-            'attachments' => [
-                $email->getAttachments()
-            ],
-            'template' => [
-                $email->viewBuilder()->getTemplate()
-            ],
-            'layout' => [
-                $email->viewBuilder()->getLayout()
-            ],
-            'viewVars' => [
-                $email->getViewVars()
-            ]
-        ];
+		$settings = [
+			'from' => [
+				$email->getFrom()
+			],
+			'to' => [
+				$email->getTo()
+			],
+			'cc' => [
+				$email->getCc()
+			],
+			'bcc' => [
+				$email->getBcc()
+			],
+			'charset' => [
+				$email->getCharset()
+			],
+			'replyTo' => [
+				$email->getReplyTo()
+			],
+			'readReceipt' => [
+				$email->getReadReceipt()
+			],
+			'returnPath' => [
+				$email->getReturnPath()
+			],
+			'messageId' => [
+				$email->getMessageId()
+			],
+			'domain' => [
+				$email->getDomain()
+			],
+			'headers' => [
+				$email->getHeaders()
+			],
+			'headerCharset' => [
+				$email->getHeaderCharset()
+			],
+			'theme' => [
+				$email->viewBuilder()->getTheme()
+			],
+			'profile' => [
+				$email->getProfile()
+			],
+			'emailFormat' => [
+				$email->getEmailFormat()
+			],
+			'subject' => method_exists($email, 'getOriginalSubject') ? [
+				$email->getOriginalSubject()
+			] : [
+				$email->getSubject()
+			],
+			'transport' => [
+				$this->_config['transport']
+			],
+			'attachments' => [
+				$email->getAttachments()
+			],
+			'template' => [
+				$email->viewBuilder()->getTemplate()
+			],
+			'layout' => [
+				$email->viewBuilder()->getLayout()
+			],
+			'viewVars' => [
+				$email->getViewVars()
+			]
+		];
 
-        foreach ($settings as $setting => $value) {
-            if (array_key_exists(0, $value) && ($value[0] === null || $value[0] === [])) {
-                unset($settings[$setting]);
-            }
-        }
+		foreach ($settings as $setting => $value) {
+			if (array_key_exists(0, $value) && ($value[0] === null || $value[0] === [])) {
+				unset($settings[$setting]);
+			}
+		}
 
-        $QueuedJobs = $this->getQueuedJobsModel();
-        $result = $QueuedJobs->createJob('Email', [
-            'settings' => $settings
-        ]);
-        $result['headers'] = '';
-        $result['message'] = '';
+		$QueuedJobs = $this->getQueuedJobsModel();
+		$result = $QueuedJobs->createJob('Email', [
+			'settings' => $settings
+		]);
+		$result['headers'] = '';
+		$result['message'] = '';
 
-        return $result->toArray();
-    }
+		return $result->toArray();
+	}
 
-    /**
-     *
-     * @return \Queue\Model\Table\QueuedTasksTable
-     */
-    protected function getQueuedJobsModel()
-    {
-        /** @var \Queue\Model\Table\QueuedTasksTable $table */
-        $table = TableRegistry::getTableLocator()->get('Queue.QueuedTasks');
+	/**
+	 *
+	 * @return \Queue\Model\Table\QueuedTasksTable
+	 */
+	protected function getQueuedJobsModel()
+	{
+		/** @var \Queue\Model\Table\QueuedTasksTable $table */
+		$table = TableRegistry::getTableLocator()->get('Queue.QueuedTasks');
 
-        return $table;
-    }
+		return $table;
+	}
 }

src/Mailer/Transport/QueueTransport.php

Indentation +27 added lines, -27 removed lines

@@ -15,34 +15,34 @@
 class QueueTransport extends AbstractTransport
 {
 
-    /**
-     * Send mail
-     *
-     * @param \Cake\Mailer\Email $email Email
-     * @return array
-     */
-    public function send(Email $email)
-    {
-        if (!empty($this->_config['queue'])) {
-            $this->_config = $this->_config['queue'] + $this->_config;
-            $email->setConfig((array)$this->_config['queue'] + [
-                'queue' => []
-            ]);
-            unset($this->_config['queue']);
-        }
+	/**
+	 * Send mail
+	 *
+	 * @param \Cake\Mailer\Email $email Email
+	 * @return array
+	 */
+	public function send(Email $email)
+	{
+		if (!empty($this->_config['queue'])) {
+			$this->_config = $this->_config['queue'] + $this->_config;
+			$email->setConfig((array)$this->_config['queue'] + [
+				'queue' => []
+			]);
+			unset($this->_config['queue']);
+		}
 
-        $transport = $this->_config['transport'];
-        $email->setTransport($transport);
+		$transport = $this->_config['transport'];
+		$email->setTransport($transport);
 
-        /** @var \Queue\Model\Table\QueuedTasksTable $QueuedTasks */
-        $QueuedTasks = TableRegistry::getTableLocator()->get('Queue.QueuedTasks');
-        $result = $QueuedTasks->createJob('Email', [
-            'transport' => $transport,
-            'settings' => $email
-        ]);
-        $result['headers'] = '';
-        $result['message'] = '';
+		/** @var \Queue\Model\Table\QueuedTasksTable $QueuedTasks */
+		$QueuedTasks = TableRegistry::getTableLocator()->get('Queue.QueuedTasks');
+		$result = $QueuedTasks->createJob('Email', [
+			'transport' => $transport,
+			'settings' => $email
+		]);
+		$result['headers'] = '';
+		$result['message'] = '';
 
-        return $result->toArray();
-    }
+		return $result->toArray();
+	}
 }

tests/TestCase/Mailer/Transport/SimpleQueueTransportTest.php

Indentation +107 added lines, -107 removed lines

@@ -11,121 +11,121 @@
 class SimpleQueueTransportTest extends TestCase
 {
 
-    /**
-     * Fixtures to load
-     *
-     * @var array
-     */
-    public $fixtures = [
-        'plugin.Queue.QueuedTasks'
-    ];
+	/**
+	 * Fixtures to load
+	 *
+	 * @var array
+	 */
+	public $fixtures = [
+		'plugin.Queue.QueuedTasks'
+	];
 
-    /**
-     *
-     * @var \Queue\Mailer\Transport\SimpleQueueTransport
-     */
-    protected $QueueTransport;
+	/**
+	 *
+	 * @var \Queue\Mailer\Transport\SimpleQueueTransport
+	 */
+	protected $QueueTransport;
 
-    /**
-     * Setup
-     *
-     * @return void
-     */
-    public function setUp()
-    {
-        parent::setUp();
-        $this->QueueTransport = new SimpleQueueTransport();
-    }
+	/**
+	 * Setup
+	 *
+	 * @return void
+	 */
+	public function setUp()
+	{
+		parent::setUp();
+		$this->QueueTransport = new SimpleQueueTransport();
+	}
 
-    /**
-     *
-     * @return void
-     */
-    public function testSendWithEmail()
-    {
-        $config = [
-            'transport' => 'queue',
-            'charset' => 'utf-8',
-            'headerCharset' => 'utf-8'
-        ];
+	/**
+	 *
+	 * @return void
+	 */
+	public function testSendWithEmail()
+	{
+		$config = [
+			'transport' => 'queue',
+			'charset' => 'utf-8',
+			'headerCharset' => 'utf-8'
+		];
 
-        $this->QueueTransport->setConfig($config);
-        $Email = new Email($config);
+		$this->QueueTransport->setConfig($config);
+		$Email = new Email($config);
 
-        $Email->setFrom('noreply@cakephp.org', 'CakePHP Test');
-        $Email->setTo('cake@cakephp.org', 'CakePHP');
-        $Email->setCc([
-            'mark@cakephp.org' => 'Mark Story',
-            'juan@cakephp.org' => 'Juan Basso'
-        ]);
-        $Email->setBcc('phpnut@cakephp.org');
-        $Email->setSubject('Testing Message');
-        $Email->setAttachments([
-            'wow.txt' => [
-                'data' => 'much wow!',
-                'mimetype' => 'text/plain',
-                'contentId' => 'important'
-            ]
-        ]);
+		$Email->setFrom('noreply@cakephp.org', 'CakePHP Test');
+		$Email->setTo('cake@cakephp.org', 'CakePHP');
+		$Email->setCc([
+			'mark@cakephp.org' => 'Mark Story',
+			'juan@cakephp.org' => 'Juan Basso'
+		]);
+		$Email->setBcc('phpnut@cakephp.org');
+		$Email->setSubject('Testing Message');
+		$Email->setAttachments([
+			'wow.txt' => [
+				'data' => 'much wow!',
+				'mimetype' => 'text/plain',
+				'contentId' => 'important'
+			]
+		]);
 
-        $Email->viewBuilder()->setLayout('test_layout');
-        $Email->viewBuilder()->setTemplate('test_template');
-        $Email->setSubject("L'utilisateur n'a pas pu être enregistré");
-        $Email->setReplyTo('noreply@cakephp.org');
-        $Email->setReadReceipt('noreply2@cakephp.org');
-        $Email->setReturnPath('noreply3@cakephp.org');
-        $Email->setDomain('cakephp.org');
-        $Email->viewBuilder()->setTheme('EuroTheme');
-        $Email->setEmailFormat('both');
-        $Email->set('var1', 1);
-        $Email->set('var2', 2);
+		$Email->viewBuilder()->setLayout('test_layout');
+		$Email->viewBuilder()->setTemplate('test_template');
+		$Email->setSubject("L'utilisateur n'a pas pu être enregistré");
+		$Email->setReplyTo('noreply@cakephp.org');
+		$Email->setReadReceipt('noreply2@cakephp.org');
+		$Email->setReturnPath('noreply3@cakephp.org');
+		$Email->setDomain('cakephp.org');
+		$Email->viewBuilder()->setTheme('EuroTheme');
+		$Email->setEmailFormat('both');
+		$Email->set('var1', 1);
+		$Email->set('var2', 2);
 
-        $result = $this->QueueTransport->send($Email);
-        $this->assertSame('Email', $result['task']);
-        $this->assertTrue(strlen($result['data']) < 10000);
+		$result = $this->QueueTransport->send($Email);
+		$this->assertSame('Email', $result['task']);
+		$this->assertTrue(strlen($result['data']) < 10000);
 
-        $output = unserialize($result['data']);
-        $emailReconstructed = new Email($config);
+		$output = unserialize($result['data']);
+		$emailReconstructed = new Email($config);
 
-        foreach ($output['settings'] as $method => $setting) {
-            $callable = $emailReconstructed;
-            if (in_array($method, [
-                'theme',
-                'template',
-                'layout'
-            ])) {
-                $callable = $callable->viewBuilder();
-            }
-            $setter = 'set' . ucfirst($method);
-            call_user_func_array([
-                $callable,
-                $setter
-            ], (array)$setting);
-        }
+		foreach ($output['settings'] as $method => $setting) {
+			$callable = $emailReconstructed;
+			if (in_array($method, [
+				'theme',
+				'template',
+				'layout'
+			])) {
+				$callable = $callable->viewBuilder();
+			}
+			$setter = 'set' . ucfirst($method);
+			call_user_func_array([
+				$callable,
+				$setter
+			], (array)$setting);
+		}
 
-        $this->assertEquals($emailReconstructed->getFrom(), $Email->getFrom());
-        $this->assertEquals($emailReconstructed->getTo(), $Email->getTo());
-        $this->assertEquals($emailReconstructed->getCc(), $Email->getCc());
-        $this->assertEquals($emailReconstructed->getBcc(), $Email->getBcc());
-        $this->assertEquals($emailReconstructed->getSubject(), $Email->getSubject());
-        $this->assertEquals($emailReconstructed->getCharset(), $Email->getCharset());
-        $this->assertEquals($emailReconstructed->getHeaderCharset(), $Email->getHeaderCharset());
-        $this->assertEquals($emailReconstructed->getEmailFormat(), $Email->getEmailFormat());
-        $this->assertEquals($emailReconstructed->getReplyTo(), $Email->getReplyTo());
-        $this->assertEquals($emailReconstructed->getReadReceipt(), $Email->getReadReceipt());
-        $this->assertEquals($emailReconstructed->getReturnPath(), $Email->getReturnPath());
-        // $this->assertEquals($emailReconstructed->getMessageId(), $Email->getMessageId());
-        $this->assertEquals($emailReconstructed->getDomain(), $Email->getDomain());
-        $this->assertEquals($emailReconstructed->viewBuilder()
-            ->getTheme(), $Email->viewBuilder()
-            ->getTheme());
-        $this->assertEquals($emailReconstructed->getProfile(), $Email->getProfile());
-        $this->assertEquals($emailReconstructed->getViewVars(), $Email->getViewVars());
-        $this->assertEquals($emailReconstructed->viewBuilder()
-            ->getTemplate(), $Email->viewBuilder()
-            ->getTemplate());
-        $this->assertEquals($emailReconstructed->viewBuilder()
-            ->getLayout(), $Email->viewBuilder()
-            ->getLayout());
-    }
+		$this->assertEquals($emailReconstructed->getFrom(), $Email->getFrom());
+		$this->assertEquals($emailReconstructed->getTo(), $Email->getTo());
+		$this->assertEquals($emailReconstructed->getCc(), $Email->getCc());
+		$this->assertEquals($emailReconstructed->getBcc(), $Email->getBcc());
+		$this->assertEquals($emailReconstructed->getSubject(), $Email->getSubject());
+		$this->assertEquals($emailReconstructed->getCharset(), $Email->getCharset());
+		$this->assertEquals($emailReconstructed->getHeaderCharset(), $Email->getHeaderCharset());
+		$this->assertEquals($emailReconstructed->getEmailFormat(), $Email->getEmailFormat());
+		$this->assertEquals($emailReconstructed->getReplyTo(), $Email->getReplyTo());
+		$this->assertEquals($emailReconstructed->getReadReceipt(), $Email->getReadReceipt());
+		$this->assertEquals($emailReconstructed->getReturnPath(), $Email->getReturnPath());
+		// $this->assertEquals($emailReconstructed->getMessageId(), $Email->getMessageId());
+		$this->assertEquals($emailReconstructed->getDomain(), $Email->getDomain());
+		$this->assertEquals($emailReconstructed->viewBuilder()
+			->getTheme(), $Email->viewBuilder()
+			->getTheme());
+		$this->assertEquals($emailReconstructed->getProfile(), $Email->getProfile());
+		$this->assertEquals($emailReconstructed->getViewVars(), $Email->getViewVars());
+		$this->assertEquals($emailReconstructed->viewBuilder()
+			->getTemplate(), $Email->viewBuilder()
+			->getTemplate());
+		$this->assertEquals($emailReconstructed->viewBuilder()
+			->getLayout(), $Email->viewBuilder()
+			->getLayout());
+	}
 }

tests/TestCase/Mailer/Transport/QueueTransportTest.php

Indentation +48 added lines, -48 removed lines

@@ -11,57 +11,57 @@
 class QueueTransportTest extends TestCase
 {
 
-    /**
-     * Fixtures to load
-     *
-     * @var array
-     */
-    public $fixtures = [
-        'plugin.Queue.QueuedTasks'
-    ];
+	/**
+	 * Fixtures to load
+	 *
+	 * @var array
+	 */
+	public $fixtures = [
+		'plugin.Queue.QueuedTasks'
+	];
 
-    /**
-     *
-     * @var \Queue\Mailer\Transport\QueueTransport
-     */
-    protected $QueueTransport;
+	/**
+	 *
+	 * @var \Queue\Mailer\Transport\QueueTransport
+	 */
+	protected $QueueTransport;
 
-    /**
-     * Setup
-     *
-     * @return void
-     */
-    public function setUp()
-    {
-        parent::setUp();
-        $this->QueueTransport = new QueueTransport();
-    }
+	/**
+	 * Setup
+	 *
+	 * @return void
+	 */
+	public function setUp()
+	{
+		parent::setUp();
+		$this->QueueTransport = new QueueTransport();
+	}
 
-    /**
-     * TestSend method
-     *
-     * @return void
-     */
-    public function testSendWithEmail()
-    {
-        $Email = new Email();
-        $Email->setFrom('noreply@cakephp.org', 'CakePHP Test');
-        $Email->setTo('cake@cakephp.org', 'CakePHP');
-        $Email->setCc([
-            'mark@cakephp.org' => 'Mark Story',
-            'juan@cakephp.org' => 'Juan Basso'
-        ]);
-        $Email->setBcc('phpnut@cakephp.org');
-        $Email->setSubject('Testing Message');
-        $Email->setTransport('queue');
-        $config = $Email->getConfig('default');
-        $this->QueueTransport->setConfig($config);
+	/**
+	 * TestSend method
+	 *
+	 * @return void
+	 */
+	public function testSendWithEmail()
+	{
+		$Email = new Email();
+		$Email->setFrom('noreply@cakephp.org', 'CakePHP Test');
+		$Email->setTo('cake@cakephp.org', 'CakePHP');
+		$Email->setCc([
+			'mark@cakephp.org' => 'Mark Story',
+			'juan@cakephp.org' => 'Juan Basso'
+		]);
+		$Email->setBcc('phpnut@cakephp.org');
+		$Email->setSubject('Testing Message');
+		$Email->setTransport('queue');
+		$config = $Email->getConfig('default');
+		$this->QueueTransport->setConfig($config);
 
-        $result = $this->QueueTransport->send($Email);
-        $this->assertSame('Email', $result['task']);
-        $this->assertTrue(strlen($result['data']) < 10000);
+		$result = $this->QueueTransport->send($Email);
+		$this->assertSame('Email', $result['task']);
+		$this->assertTrue(strlen($result['data']) < 10000);
 
-        // $output = unserialize($result['data']);
-        // $this->assertEquals('Testing Message', $output['settings']['_subject']);
-    }
+		// $output = unserialize($result['data']);
+		// $this->assertEquals('Testing Message', $output['settings']['_subject']);
+	}
 }

src/Queue/TaskFinder.php

Indentation +68 added lines, -68 removed lines

@@ -8,80 +8,80 @@
 class TaskFinder
 {
 
-    /**
-     *
-     * @var array|null
-     */
-    protected $tasks;
+	/**
+	 *
+	 * @var array|null
+	 */
+	protected $tasks;
 
-    /**
-     * Returns all possible Queue tasks.
-     *
-     * Makes sure that app tasks are prioritized over plugin ones.
-     *
-     * @return array
-     */
-    public function allAppAndPluginTasks()
-    {
-        if ($this->tasks !== null) {
-            return $this->tasks;
-        }
+	/**
+	 * Returns all possible Queue tasks.
+	 *
+	 * Makes sure that app tasks are prioritized over plugin ones.
+	 *
+	 * @return array
+	 */
+	public function allAppAndPluginTasks()
+	{
+		if ($this->tasks !== null) {
+			return $this->tasks;
+		}
 
-        $paths = App::path('Shell/Task');
-        $this->tasks = [];
+		$paths = App::path('Shell/Task');
+		$this->tasks = [];
 
-        foreach ($paths as $path) {
-            $folder = new Folder($path);
-            $this->tasks = $this->getAppPaths($folder);
-        }
-        $plugins = Plugin::loaded();
-        foreach ($plugins as $plugin) {
-            $pluginPaths = App::path('Shell/Task', $plugin);
-            foreach ($pluginPaths as $pluginPath) {
-                $folder = new Folder($pluginPath);
-                $pluginTasks = $this->getPluginPaths($folder, $plugin);
-                $this->tasks = array_merge($this->tasks, $pluginTasks);
-            }
-        }
+		foreach ($paths as $path) {
+			$folder = new Folder($path);
+			$this->tasks = $this->getAppPaths($folder);
+		}
+		$plugins = Plugin::loaded();
+		foreach ($plugins as $plugin) {
+			$pluginPaths = App::path('Shell/Task', $plugin);
+			foreach ($pluginPaths as $pluginPath) {
+				$folder = new Folder($pluginPath);
+				$pluginTasks = $this->getPluginPaths($folder, $plugin);
+				$this->tasks = array_merge($this->tasks, $pluginTasks);
+			}
+		}
 
-        return $this->tasks;
-    }
+		return $this->tasks;
+	}
 
-    /**
-     *
-     * @param \Cake\Filesystem\Folder $folder The directory
-     *
-     * @return array
-     */
-    protected function getAppPaths(Folder $folder)
-    {
-        $res = array_merge($this->tasks, $folder->find('Queue.+\.php'));
-        foreach ($res as &$r) {
-            $r = basename($r, 'Task.php');
-        }
+	/**
+	 *
+	 * @param \Cake\Filesystem\Folder $folder The directory
+	 *
+	 * @return array
+	 */
+	protected function getAppPaths(Folder $folder)
+	{
+		$res = array_merge($this->tasks, $folder->find('Queue.+\.php'));
+		foreach ($res as &$r) {
+			$r = basename($r, 'Task.php');
+		}
 
-        return $res;
-    }
+		return $res;
+	}
 
-    /**
-     *
-     * @param \Cake\Filesystem\Folder $folder The directory
-     * @param string $plugin The plugin name
-     *
-     * @return array
-     */
-    protected function getPluginPaths(Folder $folder, $plugin)
-    {
-        $res = $folder->find('Queue.+Task\.php');
-        foreach ($res as $key => $r) {
-            $name = basename($r, 'Task.php');
-            if (in_array($name, $this->tasks)) {
-                unset($res[$key]);
-                continue;
-            }
-            $res[$key] = $plugin . '.' . $name;
-        }
+	/**
+	 *
+	 * @param \Cake\Filesystem\Folder $folder The directory
+	 * @param string $plugin The plugin name
+	 *
+	 * @return array
+	 */
+	protected function getPluginPaths(Folder $folder, $plugin)
+	{
+		$res = $folder->find('Queue.+Task\.php');
+		foreach ($res as $key => $r) {
+			$name = basename($r, 'Task.php');
+			if (in_array($name, $this->tasks)) {
+				unset($res[$key]);
+				continue;
+			}
+			$res[$key] = $plugin . '.' . $name;
+		}
 
-        return $res;
-    }
+		return $res;
+	}
 }

src/Shell/QueueShell.php

Indentation +505 added lines, -505 removed lines

@@ -30,521 +30,521 @@
 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 void
-     */
-    public function runworker()
-    {
-        // 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);
-            } elseif (Configure::read('Queue.exitWhenNothingToDo')) {
-                $this->out(__d('queue', 'nothing to do, exiting.'));
-                $this->_exit = true;
-            } else {
-                $this->out(__d('queue', '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(__d('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();
-        }
-
-        if ($this->param('verbose')) {
-            $this->_log('endworker');
-        }
-    }
-
-    /**
-     *
-     * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task
-     * @return void
-     */
-    protected function runJob(QueuedTask $queuedTask)
-    {
-        $this->out('Running Job of type "' . $queuedTask->task . '"');
-        $this->_log('job ' . $queuedTask->task . ', id ' . $queuedTask->id, null, 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);
-        } catch (Exception $e) {
-            $return = false;
-
-            $failureMessage = get_class($e) . ': ' . $e->getMessage();
-            $this->_logError($taskName . "\n" . $failureMessage);
-        }
-
-        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);
-            $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 void
+	 */
+	public function runworker()
+	{
+		// 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);
+			} elseif (Configure::read('Queue.exitWhenNothingToDo')) {
+				$this->out(__d('queue', 'nothing to do, exiting.'));
+				$this->_exit = true;
+			} else {
+				$this->out(__d('queue', '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(__d('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();
+		}
+
+		if ($this->param('verbose')) {
+			$this->_log('endworker');
+		}
+	}
+
+	/**
+	 *
+	 * @param \Queue\Model\Entity\QueuedTask $queuedTask Queued task
+	 * @return void
+	 */
+	protected function runJob(QueuedTask $queuedTask)
+	{
+		$this->out('Running Job of type "' . $queuedTask->task . '"');
+		$this->_log('job ' . $queuedTask->task . ', id ' . $queuedTask->id, null, 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);
+		} catch (Exception $e) {
+			$return = false;
+
+			$failureMessage = get_class($e) . ': ' . $e->getMessage();
+			$this->_logError($taskName . "\n" . $failureMessage);
+		}
+
+		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);
+			$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 signal {0}, 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 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;
-    }
-
-    /**
-     *
-     * @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 !== null) {
-            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 signal {0}, 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 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;
+	}
+
+	/**
+	 *
+	 * @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 !== null) {
+			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);
+	}
 }