JobMonitor::afterExec() - Code Metrics - Inspection of "Refactoring" - zhuravljov/yii2-queue-monitor - Measure and Improve Code Quality continuously with Scrutinizer

Passed
Push — master ( 62ccc4...3821b3 )

by Roman
created 2019-04-28 02:29 UTC
JobMonitor::afterExec() B

↳ Parent: JobMonitor
Complexity

Conditions	8
Paths	6
Size

Total Lines	22
Code Lines	15
Duplication

Lines	0
Ratio	0 %
Importance

Changes
Metric	Value
eloc	15
dl	0
loc	22
rs	8.4444
c	0
b	0
f	0
cc	8
nc	6
nop	1
<?php
/**
 * @link https://github.com/zhuravljov/yii2-queue-monitor
 * @copyright Copyright (c) 2017 Roman Zhuravlev
 * @license http://opensource.org/licenses/BSD-3-Clause
 */

namespace zhuravljov\yii\queue\monitor;

use Yii;
use yii\base\Behavior;
use yii\base\InvalidConfigException;
use yii\helpers\ArrayHelper;
use yii\helpers\VarDumper;
use yii\queue\ExecEvent;
use yii\queue\JobEvent;
use yii\queue\JobInterface;
use yii\queue\PushEvent;
use yii\queue\Queue;
use zhuravljov\yii\queue\monitor\records\ExecRecord;
use zhuravljov\yii\queue\monitor\records\PushRecord;
use zhuravljov\yii\queue\monitor\records\WorkerRecord;

/**
 * Queue Job Monitor
 *
 * @author Roman Zhuravlev <[email protected]>
 */
class JobMonitor extends Behavior
{
    /**
     * @var array of job class names that this behavior should tracks.
     * @since 0.3.2
     */
    public $only = [];
    /**
     * @var array of job class names that this behavior should not tracks.
     * @since 0.3.2
     */
    public $except = [];
    /**
     * @var array
     */
    public $contextVars = [
        '_SERVER.argv',
        '_SERVER.REQUEST_METHOD',
        '_SERVER.REQUEST_URI',
        '_SERVER.HTTP_REFERER',
        '_SERVER.HTTP_USER_AGENT',
        '_POST',
    ];
    /**
     * @var Queue
     * @inheritdoc
     */
    public $owner;
    /**
     * @var Env
     */
    protected $env;
    /**
     * @var null|PushRecord
     */
    protected static $startedPush;

    /**
     * @param Env $env
     * @param array $config
     */
    public function __construct(Env $env, $config = [])
    {
        $this->env = $env;
        parent::__construct($config);
    }

    /**
     * @inheritdoc
     */
    public function events()
    {
        return [
            Queue::EVENT_AFTER_PUSH => 'afterPush',
            Queue::EVENT_BEFORE_EXEC => 'beforeExec',
            Queue::EVENT_AFTER_EXEC => 'afterExec',
            Queue::EVENT_AFTER_ERROR => 'afterExec',
        ];
    }

