Waca\ConsoleTasks\RunJobQueueTask::execute()

Complexity

Conditions	8
Paths	1244

Size

Total Lines

82

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

<?php
    public function execute()
    {
        $database = $this->getDatabase();

        // ensure we're running inside a tx here.
        if (!$database->hasActiveTransaction()) {
            $database->beginTransaction();
        }

        $sql = 'SELECT * FROM jobqueue WHERE status = :status ORDER BY enqueue LIMIT :lim';
        $statement = $database->prepare($sql);
        $statement->execute(array(':status' => JobQueue::STATUS_READY, ':lim' => 10));
        /** @var JobQueue[] $queuedJobs */
        $queuedJobs = $statement->fetchAll(PDO::FETCH_CLASS, JobQueue::class);

        // mark all the jobs as running, and commit the txn so we're not holding onto long-running transactions.
        // We'll re-lock the row when we get to it.
        foreach ($queuedJobs as $job) {
            $job->setDatabase($database);
            $job->setStatus(JobQueue::STATUS_WAITING);
            $job->setError(null);
            $job->setAcknowledged(null);
            $job->save();
        }

        $database->commit();

        set_error_handler(array(RunJobQueueTask::class, 'errorHandler'), E_ALL);

        foreach ($queuedJobs as $job) {
            try {
                $database->beginTransaction();

                // re-lock the job
                $job->setStatus(JobQueue::STATUS_RUNNING);
                $job->save();

                // validate we're allowed to run the requested task (whitelist)
                if (!in_array($job->getTask(), $this->taskList)) {
                    throw new ApplicationLogicException('Job task not registered');
                }

                // Create a task.
                $taskName = $job->getTask();

                if(!class_exists($taskName)) {
                    throw new ApplicationLogicException('Job task does not exist');
                }

                /** @var BackgroundTaskBase $task */
                $task = new $taskName;

                $this->setupTask($task, $job);
                $task->run();
            }
            catch (Exception $ex) {
                $database->rollBack();
                $database->beginTransaction();

                /** @var JobQueue $job */
                $job = JobQueue::getById($job->getId(), $database);
                $job->setDatabase($database);
                $job->setStatus(JobQueue::STATUS_FAILED);
                $job->setError($ex->getMessage());
                $job->setAcknowledged(0);
                $job->save();

                /** @var Request $request */
                $request = Request::getById($job->getRequest(), $database);
                if ($request === false) {
                    $request = null;
                }

                Logger::backgroundJobIssue($this->getDatabase(), $job);

                $database->commit();
            }
            finally {
                $database->commit();
            }
        }
    }

}