Workana\AsyncJobs\Worker

Complexity

Total Complexity

12

Size/Duplication

Total Lines	176
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	14

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	19	1
A	bindSignals()	0	6	1
A	shutdown()	0	6	1
A	run()	0	12	3
A	shouldContinue()	0	5	1
A	invoke()	0	21	3
A	getProcessManager()	0	4	1
A	handleRejected()	0	13	1

<?php
namespace Workana\AsyncJobs;

use Throwable;
use Exception;
use Bernard\Envelope;
use Bernard\Queue;
use Bernard\Router;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use Workana\AsyncJobs\Event\SuccessfulExecutionEvent;
use Workana\AsyncJobs\Process\ProcessManager;
use Workana\AsyncJobs\Event\AfterExecutionEvent;
use Workana\AsyncJobs\Event\BeforeExecutionEvent;
use Workana\AsyncJobs\Event\RejectedExecutionEvent;
use Workana\AsyncJobs\Event\WorkerShutdownEvent;
use Workana\AsyncJobs\Util\Sleeper;
use Workana\AsyncJobs\Retry\RetryStrategy;

/**
 * Worker class
 *
 * @author Carlos Frutos <charly@workana.com>
 */
class Worker
{
    /**
     * @var ProcessManager
     */
    protected $pm;

    /**
     * @var Stopwatch
     */
    protected $stopwatch;

    /**
     * @var Sleeper
     */
    protected $sleeper;

    /**
     * @var Router
     */
    protected $router;

    /**
     * @var Queue
     */
    protected $queue;

    /**
     * @var RetryStrategy
     */
    protected $retryStrategy;

    /**
     * @var int
     */
    protected $shutdownSignal = null;

    /**
     * Create a new Worker
     *
     * @param Queue $queue
     * @param Router $router
     * @param EventDispatcherInterface $eventDispatcher
     * @param ProcessManager $processManager
     * @param RetryStrategy $retryStrategy
     */
    public function __construct(
        Queue $queue,
        Router $router,
        EventDispatcherInterface $eventDispatcher,
        ProcessManager $processManager,
        RetryStrategy $retryStrategy
    ) {
        $this->queue = $queue;
        $this->router = $router;
        $this->eventDispatcher = $eventDispatcher;

The property eventDispatcher does not exist. Did you maybe forget to declare it?

        $this->pm = $processManager;

        $this->stopwatch = new Stopwatch();
        $this->sleeper = new Sleeper();

        $this->retryStrategy = $retryStrategy;

        $this->bindSignals();
    }

    /**
     * Bind signal handlers
     *
     * @return void
     */
    protected function bindSignals()
    {
        $this->pm->signals()->registerHandler(SIGTERM, [$this, 'shutdown']);
        $this->pm->signals()->registerHandler(SIGINT, [$this, 'shutdown']);
        $this->pm->signals()->registerHandler(SIGQUIT, [$this, 'shutdown']);
    }

    /**
     * Shutdown worker
     *
     * @return void
     */
    public function shutdown($signal = SIGTERM)
    {
        $this->shutdownSignal = $signal;

        $this->eventDispatcher->dispatch(AsyncJobsEvents::WORKER_SHUTDOWN, new WorkerShutdownEvent());
    }

    /**
     * Run worker
     *
     * @return void
     */
    public function run()
    {
        while ($this->shouldContinue()) {
            $envelope = $this->queue->dequeue();

            if ($envelope) {
                $this->invoke($envelope);
            } else {
                $this->sleeper->sleep(1);
            }
        }
    }

    /**
     * @return bool
     */
    protected function shouldContinue()
    {
        $this->pm->dispatch();
        return empty($this->shutdownSignal);
    }

    /**
     * Invoke job execution
     *
     * @param Envelope $envelope
     *
     * @return void
     */
    public function invoke(Envelope $envelope)
    {
        $this->eventDispatcher->dispatch(AsyncJobsEvents::BEFORE_EXECUTION, new BeforeExecutionEvent());

        try {
            $this->stopwatch->start();

            call_user_func($this->router->map($envelope), $envelope->getMessage());

            $this->queue->acknowledge($envelope);

            $info = new ExecutionInfo($envelope->getMessage(), $this->queue, $this->stopwatch);

$envelope->getMessage() of type object<Bernard\Message> is not a sub-type of object<Workana\AsyncJobs\Job>. It seems like you assume a concrete implementation of the interface Bernard\Message to be always present.

            $this->eventDispatcher->dispatch(AsyncJobsEvents::SUCCESSFUL_EXECUTION, new SuccessfulExecutionEvent($envelope, $this, $info));
        } catch (Throwable $t) {

The class Throwable does not exist. Is this class maybe located in a folder that is not analyzed, or in a newer version of your dependencies than listed in your composer.lock/composer.json?

            $this->handleRejected($envelope, $t);
        } catch (Exception $e) {
            $this->handleRejected($envelope, $e);
        }

        $this->eventDispatcher->dispatch(AsyncJobsEvents::AFTER_EXECUTION, new AfterExecutionEvent());
    }

    /**
     * Get Process Managerç
     *
     * @return ProcessManager
     */
    public function getProcessManager()
    {
        return $this->pm;
    }

    /**
     * Handle rejected job
     *
     * @param Envelope $envelope Envelope
     * @param Throwable|Exception $error Error
     */
    protected function handleRejected(Envelope $envelope, $error)
    {
        $info = new ExecutionInfo($envelope->getMessage(), $this->queue, $this->stopwatch);

$envelope->getMessage() of type object<Bernard\Message> is not a sub-type of object<Workana\AsyncJobs\Job>. It seems like you assume a concrete implementation of the interface Bernard\Message to be always present.

        $this->eventDispatcher->dispatch(AsyncJobsEvents::REJECTED_EXECUTION, new RejectedExecutionEvent(
            $envelope,
            $error,

It seems like $error defined by parameter $error on line 186 can also be of type object<Exception>; however, Workana\AsyncJobs\Event\RejectedExecutionEvent::__construct() does only seem to accept object<Workana\AsyncJobs\Event\Exception>|object<Workana\AsyncJobs\Event\Throwable>, maybe add an additional type check?

            $this,
            $info
        ));

        $this->queue->acknowledge($envelope);
        $this->retryStrategy->handleRetry($envelope, $error);

It seems like $error defined by parameter $error on line 186 can also be of type object<Exception>; however, Workana\AsyncJobs\Retry\RetryStrategy::handleRetry() does only seem to accept object<Workana\AsyncJobs\Retry\Throwable>|object<Workana\AsyncJobs\Retry\Exception>, maybe add an additional type check?

    }
}