    /**
     * @param PushEvent $event
     */
    public function afterPush(PushEvent $event)
    {
        if (!$this->isActive($event->job)) {
        if (!$this->isActive(/** @scrutinizer ignore-type */ $event->job)) {
            return;
        }

        if ($this->env->db->getTransaction()) {
            // create new database connection, if there is an open transaction
            // to ensure insert statement is not affected by a rollback
            $this->env->db = clone $this->env->db;
        }

        $push = new PushRecord();
        $push->parent_id = static::$startedPush ? static::$startedPush->id : null;
        $push->sender_name = $this->getSenderName($event);
        $push->job_uid = $event->id;
        $push->setJob($event->job);
        $push->ttr = $event->ttr;
        $push->delay = $event->delay;
        $push->trace = (new \Exception())->getTraceAsString();
        $push->context = $this->getContext();
        $push->pushed_at = time();
        $push->save(false);
    }

    /**
     * @param ExecEvent $event
     */
    public function beforeExec(ExecEvent $event)
    {
        if (!$this->isActive($event->job)) {
        if (!$this->isActive(/** @scrutinizer ignore-type */ $event->job)) {
            return;
        }
        static::$startedPush = $push = $this->getPushRecord($event);
        if (!$push) {

            return;
        }
        if ($push->isStopped()) {
            // Rejects job execution in case is stopped
            $event->handled = true;
            return;
        }
        $this->env->db->transaction(function () use ($event, $push) {
            $worker = $this->getWorkerRecord($event);
class A
{
    function getObject()
    {
        return null;
    }

}

$a = new A();
$object = $a->getObject();

            $exec = new ExecRecord();
            $exec->push_id = $push->id;
            if ($worker) {

                $exec->worker_id = $worker->id;
            }
            $exec->attempt = $event->attempt;
            $exec->started_at = time();
            $exec->save(false);

            $push->first_exec_id = $push->first_exec_id ?: $exec->id;
            $push->last_exec_id = $exec->id;
            $push->save(false);

            if ($worker) {

                $worker->last_exec_id = $exec->id;
                $worker->save(false);
            }
        });
    }

    /**
     * @param ExecEvent $event
     */
    public function afterExec(ExecEvent $event)
    {
        if (!$this->isActive($event->job)) {
        if (!$this->isActive(/** @scrutinizer ignore-type */ $event->job)) {
            return;
        }
        $push = static::$startedPush ?: $this->getPushRecord($event);
        if (!$push) {

            return;
        }
        if ($push->isStopped()) {
            // Breaks retry in case is stopped
            $event->retry = false;
        }
        if ($push->last_exec_id) {
0   == false // true
0   == null  // true
123 == false // false
123 == null  // false

// It is often better to use strict comparison
0 === false // false
0 === null  // false
            ExecRecord::updateAll([
                'finished_at' => time(),
                'memory_usage' => static::$startedPush ? memory_get_peak_usage() : null,
                'error' => $event->error,
                'result_data' => $event->result !== null ? serialize($event->result) : null,
                'retry' => $event->retry,
            ], [
                'id' => $push->last_exec_id
            ]);
        }
    }

    /**
     * @param JobInterface $job
     * @return bool
     * @since 0.3.2
     */
    protected function isActive(JobInterface $job)
    {
        $onlyMatch = true;
        if ($this->only) {

            $onlyMatch = false;
            foreach ($this->only as $className) {
                if (is_a($job, $className)) {
                    $onlyMatch = true;
                    break;
                }
            }
        }

        $exceptMatch = false;
        foreach ($this->except as $className) {
            if (is_a($job, $className)) {
                $exceptMatch = true;
                break;
            }
        }

        return !$exceptMatch && $onlyMatch;
    }

    /**
     * @param JobEvent $event
     * @throws
     * @return string
     */
    protected function getSenderName($event)
    {
        foreach (Yii::$app->getComponents(false) as $id => $component) {
            if ($component === $event->sender) {
                return $id;
            }
        }
        throw new InvalidConfigException('Queue must be an application component.');
    }

    /**
     * @return string
     */
    protected function getContext()
    {
        $context = ArrayHelper::filter($GLOBALS, $this->contextVars);
        $result = [];
        foreach ($context as $key => $value) {
            $result[] = "\${$key} = " . VarDumper::dumpAsString($value);
        }

        return implode("\n\n", $result);
    }

    /**
     * @param JobEvent $event
     * @return PushRecord
     */
    protected function getPushRecord(JobEvent $event)
    {
        if ($event->id !== null) {
            return $this->env->db->useMaster(function () use ($event) {
                return PushRecord::find()
                    ->byJob($this->getSenderName($event), $event->id)
                    ->one();
            });
        } else {
            return null;
        }
    }

    /**
     * @param ExecEvent $event
     * @return WorkerRecord|null
     */
    protected function getWorkerRecord(ExecEvent $event)
    {
        if ($event->sender->getWorkerPid() === null) {
class A
{
    function getObject()
    {
        return null;
    }

}

$a = new A();
if ($a->getObject()) {
            return null;
        }
        if (!$this->isWorkerMonitored()) {
            return null;
        }

        return $this->env->db->useMaster(function () use ($event) {
            return WorkerRecord::find()
                ->byEvent($this->env->getHost(), $event->sender->getWorkerPid())
                ->active()
                ->one();
        });
    }

    /**
     * @return bool whether workers are monitored.
     */
    private function isWorkerMonitored()
    {
        foreach ($this->owner->getBehaviors() as $behavior) {
            if ($behavior instanceof WorkerMonitor) {
                return true;
            }
        }
        return false;
    }
}

zhuravljov / yii2-queue-monitor

Push — master ( 62ccc4...3821b3 )

JobMonitor::afterExec() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
