Ackintosh\Snidel\Worker

Complexity

Total Complexity

11

Size/Duplication

Total Lines	145
Duplicated Lines	0 %

Coupling/Cohesion

Components	2
Dependencies	12

Importance

Changes	11
Bugs	0	Features	0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	error()	0	20	2
A	terminate()	0	6	1
A	hasTask()	0	4	1
A	__construct()	0	11	1
A	setTaskQueue()	0	4	1
A	setResultQueue()	0	4	1
A	getPid()	0	4	1
A	run()	0	4	1
A	task()	0	17	1
A	isInProgress()	0	4	1

<?php
namespace Ackintosh\Snidel;

use Ackintosh\Snidel\Result\QueueInterface as ResultQueueInterface;
use Ackintosh\Snidel\Result\Result;
use Ackintosh\Snidel\Result\Formatter as ResultFormatter;
use Ackintosh\Snidel\Task\Formatter as TaskFormatter;
use Ackintosh\Snidel\Task\QueueInterface as TaskQueueInterface;
use Bernard\Consumer;
use Bernard\Message\PlainMessage;
use Bernard\Producer;
use Bernard\QueueFactory\PersistentFactory;
use Bernard\Router\SimpleRouter;
use Bernard\Serializer;
use Symfony\Component\EventDispatcher\EventDispatcher;

class Worker
{
    /** @var \Ackintosh\Snidel\Task\Task */
    private $latestTask;

    /** @var \Ackintosh\Snidel\Fork\Process */
    private $process;

    /** @var \Ackintosh\Snidel\Task\QueueInterface */
    private $taskQueue;

    /** @var \Ackintosh\Snidel\Result\QueueInterface */
    private $resultQueue;

    /** @var \Ackintosh\Snidel\Pcntl */
    private $pcntl;

    /** @var bool */
    private $isInProgress = false;

    private $factory;
    private $consumer;
    private $producer;

    /**
     * @param \Ackintosh\Snidel\Fork\Process $process
     * @param \Bernard\Driver $driver
     */
    public function __construct($process, $driver)
    {
        $this->pcntl = new Pcntl();
        $this->process = $process;

        $this->factory = new PersistentFactory($driver, new Serializer());
        $router = new SimpleRouter();
        $router->add('Task', $this);
        $this->consumer = new Consumer($router, new EventDispatcher());
        $this->producer = new Producer($this->factory, new EventDispatcher());
    }

    /**
     * @param   \Ackintosh\Snidel\Task\QueueInterface
     * @return  void
     */
    public function setTaskQueue(TaskQueueInterface $queue)
    {
        $this->taskQueue = $queue;
    }

    /**
     * @param   \Ackintosh\Snidel\Result\QueueInterface
     * @return  void
     */
    public function setResultQueue(ResultQueueInterface $queue)
    {
        $this->resultQueue = $queue;
    }

    /**
     * @return  int
     */
    public function getPid()
    {
        return $this->process->getPid();
    }

    /**
     * @return  void
     * @throws  \RuntimeException
     */
    public function run()
    {
        $this->consumer->consume($this->factory->create('task'));
    }

    public function task($message)
    {
        $this->isInProgress = true;
        $this->latestTask = $task = TaskFormatter::unserialize($message['task']);

$task = \Ackintosh\Snidel\Task\Formatter::unserialize($message['task']) is of type object<Ackintosh\Snidel\Task\TaskInterface>, but the property $latestTask was declared to be of type object<Ackintosh\Snidel\Task\Task>. Are you sure that you always receive this specific sub-class here, or does it make sense to add an instanceof check?

        $result = $task->execute();
        $result->setProcess($this->process);

        $this->producer->produce(
            new PlainMessage(
                'Result',
                [
                    'result' => ResultFormatter::serialize($result),
                ]
            )
        );
        $this->isInProgress = false;
    }

    /**
     * @return  void
     * @throws  \RuntimeException
     */
    public function error()
    {
        $result = new Result();
        $result->setError(error_get_last());
        $result->setTask($this->latestTask);
        $result->setProcess($this->process);

        try {
            $this->producer->produce(
                new PlainMessage(
                    'Result',
                    [
                        'result' => ResultFormatter::serialize($result),
                    ]
                )
            );
        } catch (\RuntimeException $e) {
            throw $e;
        }
    }

    /**
     * @param   int     $sig
     * @return  void
     */
    public function terminate($sig)
    {
        posix_kill($this->process->getPid(), $sig);
        $status = null;
        $this->pcntl->waitpid($this->process->getPid(), $status);
    }

    /**
     * @return bool
     */
    public function hasTask()
    {
        return $this->latestTask !== null;
    }

    /**
     * @return bool
     */
    public function isInProgress()
    {
        return $this->isInProgress;
    }
}