Redis::handleJob() - Code Metrics - suricate-php/framework - Measure and Improve Code Quality continuously with Scrutinizer

Redis::handleJob() A
last analyzed 2019-08-10 15:27 UTC

↳ Parent: Redis

Complexity

Conditions	1
Paths	1

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
cc	1
eloc	3
c	1
b	0
f	1
nc	1
nop	1
dl	0
loc	6
ccs	0
cts	4
cp	0
crap	2
rs	10

<?php

declare(strict_types=1);

namespace Suricate\Worker;

use Suricate\Suricate;

/**
 * Redis worker class
 */
class Redis
{
    /** @var string $redisHost Redis host name */
    protected $redisHost;

    /** @var string $redisPort Redis port number */
    protected $redisPort;

    /** @var int $maxChildren Number of children to launch */
    protected $maxChildren = 1;

    /** @var string $redisFifoName Redis key to listen to */
    protected $redisFifoName;

    /** @var int $redisFifoTimeout Redis blpop timeout value */
    protected $redisFifoTimeout = 2;

    private $childSlots = [];
    private $includedFiles = [];

    /** @var \Suricate\Logger $logger */
    private $logger;
    private $logPrefix = 'FATHER';

    public function __construct()
    {
        if ($this->redisHost === '') {
            throw new \Exception("Redis host is not set");
        }

        if ($this->redisPort === '') {
            throw new \Exception("Redis port is not set");
        }

        if ($this->redisFifoName === '') {
            throw new \Exception("Redis fifo name is not set");
        }

        $this->logger = Suricate::Logger();
    }

    /**
     * Worker log helper
     *
     * @param string $message
     * @return void
     */
    protected function log($message)
    {
        $this->logger->info('[' . $this->logPrefix . '] ' . $message);
    }

    /**
     * dummy function to hande an incoming job
     * must be overriden in inherited class
     *
     * @param array|null $payload
     * @return void
     */
    public function handleJob(?array $payload)
    {
        $this->logger->fatal(
            'received ' . json_encode($payload) . 'but no handle job defined'
        );
        exit();

    }

    /**
     * Enqueue a job into fifo
     *
     * @param mixed $payload
     * @return void
     */
    public function enqueue($payload)
    {
        $redisSrv = new \Predis\Client([
filter:
    dependency_paths: ["lib/*"]
            'scheme' => 'tcp',
            'host' => $this->redisHost,
            'port' => $this->redisPort,
            'read_write_timeout' => $this->redisFifoTimeout
        ]);

        $redisSrv->rpush($this->redisFifoName, json_encode($payload));
    }
    /**
     * Worker main run function
     *
     * @return void
     */
    public function run()
    {
        set_time_limit(0);
        posix_setsid();
        set_error_handler([$this, "errorHandler"]);

        $this->log(
            sprintf(
                'Starting worker, redis: %s:%s, queue name %s',
                $this->redisHost,
                $this->redisPort,
                $this->redisFifoName
            )
        );
        $this->log('Launching childrens');
        $father = true;
        // Forking children
        for ($i = 0; $i < $this->maxChildren; $i++) {
            $pid = pcntl_fork();
            if ($pid == -1) {
                die('Impossible to fork');

            } elseif ($pid > 0) {
                $father = true;
                $this->childSlots[$i]['pid'] = $pid;
                $this->childSlots[$i]['start_time'] = time();
                $this->log("Launching child " . ($i + 1) . " with PID " . $pid);
            } elseif ($pid == 0) {
                $childNum = $i + 1;
                $this->logPrefix = 'CHILD_' . $childNum;
                $father = false;
                $this->log("Launched");
                break;
            }
        }
        if ($father) {
            $this->startFather();
        }
        if (!$father) {

            $this->listen();
        }
    }

    public function shutdown()
    {
        $this->signalHandler(SIGTERM);
    }

