StalledTrait::runStalledLoop() - Code Metrics - mmucklo/DtcQueueBundle - Measure and Improve Code Quality continuously with Scrutinizer

StalledTrait::runStalledLoop() A
last analyzed 2025-06-22 22:56 UTC

↳ Parent: StalledTrait
Complexity

Conditions	4
Paths	3
Size

Total Lines	13
Code Lines	7
Duplication

Lines	0
Ratio	0 %
Code Coverage

Tests	8
CRAP Score	4
Importance

Changes	1
Bugs	0	Features	0
Metric	Value
cc	4
eloc	7
c	1
b	0
f	0
nc	3
nop	4
dl	0
loc	13
ccs	8
cts	8
cp	1
crap	4
rs	10
<?php

namespace Dtc\QueueBundle\Doctrine;

use Dtc\QueueBundle\Model\BaseJob;
use Dtc\QueueBundle\Model\StallableJob;
use Dtc\QueueBundle\Util\Util;

trait StalledTrait
{
    /**
     * @param int $if
     * @param int $count
     * @param int $saveCount
     *
     * @return int
     */
    protected function runStalledLoop($i, $count, $saveCount, array $stalledJobs)
    {
        $resetCount = 0;
        for ($j = $i, $max = $i + $saveCount; $j < $max && $j < $count; ++$j) {
            /* StallableJob $job */
            $job = $stalledJobs[$j];
            $job->setStatus(StallableJob::STATUS_STALLED);
            if ($this->saveHistory($job)) {
            if ($this->/** @scrutinizer ignore-call */ saveHistory($job)) {
                ++$resetCount;
            }
        }

        return $resetCount;
    }

    /**
     * @param int $limit
     * @param int $offset
     */
    private function resetJobsByCriterion(
        array $criterion,
        $limit,
        $offset
    ) {
        $objectManager = $this->getObjectManager();
        $objectManager = $this->getObjectManager();
        $objectName = $this->getJobClass();
        $objectName = $this->getJobClass();
        $archiveObjectName = $this->getJobArchiveClass();
        $archiveObjectName = $this->getJobArchiveClass();
        $jobRepository = $objectManager->getRepository($objectName);
        $jobArchiveRepository = $objectManager->getRepository($archiveObjectName);
        $className = $jobRepository->getClassName();
        $metadata = $objectManager->getClassMetadata($className);
        $this->stopIdGenerator($objectName);
               stopIdGenerator($objectName);
        $identifierData = $metadata->getIdentifier();
        $idColumn = isset($identifierData[0]) ? $identifierData[0] : 'id';
        $results = $jobArchiveRepository->findBy($criterion, [$idColumn => 'ASC'], $limit, $offset);
        $countProcessed = 0;

        foreach ($results as $jobArchive) {
            $countProcessed += $this->resetArchiveJob($jobArchive);
        }
        $objectManager->flush();

        $this->restoreIdGenerator($objectName);
               restoreIdGenerator($objectName);

        return $countProcessed;
    }

    protected function getStalledJobs($workerName = null, $method = null)
    {
        $count = $this->countJobsByStatus($this->getJobClass(), BaseJob::STATUS_RUNNING, $workerName, $method);
        $count = $this->countJobsByStatus($this->getJobClass(), BaseJob::STATUS_RUNNING, $workerName, $method);

        $criterion = ['status' => BaseJob::STATUS_RUNNING];
        $this->addWorkerNameMethod($criterion, $workerName, $method);
               addWorkerNameMethod($criterion, $workerName, $method);

        $runningJobs = $this->findRunningJobs($criterion, $count);

        return $this->extractStalledJobs($runningJobs);
    }

    /**
     * @return int Number of jobs reset
     */
    protected function resetArchiveJob(StallableJob $jobArchive)
    {
        $objectManager = $this->getObjectManager();
        if ($this->updateMaxStatus($jobArchive, StallableJob::STATUS_MAX_RETRIES, $jobArchive->getMaxRetries(), $jobArchive->getRetries())) {
        if ($this->/** @scrutinizer ignore-call */ updateMaxStatus($jobArchive, StallableJob::STATUS_MAX_RETRIES, $jobArchive->getMaxRetries(), $jobArchive->getRetries())) {
            $objectManager->persist($jobArchive);

            return 0;
        }

        /** @var StallableJob $job */
        $className = $this->getJobClass();
        $newJob = new $className();
        Util::copy($jobArchive, $newJob);
        $this->resetJob($newJob);
               resetJob($newJob);
        $objectManager->remove($jobArchive);
        $this->flush();
               flush();

        return 1;
    }

    protected function findRunningJobs($criterion, $count)
    {
        $repository = $this->getRepository();
        $repository = $this->getRepository();
        $runningJobsById = [];

        $metadata = $this->getObjectManager()->getClassMetadata($this->getJobClass());
        $identifierData = $metadata->getIdentifier();
        $idColumn = isset($identifierData[0]) ? $identifierData[0] : 'id';

        $fetchCount = $this->getFetchCount($count);
        $fetchCount = $this->getFetchCount($count);
        for ($i = 0; $i < $count; $i += $fetchCount) {
            $runningJobs = $repository->findBy($criterion, [$idColumn => 'ASC'], $fetchCount, $i);
            if (!empty($runningJobs)) {
                foreach ($runningJobs as $job) {
                    /** @var StallableJob $job */
                    $runId = $job->getRunId();
                    $runningJobsById[$runId][] = $job;
                }
            }
        }

        return $runningJobsById;
    }

    private function extractStalledJobsRunArchive(array $runningJobsById, array &$stalledJobs, $runId)
    {
        $runManager = $this->getRunManager();
        $runManager = $this->getRunManager();
        if (!method_exists($runManager, 'getObjectManager')) {
            return;
        }
        if (!method_exists($runManager, 'getRunArchiveClass')) {
            return;
        }

        /** @var EntityRepository|DocumentRepository $runArchiveRepository */
        $runArchiveRepository = $runManager->getObjectManager()->getRepository($runManager->getRunArchiveClass());
        /** @var Run $run */
        if ($run = $runArchiveRepository->find($runId)) {
            if ($endTime = $run->getEndedAt()) {
                // Did it end over an hour ago
                if ((time() - $endTime->getTimestamp()) > static::STALLED_SECONDS) {

                    $stalledJobs = array_merge($stalledJobs, $runningJobsById[$runId]);
                }
            }
        }
    }

    /**
     * @return array
     */
    private function extractStalledJobs(array $runningJobsById)
    {
        $stalledJobs = [];
        foreach (array_keys($runningJobsById) as $runId) {
            if (!$runId && 0 !== $runId) {
                $stalledJobs = array_merge($stalledJobs, $runningJobsById[$runId]);
                continue;
            }
            $this->extractStalledLiveRuns($runId, $runningJobsById[$runId], $stalledJobs);
            $this->extractStalledJobsRunArchive($runningJobsById, $stalledJobs, $runId);
        }

        return $stalledJobs;
    }

    /**
     * @param $runId
     */
    private function extractStalledLiveRuns($runId, array $jobs, array &$stalledJobs)
    {
        $objectManager = $this->getObjectManager();
        $runRepository = $objectManager->getRepository($this->getRunManager()->getRunClass());
        if ($run = $runRepository->find($runId)) {
            foreach ($jobs as $job) {
                if ($run->getCurrentJobId() == $job->getId()) {
                    continue;
                }
                $stalledJobs[] = $job;
            }
        }
    }
}

mmucklo / DtcQueueBundle

StalledTrait::runStalledLoop() A last analyzed 2025-06-22 22:56 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

StalledTrait::runStalledLoop() A
last analyzed 2025-06-22 22:56 UTC
