BrainExe\Core\MessageQueue\Worker

Complexity

Total Complexity

7

Size/Duplication

Total Lines	93
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	1

Test Coverage

Coverage

100%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	executeJob()	0	9	2
A	handleCronEvent()	0	12	2
B	execute()	0	31	2

<?php

namespace BrainExe\Core\MessageQueue;

use BrainExe\Core\Annotations\Service;
use BrainExe\Core\Cron\Expression;
use BrainExe\Core\EventDispatcher\AbstractEvent;
use BrainExe\Core\EventDispatcher\CronEvent;
use BrainExe\Core\EventDispatcher\JobEvent;
use BrainExe\Core\Traits\EventDispatcherTrait;
use BrainExe\Core\Traits\LoggerTrait;
use BrainExe\Core\Traits\TimeTrait;

use Throwable;

/**
 * @Service
 */
class Worker
{

    use LoggerTrait;
    use TimeTrait;
    use EventDispatcherTrait;

    /**
     * @var Gateway
     */
    private $gateway;

    /**
     * @var Expression
     */
    private $cron;

    /**
     * @param Gateway $gateway
     * @param Expression $cron
     */
    public function __construct(Gateway $gateway, Expression $cron)
    {
        $this->gateway = $gateway;
        $this->cron    = $cron;
    }

    /**
     * @param Job $job
     */
    public function executeJob(Job $job)
    {
        try {
            $this->execute($job);
        } catch (Throwable $e) {
            $this->error($e->getMessage(), ['exception' => $e]);
            $this->gateway->restoreJob($job);
        }
    }

    /**
     * @param Job $job
     */
    private function execute(Job $job)
    {
        $event = $job->getEvent();
        if ($event instanceof CronEvent) {
            $event = $this->handleCronEvent($job, $event);
        }

        $logStart = microtime(true);

        $this->dispatchEvent($event);

        $neededTime = microtime(true) - $logStart;

        $this->info(
            sprintf(
                '[MQ]: %s. Time: %0.2fms',
                $event->getEventName(),
                $neededTime * 1000
            ),
            [
                'channel'   => 'message_queue',
                'time'      => round($neededTime * 1000, 2),
                'eventName' => $event->getEventName(),
                'jobId'     => $job->getJobId(),
                'event'     => json_encode($event)
            ]
        );

        $handledEvent = new JobEvent(JobEvent::HANDLED, $job);
        $this->dispatcher->dispatchEvent($handledEvent);
    }

    /**
     * @param Job $job
     * @param CronEvent $event
     * @return AbstractEvent
     */
    private function handleCronEvent(Job $job, CronEvent $event) : AbstractEvent
    {
        if (!$event->isPropagationStopped()) {
            $nextRun = $this->cron->getNextRun($event->getExpression());

            $job->setStartTime($this->now());
            $job->setTimestamp($nextRun);
            $this->gateway->addJob($job);
        }

        return $event->getEvent();
    }
}