    /**
     * Father process main loop
     * declare sig handler and enter in an infinite loop
     *
     * @return void
     */
    private function startFather()
    {
        register_shutdown_function([$this, 'shutdown']);

        declare(ticks=1);
        pcntl_signal(SIGINT, [$this, 'signalHandler']);
        pcntl_signal(SIGTERM, [$this, 'signalHandler']);
        pcntl_signal(SIGHUP, [$this, 'signalHandler']);
        pcntl_signal(SIGUSR1, [$this, 'signalHandler']);
        pcntl_signal(SIGCHLD, [$this, 'signalHandler']);
        while (true) {
            sleep(2);
            if (!$this->checkIncludedFiles()) {
                exit();

            }
        }
    }

    /**
     * Children listening function
     * connect to redis and wait for a job
     *
     * @return void
     */
    private function listen()
    {
        $this->log('Connecting to redis');

        $redisSrv = new \Predis\Client([
            'scheme' => 'tcp',
            'host' => $this->redisHost,
            'port' => $this->redisPort,
            'read_write_timeout' => 0
        ]);
        $this->setIncludedFiles();

        while (true) {
            $result = $redisSrv->blpop(
                $this->redisFifoName,
                $this->redisFifoTimeout
            );
            if ($result) {
                $this->log('Received job ' . json_encode($result[1]));
                $this->handleJob($result[1]);
            }

            if (!$this->checkIncludedFiles()) {
                exit();

            }
        }
    }

    /**
     * Store modification time of all included files of the worker
     * to detect modification and do restart
     *
     * @return void
     */
    private function setIncludedFiles()
    {
        $this->includedFiles = [];
        $allIncludedFiles = get_included_files();

        foreach ($allIncludedFiles as $filename) {
            $includeVersion = filemtime($filename);
            $this->includedFiles[$filename] = $includeVersion;
        }
    }

    /**
     * Compare included files modification times with original ones
     *
     * @return bool
     */
    private function checkIncludedFiles()
    {
        clearstatcache();
        $allIncludedFiles = get_included_files();
        foreach ($allIncludedFiles as $filename) {
            $includeVersion = filemtime($filename);
            if (isset($this->includedFiles[$filename])) {
                $version = $this->includedFiles[$filename];

                if ($includeVersion > $version) {
                    return false;
                }
            } else {
                $this->includedFiles[$filename] = $includeVersion;
            }
        }
        return true;
    }
    /**
     * signal handler, kill children when father receive signal
     *
     * @param int $signalNumber
     * @return void
     */
    private function signalHandler($signalNumber)
    {
        $this->log('received signal ' . $signalNumber);
        foreach ($this->childSlots as $index => $child) {
            $this->log(
                "Child[" . $index . "] Sending SIGTERM TO " . $child['pid']
            );
            posix_kill($child['pid'], SIGTERM);
            pcntl_wait($status);
            $this->log(
                "Child[" . $index . "] returned with status[" . $status . "]"
            );
        }

        exit();

    }

    private function errorHandler(
        $errNumber,
        $errMessage,
        $filename,
        $lineNumber,
        $vars
    ) {
        $this->log(
            'Error occured: ' .
                $errMessage .
                '/' .
                $errNumber .
                ', filename: ' .
                $filename .
                ' on line: ' .
                $lineNumber .
                '. vars : ' .
                json_encode($vars)
        );
    }
}


