Dtc\QueueBundle\Model\WorkerManager

Complexity

Total Complexity

19

Size/Duplication

Total Lines	128
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

69.39%

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	1
A	addWorker()	0	12	3
A	getWorker()	0	8	2
A	getWorkers()	0	4	1
A	setLoggingFunc()	0	4	1
A	log()	0	14	3
A	run()	0	9	2
B	runJob()	0	47	6

<?php

namespace Dtc\QueueBundle\Model;

use Dtc\QueueBundle\EventDispatcher\Event;
use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
use Monolog\Logger;

class WorkerManager
{
    protected $workers;
    protected $jobManager;
    protected $logger;
    protected $eventDispatcher;
    protected $logFunc;

    public function __construct(JobManagerInterface $jobManager, EventDispatcher $eventDispatcher, Logger $logger = null)
    {
        $this->workers = array();
        $this->jobManager = $jobManager;
        $this->logger = $logger;
        $this->eventDispatcher = $eventDispatcher;
    }

    public function addWorker(Worker $worker)
    {
        if ($this->logger) {
            $this->logger->debug("Added worker: {$worker->getName()}");
        }

        if (isset($this->workers[$worker->getName()])) {
            throw new \Exception("{$worker->getName()} already exists in worker manager");
        }

        $this->workers[$worker->getName()] = $worker;
    }

    public function getWorker($name)
    {
        if (isset($this->workers[$name])) {
            return $this->workers[$name];
        }

        return null;
    }

    public function getWorkers()
    {
        return $this->workers;
    }

    public function setLoggingFunc(callable $callable)
    {
        $this->logFunc = $callable;
    }

    public function log($level, $msg, array $context = [])
    {
        if ($this->logFunc) {
            call_user_func_array($this->logFunc, [$level, $msg, $context]);

            return;
        }

        if ($this->logger) {
            $this->logger->$level($msg, $context);

            return;
        }
    }

    /**
     * @param null $workerName
     * @param null $methodName
     * @param bool $prioritize
     *
     * @return null|Job
     */
    public function run($workerName = null, $methodName = null, $prioritize = true, $runId = null)
    {
        $job = $this->jobManager->getJob($workerName, $methodName, $prioritize, $runId);
        if (!$job) {
            return null; // no job to run
        }

        return $this->runJob($job);
    }

    public function runJob(Job $job)
    {
        $event = new Event($job);
        $this->eventDispatcher->dispatch(Event::PRE_JOB, $event);

        $start = microtime(true);
        $handleException = function($exception) use ($job) {
            /** @var \Exception $exception */
            $exceptionMessage = get_class($exception)."\n".$exception->getCode().' - '.$exception->getMessage()."\n".$exception->getTraceAsString();
            $this->log('debug', "Failed: {$job->getClassName()}->{$job->getMethod()}");
            $job->setStatus(BaseJob::STATUS_ERROR);
            if ($job instanceof RetryableJob) {
                $job->setErrorCount($job->getErrorCount() + 1);
                if (null !== ($maxError = $job->getMaxError()) && $job->getErrorCount() >= $maxError) {
                    $job->setStatus(RetryableJob::STATUS_MAX_ERROR);
                }
            }
            $job->setMessage($exceptionMessage);
        };
        try {
            $worker = $this->getWorker($job->getWorkerName());
            $this->log('debug', "Start: {$job->getClassName()}->{$job->getMethod()}", $job->getArgs());
            $job->setStartedAt(new \DateTime());
            call_user_func_array(array($worker, $job->getMethod()), $job->getArgs());

            // Job finshed successfuly... do we remove the job from database?
            $job->setStatus(BaseJob::STATUS_SUCCESS);
            $job->setMessage(null);
        } catch (\Throwable $exception) {
            $handleException($exception);
        } catch (\Exception $exception) {
            $handleException($exception);
        }

        // save Job history
        $elapsed = microtime(true) - $start;
        $job->setFinishedAt(new \DateTime());
        $job->setElapsed($elapsed);

        $this->log('debug', "Finished: {$job->getClassName()}->{$job->getMethod()} in {$elapsed} micro-seconds");
        $this->log('debug', "Save job history: {$job->getId()}");

        $this->jobManager->saveHistory($job);
        $this->eventDispatcher->dispatch(Event::POST_JOB, $event);

        return $job;
    }
}