Queue\Model\Table\QueuedTasksTable

Complexity

Total Complexity

55

Size/Duplication

Total Lines	603
Duplicated Lines	0 %

Importance

Changes

0

22 Methods

Rating	Name	Duplication	Size	Complexity
A	getDateTime()	0	7	2
A	getTypes()	0	13	1
A	truncate()	0	5	2
A	clearKey()	0	3	1
B	getFullStats()	0	67	9
A	_getDriverName()	0	6	1
A	isQueued()	0	15	2
A	key()	0	11	3
A	getLength()	0	12	2
A	cleanOldJobs()	0	8	2
B	requestJob()	0	84	7
A	initialize()	0	9	1
A	getStats()	0	35	2
A	getFailedStatus()	0	14	3
A	defaultConnectionName()	0	8	2
A	markJobFailed()	0	9	1
A	rerun()	0	15	1
A	beforeMarshal()	0	4	3
A	markJobDone()	0	8	1
A	reset()	0	17	2
A	createJob()	0	15	2
A	addFilter()	0	20	5

<?php
namespace Queue\Model\Table;

use ArrayObject;
use Cake\Core\Configure;
use Cake\Event\Event;
use Cake\I18n\FrozenTime;
use Cake\ORM\Query;
use Cake\ORM\Table;
use InvalidArgumentException;
use Queue\Model\Entity\QueuedTask;
use RuntimeException;

/**
 * QueuedTasks Model
 *
 * @method \Queue\Model\Entity\QueuedTask get($primaryKey, $options = [])
 * @method \Queue\Model\Entity\QueuedTask newEntity($data = null, array $options = [])
 * @method \Queue\Model\Entity\QueuedTask[] newEntities(array $data, array $options = [])
 * @method \Queue\Model\Entity\QueuedTask|bool save(\Cake\Datasource\EntityInterface $entity, $options = [])
 * @method \Queue\Model\Entity\QueuedTask saveOrFail(\Cake\Datasource\EntityInterface $entity, $options = [])
 * @method \Queue\Model\Entity\QueuedTask patchEntity(\Cake\Datasource\EntityInterface $entity, array $data, array $options = [])
 * @method \Queue\Model\Entity\QueuedTask[] patchEntities($entities, array $data, array $options = [])
 * @method \Queue\Model\Entity\QueuedTask findOrCreate($search, callable $callback = null, $options = [])
 *
 * @mixin \Cake\ORM\Behavior\TimestampBehavior
 */
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)

The parameter $options is not used and could be removed.

The parameter $event is not used and could be removed.

    {
        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) {

The expression $types of type array is implicitly converted to a boolean; are you sure this is intended? If so, consider using ! empty($expr) instead to make it clear that you intend to check for an array without elements.

            $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, [

It seems like $task can also be of type array; however, parameter $entity of Queue\Model\Table\QueuedTasksTable::patchEntity() does only seem to accept Cake\Datasource\EntityInterface, maybe add an additional type check?

                '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) {

The expression $id of type integer|null is loosely compared to true; this is ambiguous if the integer can be 0. You might want to explicitly use !== null instead.

            $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);
    }
}