Oefenweb / cakephp-queue

src/Model/Table/QueuedTasksTable.php

Indentation +591 added lines, -591 removed lines

@@ -28,595 +28,595 @@
 class QueuedTasksTable extends Table
 {
 
-    const DRIVER_MYSQL = 'Mysql';
-
-    const DRIVER_POSTGRES = 'Postgres';
-
-    const DRIVER_SQLSERVER = 'Sqlserver';
-
-    const STATS_LIMIT = 100000;
-
-    /**
-     *
-     * @var string|null
-     */
-    protected $_key;
-
-    /**
-     * 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
-     * @param \ArrayObject $data
-     * @param \ArrayObject $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);
-    }
-
-    /**
-     *
-     * @param string|null $taskName
-     *
-     * @return bool
-     *
-     * @throws \InvalidArgumentException
-     */
-    public function isQueued($taskName = null)
-    {
-        $conditions = [
-            'completed IS' => null
-        ];
-        if ($taskName) {
-            $conditions['task'] = $taskName;
-        }
-
-        return (bool)$this->find()
-            ->where($conditions)
-            ->select([
-            'id'
-        ])
-            ->first();
-    }
-
-    /**
-     * 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 $task
-     * @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 = $this->key();
-            $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
-     *
-     * @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
-     *
-     * @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
-     * @param array $taskConfiguration
-     * @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';
-    }
-
-    /**
-     * Generates a unique Identifier for the current worker thread.
-     *
-     * Useful to identify the currently running processes for this thread.
-     *
-     * @return string Identifier
-     */
-    public function key()
-    {
-        if ($this->_key !== null) {
-            return $this->_key;
-        }
-        $this->_key = sha1(microtime());
-        if (!$this->_key) {
-            throw new RuntimeException('Invalid key generated');
-        }
-
-        return $this->_key;
-    }
-
-    /**
-     * Resets worker Identifier
-     *
-     * @return void
-     */
-    public function clearKey()
-    {
-        $this->_key = null;
-    }
-
-    /**
-     * 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
-     * @param string $key
-     * @param array $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
-     *
-     * @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;
+
+	/**
+	 *
+	 * @var string|null
+	 */
+	protected $_key;
+
+	/**
+	 * 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
+	 * @param \ArrayObject $data
+	 * @param \ArrayObject $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);
+	}
+
+	/**
+	 *
+	 * @param string|null $taskName
+	 *
+	 * @return bool
+	 *
+	 * @throws \InvalidArgumentException
+	 */
+	public function isQueued($taskName = null)
+	{
+		$conditions = [
+			'completed IS' => null
+		];
+		if ($taskName) {
+			$conditions['task'] = $taskName;
+		}
+
+		return (bool)$this->find()
+			->where($conditions)
+			->select([
+			'id'
+		])
+			->first();
+	}
+
+	/**
+	 * 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 $task
+	 * @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 = $this->key();
+			$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
+	 *
+	 * @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
+	 *
+	 * @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
+	 * @param array $taskConfiguration
+	 * @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';
+	}
+
+	/**
+	 * Generates a unique Identifier for the current worker thread.
+	 *
+	 * Useful to identify the currently running processes for this thread.
+	 *
+	 * @return string Identifier
+	 */
+	public function key()
+	{
+		if ($this->_key !== null) {
+			return $this->_key;
+		}
+		$this->_key = sha1(microtime());
+		if (!$this->_key) {
+			throw new RuntimeException('Invalid key generated');
+		}
+
+		return $this->_key;
+	}
+
+	/**
+	 * Resets worker Identifier
+	 *
+	 * @return void
+	 */
+	public function clearKey()
+	{
+		$this->_key = null;
+	}
+
+	/**
+	 * 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
+	 * @param string $key
+	 * @param array $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
+	 *
+	 * @return \Cake\I18n\FrozenTime|\Cake\I18n\Time
+	 */
+	protected function getDateTime($notBefore = null)
+	{
+		if (is_object($notBefore)) {
+			return $notBefore;
+		}
+
+		return new FrozenTime($notBefore);
+	}
 }

Spacing +3 added lines, -3 removed lines