1			<?php
2
3			declare(strict_types=1);
4
5			namespace Suricate\Worker;
6
7			use Suricate\Suricate;
8
9			/**
10			* Redis worker class
11			*/
12			class Redis
13			{
14			/** @var string $redisHost Redis host name */
15			protected $redisHost;
16
17			/** @var string $redisPort Redis port number */
18			protected $redisPort;
19
20			/** @var int $maxChildren Number of children to launch */
21			protected $maxChildren = 1;
22
23			/** @var string $redisFifoName Redis key to listen to */
24			protected $redisFifoName;
25
26			/** @var int $redisFifoTimeout Redis blpop timeout value */
27			protected $redisFifoTimeout = 2;
28
29			private $childSlots = [];
30			private $includedFiles = [];
31
32			/** @var \Suricate\Logger $logger */
33			private $logger;
34			private $logPrefix = 'FATHER';
35
36			public function __construct()
37			{
38			if ($this->redisHost === '') {
39			throw new \Exception("Redis host is not set");
40			}
41
42			if ($this->redisPort === '') {
43			throw new \Exception("Redis port is not set");
44			}
45
46			if ($this->redisFifoName === '') {
47			throw new \Exception("Redis fifo name is not set");
48			}
49
50			$this->logger = Suricate::Logger();
51			}
52
53			/**
54			* Worker log helper
55			*
56			* @param string $message
57			* @return void
58			*/
59			protected function log($message)
60			{
61			$this->logger->info('[' . $this->logPrefix . '] ' . $message);
62			}
63
64			/**
65			* dummy function to hande an incoming job
66			* must be overriden in inherited class
67			*
68			* @param array\|null $payload
69			* @return void
70			*/
71			public function handleJob(?array $payload)
72			{
73			$this->logger->fatal(
74			'received ' . json_encode($payload) . 'but no handle job defined'
75			);
76			exit();
			0 ignored issues – show Best Practice introduced 2019-06-30 14:49 UTC by Report Bug Copy Issue Report Using `exit` here is not recommended. In general, usage of exit should be done with care and only when running in a scripting context like a CLI script. Loading history...
77			}
78
79			/**
80			* Enqueue a job into fifo
81			*
82			* @param mixed $payload
83			* @return void
84			*/
85			public function enqueue($payload)
86			{
87			$redisSrv = new \Predis\Client([
			0 ignored issues – show Bug introduced 2019-06-30 15:53 UTC by Report Bug Copy Issue Report The type `Predis\Client` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
88			'scheme' => 'tcp',
89			'host' => $this->redisHost,
90			'port' => $this->redisPort,
91			'read_write_timeout' => $this->redisFifoTimeout
92			]);
93
94			$redisSrv->rpush($this->redisFifoName, json_encode($payload));
95			}
96			/**
97			* Worker main run function
98			*
99			* @return void
100			*/
101			public function run()
102			{
103			set_time_limit(0);
104			posix_setsid();
105			set_error_handler([$this, "errorHandler"]);
106
107			$this->log(
108			sprintf(
109			'Starting worker, redis: %s:%s, queue name %s',
110			$this->redisHost,
111			$this->redisPort,
112			$this->redisFifoName
113			)
114			);
115			$this->log('Launching childrens');
116			$father = true;
117			// Forking children
118			for ($i = 0; $i < $this->maxChildren; $i++) {
119			$pid = pcntl_fork();
120			if ($pid == -1) {
121			die('Impossible to fork');
			0 ignored issues – show Best Practice introduced 2019-06-30 14:49 UTC by Report Bug Copy Issue Report Using `exit` here is not recommended. In general, usage of exit should be done with care and only when running in a scripting context like a CLI script. Loading history...
122			} elseif ($pid > 0) {
123			$father = true;
124			$this->childSlots[$i]['pid'] = $pid;
125			$this->childSlots[$i]['start_time'] = time();
126			$this->log("Launching child " . ($i + 1) . " with PID " . $pid);
127			} elseif ($pid == 0) {
128			$childNum = $i + 1;
129			$this->logPrefix = 'CHILD_' . $childNum;
130			$father = false;
131			$this->log("Launched");
132			break;
133			}
134			}
135			if ($father) {
136			$this->startFather();
137			}
138			if (!$father) {
			0 ignored issues – show introduced 2019-06-30 15:29 UTC by Report Bug Copy Issue Report The condition `$father` is always `false`. Loading history...
139			$this->listen();
140			}
141			}
142
143			public function shutdown()
144			{
145			$this->signalHandler(SIGTERM);
146			}
147
148			/**
149			* Father process main loop
150			* declare sig handler and enter in an infinite loop
151			*
152			* @return void
153			*/
154			private function startFather()
155			{
156			register_shutdown_function([$this, 'shutdown']);
157
158			declare(ticks=1);
159			pcntl_signal(SIGINT, [$this, 'signalHandler']);
160			pcntl_signal(SIGTERM, [$this, 'signalHandler']);
161			pcntl_signal(SIGHUP, [$this, 'signalHandler']);
162			pcntl_signal(SIGUSR1, [$this, 'signalHandler']);
163			pcntl_signal(SIGCHLD, [$this, 'signalHandler']);
164			while (true) {
165			sleep(2);
166			if (!$this->checkIncludedFiles()) {
167			exit();
			0 ignored issues – show Best Practice introduced 2019-06-30 14:49 UTC by Report Bug Copy Issue Report Using `exit` here is not recommended. In general, usage of exit should be done with care and only when running in a scripting context like a CLI script. Loading history...
168			}
169			}
170			}
171
172			/**
173			* Children listening function
174			* connect to redis and wait for a job
175			*
176			* @return void
177			*/
178			private function listen()
179			{
180			$this->log('Connecting to redis');
181
182			$redisSrv = new \Predis\Client([
183			'scheme' => 'tcp',
184			'host' => $this->redisHost,
185			'port' => $this->redisPort,
186			'read_write_timeout' => 0
187			]);
188			$this->setIncludedFiles();
189
190			while (true) {
191			$result = $redisSrv->blpop(
192			$this->redisFifoName,
193			$this->redisFifoTimeout
194			);
195			if ($result) {
196			$this->log('Received job ' . json_encode($result[1]));
197			$this->handleJob($result[1]);
198			}
199
200			if (!$this->checkIncludedFiles()) {
201			exit();
			0 ignored issues – show Best Practice introduced 2019-06-30 14:49 UTC by Report Bug Copy Issue Report Using `exit` here is not recommended. In general, usage of exit should be done with care and only when running in a scripting context like a CLI script. Loading history...
202			}
203			}
204			}
205
206			/**
207			* Store modification time of all included files of the worker
208			* to detect modification and do restart
209			*
210			* @return void
211			*/
212			private function setIncludedFiles()
213			{
214			$this->includedFiles = [];
215			$allIncludedFiles = get_included_files();
216
217			foreach ($allIncludedFiles as $filename) {
218			$includeVersion = filemtime($filename);
219			$this->includedFiles[$filename] = $includeVersion;
220			}
221			}
222
223			/**
224			* Compare included files modification times with original ones
225			*
226			* @return bool
227			*/
228			private function checkIncludedFiles()
229			{
230			clearstatcache();
231			$allIncludedFiles = get_included_files();
232			foreach ($allIncludedFiles as $filename) {
233			$includeVersion = filemtime($filename);
234			if (isset($this->includedFiles[$filename])) {
235			$version = $this->includedFiles[$filename];
236
237			if ($includeVersion > $version) {
238			return false;
239			}
240			} else {
241			$this->includedFiles[$filename] = $includeVersion;
242			}
243			}
244			return true;
245			}
246			/**
247			* signal handler, kill children when father receive signal
248			*
249			* @param int $signalNumber
250			* @return void
251			*/
252			private function signalHandler($signalNumber)
253			{
254			$this->log('received signal ' . $signalNumber);
255			foreach ($this->childSlots as $index => $child) {
256			$this->log(
257			"Child[" . $index . "] Sending SIGTERM TO " . $child['pid']
258			);
259			posix_kill($child['pid'], SIGTERM);
260			pcntl_wait($status);
261			$this->log(
262			"Child[" . $index . "] returned with status[" . $status . "]"
263			);
264			}
265
266			exit();
			0 ignored issues – show Best Practice introduced 2019-06-30 14:49 UTC by Report Bug Copy Issue Report Using `exit` here is not recommended. In general, usage of exit should be done with care and only when running in a scripting context like a CLI script. Loading history...
267			}
268
269			private function errorHandler(
270			$errNumber,
271			$errMessage,
272			$filename,
273			$lineNumber,
274			$vars
275			) {
276			$this->log(
277			'Error occured: ' .
278			$errMessage .
279			'/' .
280			$errNumber .
281			', filename: ' .
282			$filename .
283			' on line: ' .
284			$lineNumber .
285			'. vars : ' .
286			json_encode($vars)
287			);
288			}
289			}
290

suricate-php / framework

Redis::handleJob() A last analyzed 2019-08-10 15:27 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Redis::handleJob() A
last analyzed 2019-08-10 15:27 UTC