Wandu\Q\Worker

Complexity

Total Complexity

15

Size/Duplication

Total Lines	94
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Test Coverage

Coverage

0%

Importance

Changes	1
Bugs	0	Features	0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	flush()	0	4	1
A	stop()	0	7	2
A	work()	0	8	1
D	listen()	0	36	10

<?php
namespace Wandu\Q;

use Psr\Container\ContainerInterface;
use Psr\Log\LoggerInterface;
use Wandu\Q\Exception\WorkerStopException;

class Worker
{
    /** @var \Wandu\Q\Queue */
    protected $queue;
    
    /** @var \Psr\Container\ContainerInterface */
    protected $container;
    
    /** @var \Psr\Log\LoggerInterface */
    protected $logger;
    
    /** @var bool */
    protected $running = false;
    
    public function __construct(Queue $queue, ContainerInterface $container, LoggerInterface $logger = null)
    {
        $this->queue = $queue;
        $this->container = $container;
        $this->logger = $logger;
    }

    /**
     * @return void 
     */
    public function flush()
    {
        $this->queue->flush();
    }
    
    /**
     * @return void 
     */
    public function stop()
    {
        if ($this->logger) {
            $this->logger->info("stop by stop method.");
        }
        $this->running = false;
    }

    /**
     * @param string $class
     * @param string $method
     * @param array $arguments
     */
    public function work($class, $method, $arguments = [])
    {
        $this->queue->send([
            'class' => $class,
            'method' => $method,
            'arguments' => $arguments,
        ]);
    }
    
    /**
     * @param int $tick
     */
    public function listen($tick = 200000)
    {
        $this->running = true;
        $signalEnabled = false;
        if (function_exists('pcntl_signal') && function_exists('pcntl_signal_dispatch')) {
            pcntl_signal(SIGINT, [$this, 'stop']);
            pcntl_signal(SIGTERM, [$this, 'stop']);
            pcntl_signal(SIGHUP, [$this, 'stop']);
            $signalEnabled = true;
        } elseif ($this->logger) {
            $this->logger->info('if use pcntl_*, work more safety.');
        }
        try {
            while ($this->running) {
                if ($job = $this->queue->receive()) {
                    $result = $job->read();
                    call_user_func_array([
                        $this->container->get($result['class']),
                        $result['method']
                    ], $result['arguments'] ?? []);
                    if ($this->logger) {
                        $this->logger->info(sprintf("execute %s@%s", $result['method'], $result['class']));
                    }
                    $job->delete();
                }
                if ($signalEnabled) {
                    pcntl_signal_dispatch();
                }
                usleep($tick);
            }
        } catch (WorkerStopException $e) {
            if ($this->logger) {
                $this->logger->info("stop by WorkerStopException.");
            }
        }
    }
}