@@ -189,7 +189,7 @@ 
     {
         $driverName = $this->_getDriverName();
         $options = [
-            'fields' => function (Query $query) use ($driverName) {
+            '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))');
@@ -236,7 +236,7 @@ 
     public function getFullStats($taskName = null)
     {
         $driverName = $this->_getDriverName();
-        $fields = function (Query $query) use ($driverName) {
+        $fields = function(Query $query) use ($driverName) {
             $runtime = $query->newExpr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
             switch ($driverName) {
                 case static::DRIVER_SQLSERVER:
@@ -370,7 +370,7 @@ 
         }
 
         /** @var \Queue\Model\Entity\QueuedTask|null $task */
-        $task = $this->getConnection()->transactional(function () use ($query, $options, $now) {
+        $task = $this->getConnection()->transactional(function() use ($query, $options, $now) {
             $task = $query->find('all', $options)
                 ->enableAutoFields(true)
                 ->epilog('FOR UPDATE')

tests/bootstrap.php

Indentation +52 added lines, -52 removed lines

@@ -2,14 +2,14 @@ 
 use Cake\Datasource\ConnectionManager;
 
 if (!defined('DS')) {
-    define('DS', DIRECTORY_SEPARATOR);
+	define('DS', DIRECTORY_SEPARATOR);
 }
 if (!defined('WINDOWS')) {
-    if (DS === '\\' || substr(PHP_OS, 0, 3) === 'WIN') {
-        define('WINDOWS', true);
-    } else {
-        define('WINDOWS', false);
-    }
+	if (DS === '\\' || substr(PHP_OS, 0, 3) === 'WIN') {
+		define('WINDOWS', true);
+	} else {
+		define('WINDOWS', false);
+	}
 }
 
 define('ROOT', dirname(__DIR__));
@@ -32,25 +32,25 @@ 
 require CORE_PATH . 'config/bootstrap.php';
 
 Cake\Core\Configure::write('App', [
-    'namespace' => 'App',
-    'encoding' => 'UTF-8',
-    'paths' => [
-        'templates' => [ROOT . DS . 'tests' . DS . 'test_app' . DS . 'src' . DS . 'Template' . DS],
-    ]
+	'namespace' => 'App',
+	'encoding' => 'UTF-8',
+	'paths' => [
+		'templates' => [ROOT . DS . 'tests' . DS . 'test_app' . DS . 'src' . DS . 'Template' . DS],
+	]
 ]);
 
 Cake\Core\Configure::write('debug', true);
 
 Cake\Core\Configure::write('EmailTransport', [
-    'default' => [
-        'className' => 'Debug',
-    ],
+	'default' => [
+		'className' => 'Debug',
+	],
 ]);
 Cake\Core\Configure::write('Email', [
-    'default' => [
-        'transport' => 'default',
-        'from' => 'you@localhost',
-    ],
+	'default' => [
+		'transport' => 'default',
+		'from' => 'you@localhost',
+	],
 ]);
 
 mb_internal_encoding('UTF-8');
@@ -61,24 +61,24 @@ 
 $Tmp->create(TMP . 'cache/views', 0770);
 
 $cache = [
-    'default' => [
-        'engine' => 'File',
-        'path' => CACHE,
-    ],
-    '_cake_core_' => [
-        'className' => 'File',
-        'prefix' => 'crud_myapp_cake_core_',
-        'path' => CACHE . 'persistent/',
-        'serialize' => true,
-        'duration' => '+10 seconds',
-    ],
-    '_cake_model_' => [
-        'className' => 'File',
-        'prefix' => 'crud_my_app_cake_model_',
-        'path' => CACHE . 'models/',
-        'serialize' => 'File',
-        'duration' => '+10 seconds',
-    ],
+	'default' => [
+		'engine' => 'File',
+		'path' => CACHE,
+	],
+	'_cake_core_' => [
+		'className' => 'File',
+		'prefix' => 'crud_myapp_cake_core_',
+		'path' => CACHE . 'persistent/',
+		'serialize' => true,
+		'duration' => '+10 seconds',
+	],
+	'_cake_model_' => [
+		'className' => 'File',
+		'prefix' => 'crud_my_app_cake_model_',
+		'path' => CACHE . 'models/',
+		'serialize' => 'File',
+		'duration' => '+10 seconds',
+	],
 ];
 
 Cake\Cache\Cache::setConfig($cache);
@@ -86,37 +86,37 @@ 
 Cake\Core\Plugin::getCollection()->add(new \Queue\Plugin());
 
 Cake\Mailer\TransportFactory::setConfig('default', [
-    'className' => 'Debug',
+	'className' => 'Debug',
 ]);
 Cake\Mailer\TransportFactory::setConfig('queue', [
-    'className' => 'Queue.Queue',
+	'className' => 'Queue.Queue',
 ]);
 Cake\Mailer\Email::setConfig('default', [
-    'transport' => 'default',
+	'transport' => 'default',
 ]);
 
 // Allow local overwrite
 // E.g. in your console: export db_dsn="mysql://root:secret@127.0.0.1/cake_test"
 if (!getenv('db_class') && getenv('db_dsn')) {
-    ConnectionManager::setConfig('test', ['url' => getenv('db_dsn')]);
+	ConnectionManager::setConfig('test', ['url' => getenv('db_dsn')]);
 
-    return;
+	return;
 }
 if (!getenv('db_class')) {
-    putenv('db_class=Cake\Database\Driver\Sqlite');
-    putenv('db_dsn=sqlite::memory:');
+	putenv('db_class=Cake\Database\Driver\Sqlite');
+	putenv('db_dsn=sqlite::memory:');
 }
 
 // Uses Travis config then (MySQL, Postgres, ...)
 ConnectionManager::setConfig('test', [
-    'className' => 'Cake\Database\Connection',
-    'driver' => getenv('db_class'),
-    'dsn' => getenv('db_dsn'),
-    'database' => getenv('db_database'),
-    'username' => getenv('db_username'),
-    'password' => getenv('db_password'),
-    'timezone' => 'UTC',
-    'quoteIdentifiers' => true,
-    'cacheMetadata' => true,
+	'className' => 'Cake\Database\Connection',
+	'driver' => getenv('db_class'),
+	'dsn' => getenv('db_dsn'),
+	'database' => getenv('db_database'),
+	'username' => getenv('db_username'),
+	'password' => getenv('db_password'),
+	'timezone' => 'UTC',
+	'quoteIdentifiers' => true,
+	'cacheMetadata' => true,
 
 ]);

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.');
+	}
 }