Waca\ConsoleTasks\RunJobQueueTask::execute()

Complexity

Conditions	9
Paths	2052

Size

Total Lines	101
Code Lines	54

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	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' => $this->getSiteConfiguration()->getJobQueueBatchSize()
        ));

        /** @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 {
                // refresh from the database
                /** @var JobQueue $job */
                $job = JobQueue::getById($job->getId(), $database);

                if ($job->getStatus() !== JobQueue::STATUS_WAITING) {
                    continue;
                }

                $database->beginTransaction();
                $job->setStatus(JobQueue::STATUS_RUNNING);
                $job->save();
                $database->commit();

                $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();
                try {
                    $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();

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

                    $database->commit();
                }
                catch (Exception $ex) {
                    // oops, something went horribly wrong trying to handle this in a nice way; let's just fall back to
                    // logging this to disk for a tool root to investigate.
                    ExceptionHandler::logExceptionToDisk($ex, $this->getSiteConfiguration());
                }
            }
            finally {
                $database->commit();
            }
        }

        $this->stageQueuedTasks($database);
    }