zhuravljov\yii\queue\monitor\JobMonitor

Complexity

Total Complexity

27

Size/Duplication

Total Lines	195
Duplicated Lines	0 %

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
B	beforeExec()	0	30	6
A	events()	0	7	1
A	afterPush()	0	10	1
A	isWorkerMonitored()	0	8	3
A	afterError()	0	17	4
A	getPushRecord()	0	10	2
A	afterExec()	0	13	3
A	getSenderName()	0	8	3
A	getWorkerRecord()	0	14	3
A	__construct()	0	4	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\queue\ErrorEvent;
use yii\queue\ExecEvent;
use yii\queue\JobEvent;
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 <zhuravljov@gmail.com>
 */
class JobMonitor extends Behavior
{
    /**
     * @var Queue
     * @inheritdoc
     */
    public $owner;
    /**
     * @var Env
     */
    protected $env;

    /**
     * @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 => 'afterError',
        ];
    }

    /**
     * @param PushEvent $event
     */
    public function afterPush(PushEvent $event)
    {
        $push = new PushRecord();
        $push->sender_name = $this->getSenderName($event);
        $push->job_uid = $event->id;
        $push->setJob($event->job);
        $push->ttr = $event->ttr;
        $push->delay = $event->delay;
        $push->pushed_at = time();
        $push->save(false);
    }

    /**
     * @param ExecEvent $event
     */
    public function beforeExec(ExecEvent $event)
    {
        $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);

Are you sure the assignment to $worker is correct as $this->getWorkerRecord($event) targeting zhuravljov\yii\queue\monitor\JobMonitor::getWorkerRecord() seems to always return null.

            $exec = new ExecRecord();
            $exec->push_id = $push->id;
            if ($worker) {
                $exec->worker_id = $worker->id;
            }
            $exec->attempt = $event->attempt;
            $exec->reserved_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)
    {
        $push = $this->getPushRecord($event);
        if (!$push) {
            return;
        }
        if ($push->last_exec_id) {
            ExecRecord::updateAll([
                'done_at' => time(),
                'error' => null,
                'retry' => false,
            ], [
                'id' => $push->last_exec_id
            ]);
        }
    }

    /**
     * @param ErrorEvent $event
     */
    public function afterError(ErrorEvent $event)
    {
        $push = $this->getPushRecord($event);
        if (!$push) {
            return;
        }
        if ($push->isStopped()) {
            // Breaks retry in case is stopped
            $event->retry = false;
        }
        if ($push->last_exec_id) {
            ExecRecord::updateAll([
                'done_at' => time(),
                'error' => $event->error,
                'retry' => $event->retry,
            ], [
                'id' => $push->last_exec_id
            ]);
        }
    }

    /**
     * @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.');
    }

    /**
     * @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) {

The condition $event->sender->getWorkerPid() === null can never be false.

Are you sure the usage of $event->sender->getWorkerPid() targeting yii\queue\Queue::getWorkerPid() seems to always return null.

            return null;
        }
        if (!$this->isWorkerMonitored()) {
            return null;
        }

        return $this->env->db->useMaster(function () use ($event) {
            return WorkerRecord::find()
                ->byPid($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;
    }
}