mmucklo / DtcQueueBundle

Doctrine/StalledTrait.php

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

It seems like saveHistory() must be provided by classes using this trait. How about adding it as abstract method to this trait?

                ++$resetCount;
            }
        }

        return $resetCount;
    }

    /**
     * @param int $limit
     * @param int $offset
     */
    private function resetJobsByCriterion(
        array $criterion,
        $limit,
        $offset
    ) {
        $objectManager = $this->getObjectManager();

It seems like getObjectManager() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        $objectName = $this->getJobClass();

It seems like getJobClass() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        $archiveObjectName = $this->getJobArchiveClass();

It seems like getJobArchiveClass() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        $jobRepository = $objectManager->getRepository($objectName);
        $jobArchiveRepository = $objectManager->getRepository($archiveObjectName);
        $className = $jobRepository->getClassName();
        $metadata = $objectManager->getClassMetadata($className);
        $this->stopIdGenerator($objectName);

It seems like stopIdGenerator() must be provided by classes using this trait. How about adding it as abstract method to this trait?

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

It seems like restoreIdGenerator() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        return $countProcessed;
    }

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

It seems like countJobsByStatus() must be provided by classes using this trait. How about adding it as abstract method to this trait?

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

It seems like addWorkerNameMethod() must be provided by classes using this trait. How about adding it as abstract method to this trait?

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

It seems like updateMaxStatus() must be provided by classes using this trait. How about adding it as abstract method to this trait?

            $objectManager->persist($jobArchive);

            return 0;
        }

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

The method resetJob() does not exist on Dtc\QueueBundle\Doctrine\StalledTrait. Did you maybe mean resetJobsByCriterion()?

        $objectManager->remove($jobArchive);
        $this->flush();

It seems like flush() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        return 1;
    }

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

It seems like getRepository() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        $runningJobsById = [];

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

        $fetchCount = $this->getFetchCount($count);

It seems like getFetchCount() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        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();

It seems like getRunManager() must be provided by classes using this trait. How about adding it as abstract method to this trait?

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

The constant Dtc\QueueBundle\Doctrine\StalledTrait::STALLED_SECONDS was not found. Maybe you did not declare it correctly or list all dependencies?

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