Issues in Manager.php (develop) - Issues in develop - cronario/cronario - Measure and Improve Code Quality continuously with Scrutinizer

Issues (18)

Security Analysis no request data

Cross-Site Scripting

Cross-Site Scripting enables an attacker to inject code into the response of a web-request that is viewed by other users. It can for example be used to bypass access controls, or even to take over other users' accounts.

File Exposure

File Exposure allows an attacker to gain access to local files that he should not be able to access. These files can for example include database credentials, or other configuration files.

File Manipulation

File Manipulation enables an attacker to write custom data to files. This potentially leads to injection of arbitrary code on the server.

Object Injection

Object Injection enables an attacker to inject an object into PHP code, and can lead to arbitrary code execution, file exposure, or file manipulation attacks.

Code Injection

Code Injection enables an attacker to execute arbitrary code on the server.

Response Splitting

Response Splitting can be used to send arbitrary responses.

File Inclusion

File Inclusion enables an attacker to inject custom files into PHP's file loading mechanism, either explicitly passed to include, or for example via PHP's auto-loading mechanism.

Command Injection

Command Injection enables an attacker to inject a shell command that is execute with the privileges of the web-server. This can be used to expose sensitive data, or gain access of your server.

SQL Injection

SQL Injection enables an attacker to execute arbitrary SQL code on your database server gaining access to user data, or manipulating user data.

XPath Injection

XPath Injection enables an attacker to modify the parts of XML document that are read. If that XML document is for example used for authentication, this can lead to further vulnerabilities similar to SQL Injection.

LDAP Injection

LDAP Injection enables an attacker to inject LDAP statements potentially granting permission to run unauthorized queries, or modify content inside the LDAP tree.

Header Injection

Other Vulnerability

This category comprises other attack vectors such as manipulating the PHP runtime, loading custom extensions, freezing the runtime, or similar.

Regex Injection

Regex Injection enables an attacker to execute arbitrary code in your PHP process.

XML Injection

XML Injection enables an attacker to read files on your local filesystem including configuration files, or can be abused to freeze your web-server process.

Variable Injection

Variable Injection enables an attacker to overwrite program variables with custom data, and can lead to further vulnerabilities.

Unfortunately, the security analysis is currently not available for your project. If you are a non-commercial open-source project, please contact support to gain access.

src/Manager.php (5 issues)

Labels

Duplication 5

Severity

Informational 5

Upgrade to new PHP Analysis Engine

These results are based on our legacy PHP analysis, consider migrating to our new PHP analysis engine instead. Learn more

<?php

namespace Cronario;

use Cronario\Exception\JobException;
use Cronario\Exception\ResultException;


// IMPORTANT !!!
// use here 'static' not 'self' in child class of \Thread
// cause it is BUG - will get 'Segmentation Error' sometimes


class Manager extends \Thread
{

    /**
     * cant use cause
     * pthreads detected an attempt to call private method Cronario\Manager::setOptions from outside the threading context
     *
     * use TraitOptions;
     */


    const P_WORKER_CLASS = 'workerClass';
    const P_ID = 'id';
    const P_APP_ID = 'appId';
    const P_BOOTSTRAP_FILE = 'bootstrapFile';

    const REDIS_NS_LIVE = 'cronario@manager-live';
    const REDIS_NS_STATS = 'cronario@manager-stats';

    protected $appId;
    protected $id;
    protected $bootstrapFile;
    protected $workerClass;
    protected $startOn;

    protected $eventTriggerSet = []; // need for speed enter for this values


    //region INIT *******************************************************

    /**
     * @param $id
     * @param $appId
     * @param $workerClass
     * @param $bootstrapFile
     */
    public function __construct($id, $appId, $workerClass, $bootstrapFile)
    {
        $this->setId($id);
        $this->setAppId($appId);
        $this->setWorkerClass($workerClass);
        $this->setBootstrapFile($bootstrapFile);

        $this->startOn = time();

        return $this->start(PTHREADS_INHERIT_NONE);
    }

    // endregion *******************************************************


    /**
     * @return mixed
     */
    protected function getBootstrapFile()
    {
        return $this->bootstrapFile;
    }

    /**
     * @param $bootstrapFile
     *
     * @return $this
     */
    protected function setBootstrapFile($bootstrapFile)
    {
        $this->bootstrapFile = $bootstrapFile;

        return $this;
    }

    /**
     * @return mixed
     */
    protected function getAppId()
    {
        return $this->appId;
    }

    /**
     * @param $appId
     *
     * @return $this
     */
    protected function setAppId($appId)
    {
        $this->appId = $appId;

        return $this;
    }

    /**
     * @return mixed
     */
    protected function getId()
    {
        return $this->id;
    }


    /**
     * @param $id
     *
     * @return $this
     */
    protected function setId($id)
    {
        $this->id = $id;

        return $this;
    }

    /**
     * @return mixed
     */
    protected function getWorkerClass()
    {
        return $this->workerClass;
    }

    /**
     * @param $workerClass
     *
     * @return $this
     */
    protected function setWorkerClass($workerClass)
    {
        $this->workerClass = $workerClass;

        return $this;
    }

    /**
     * @return mixed
     */
    protected function getStartOn()
    {
        return $this->startOn;
    }


    //endregion *******************************************************


    // region PRODUCER GOODS *************************************************

    /**
     * @var Producer
     */
    protected static $producer;

    /**
     * @return Producer
     * @throws Exception\FacadeException
     */
    protected function getProducer()
    {
        if (null === static::$producer) {
            static::$producer = Facade::getProducer($this->getAppId());
        }

        return static::$producer;
    }

    /**
     * @var \Predis\Client
     */
    protected static $redis;

    /**
     * @return \Predis\Client
     * @throws Exception\FacadeException
     */
    protected function getRedis()
    {
        if (null === static::$redis) {
            static::$redis = static::getProducer()->getRedis();
        }

        return static::$redis;
    }

    /**
     * @var \Psr\Log\LoggerInterface
     */
    protected static $logger;

    /**
     * @return \Psr\Log\LoggerInterface
     */
    protected function getLogger()
    {
        if (null === static::$logger) {
            static::$logger = static::getProducer()->getLogger();
        }

        return static::$logger;
    }

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

    /**
     * @return Queue
     */
    protected function getQueue()
    {
        if (null === static::$queue) {
            static::$queue = static::getProducer()->getQueue();
        }

        return static::$queue;
    }

    // endregion *******************************************************


    //region Stats ***********************************************************

    const EVENT_SUCCESS = 'success';
    const EVENT_FAIL = 'fail';
    const EVENT_ERROR = 'error';
    const EVENT_RETRY = 'retry';
    const EVENT_REDIRECT = 'redirect';

    /**
     * @param $event
     */
    protected function eventTrigger($event)
    {
        $this->eventTriggerSet[$event]++;

        $key1 = $this->buildManagerStatKey();
        $this->getRedis()->hincrby($key1, $event, 1);

        $key2 = $this->buildManagerLiveKey($this->getId());
        $this->getRedis()->hincrby($key2, $event, 1);
    }


    const P_STATS_KEY_NAMESPACE = 'namespace';
    const P_STATS_KEY_APP_ID = 'appId';
    const P_STATS_KEY_WORKER_CLASS = 'workerClass';

    /**
     * @param $key
     *
     * @return array
     */
    public static function parseManagerStatKey($key)
    {
        list($namespace, $appId, $workerClass) = explode(':', $key);

        return [
            static::P_STATS_KEY_NAMESPACE    => $namespace,
            static::P_STATS_KEY_APP_ID       => $appId,
            static::P_STATS_KEY_WORKER_CLASS => $workerClass,
        ];
    }

    /**
     * @return string
     */
    protected function buildManagerStatKey()
    {
        return implode(':', [static::REDIS_NS_STATS, $this->getAppId(), $this->getWorkerClass(),]);
    }

    /**
     * @param $id
     *
     * @return string
     */
    protected function buildManagerLiveKey($id)
    {
        return implode(':', [static::REDIS_NS_LIVE, $this->getAppId(), $id]);
    }


    const P_LIVE_KEY_NAMESPACE = 'namespace';
    const P_LIVE_KEY_APP_ID = 'appId';
    const P_LIVE_KEY_WORKER_CLASS = 'workerClass';
    const P_LIVE_KEY_MANAGER_ID = 'managerId';
    const P_LIVE_KEY_STARTED_TIME = 'started';

    /**
     * @return $this
     */
    protected function startManagerLive()
    {
        $key = $this->buildManagerLiveKey($this->getId());
        $this->getRedis()->hmset($key, [
            static::P_LIVE_KEY_APP_ID       => $this->getAppId(),
            static::P_LIVE_KEY_WORKER_CLASS => $this->getWorkerClass(),
            static::P_LIVE_KEY_MANAGER_ID   => $this->getId(),
            static::P_LIVE_KEY_STARTED_TIME => $this->getStartOn(),
        ]);

        return $this;
    }

    /**
     * @return $this
     */
    protected function finishManagerLive()
    {
        $key = $this->buildManagerLiveKey($this->getId());
        $this->getRedis()->del($key);

        return $this;
    }

    //endregion ***********************************************************


    //region MainLoop ********************************************************

    /**
     * @return bool
     */
    public function run()
    {

        $file = $this->getBootstrapFile();

        require_once($file);

        $this->startManagerLive();

        $workerClass = $this->getWorkerClass();
        $managerId = $this->getId();
        $logger = $this->getLogger();
        $queue = $this->getQueue();
        $logger->info("Manager {$managerId} start work {$workerClass}", [__NAMESPACE__]);


        try {
            $worker = AbstractWorker::factory($workerClass);
        } catch (Exception\WorkerException $ex) {

            $queue->stop($workerClass);
            $logger->warning($ex->getMessage(), [__NAMESPACE__]);
            $logger->info("Manager {$managerId} Queue stop {$workerClass}", [__NAMESPACE__]);
            $logger->info("Manager {$managerId} finish work {$workerClass}", [__NAMESPACE__]);

            return false;
        }

        try {

            $waitForJob
                = $this->getWorkerConfig(AbstractWorker::CONFIG_P_MANAGER_IDLE_DIE_DELAY);

            while ($jobId = $queue->reserveJob($workerClass, $waitForJob)) {

                if ($jobId === false) {
                    $logger->debug("Manager {$managerId} queue is empty so {$workerClass}", [__NAMESPACE__]);
                    break;
                }

                $logger->debug("Manager {$managerId} reserve job {$jobId}", [__NAMESPACE__]);

                $job = null;
                try {
                    /** @var AbstractJob $job */
                    $job = Facade::getStorage($this->getAppId())->find($jobId);
                } catch (JobException $ex) {
                    $logger->warning($ex->getMessage(), [__NAMESPACE__]);
                    $queue->deleteJob($jobId);
                    continue;
                }

                if ($job === null) {
                    $logger->error("Manager {$managerId} job is not instanceof of AbstractJob {$jobId}",
                        [__NAMESPACE__]);
                    $queue->deleteJob($jobId);
                    continue;
                }

                $logger->debug("Manager {$managerId} Worker start working at {$jobId}...", [__NAMESPACE__]);

                // DO JOB!

                /** @var ResultException $result */
                $result = $worker($job);

                $logger->debug("Manager {$managerId}  Worker finish working at {$jobId} : {$result->getGlobalCode()}",
                    [__NAMESPACE__]);

                if ($result instanceof ResultException) {

                    if ($result->isSuccess()) {

                        $queue->deleteJob($jobId);
                        $logger->debug("Manager {$managerId} Job ResultException isSuccess {$jobId}", [__NAMESPACE__]);
                        $this->eventTrigger(self::EVENT_SUCCESS);

                    } elseif ($result->isFailure()) {

                        $queue->deleteJob($jobId);
                        $logger->debug("Manager {$managerId} Job ResultException isFail {$jobId}", [__NAMESPACE__]);
                        $this->eventTrigger(self::EVENT_FAIL);

                    } elseif ($result->isError()) {

                        $queue->buryJob($jobId);
                        $logger->debug("Manager {$managerId} Job ResultException isError {$jobId}", [__NAMESPACE__]);
                        $this->eventTrigger(self::EVENT_ERROR);

                    } elseif ($result->isRetry()) {

                        $logger->debug("Manager {$managerId} Job ResultException isRetry {$jobId}", [__NAMESPACE__]);
                        $job->addAttempts();
                        $job->save(); // important this will saved result to job !!!

                        $attemptCount = $job->getAttempts();
                        $attemptDelay = $job->countAttemptQueueDelay();

                        // can be new worker class "like redirect in sms/alpha to sms"
                        $gatewayClass = $job->getWorkerClass();
                        $queue->deleteJob($jobId);

                        if ($job->hasAttempt()) {
                            $logger->debug("job {$jobId} has {$attemptCount} attempts (max:{$job->getAttemptsMax()}) and will be delayed {$attemptDelay}",
                                [__NAMESPACE__]);

                            $queue->putJob($gatewayClass, $jobId, $attemptDelay);
                            $this->eventTrigger(self::EVENT_RETRY);
                        } else {
                            $logger->debug("job {$jobId} has {$attemptCount} attempts (max:{$job->getAttemptsMax()})  and will have bad result",
                                [__NAMESPACE__]);

                            $job->setResult(new ResultException(ResultException::FAILURE_MAX_ATTEMPTS));
                            $job->save();
                            $this->eventTrigger(self::EVENT_FAIL);
                        }

                    } elseif ($result->isRedirect()) {

                        $queue->deleteJob($jobId);
                        $queue->putJob($job->getWorkerClass(), $jobId);

                        $logger->debug("Job ResultException isRedirect {$jobId} to {$job->getWorkerClass()}",
                            [__NAMESPACE__]);

                        $this->eventTrigger(self::EVENT_REDIRECT);

                    } else {
                        $queue->deleteJob($jobId);
                        $logger->error("Undefined result job id : {$jobId}", [__NAMESPACE__]);
                    }
                } else {
                    $logger->error('job result is not type of AbstractResultException', [__NAMESPACE__]);
                }


                if ($job->getSchedule()) {
                    $newJob = clone $job;
                    $newJob();

                    $logger->debug("Manager {$managerId} catch daemon  shutdown, finish listening queue",
                        [__NAMESPACE__]);
                }

                if ($this->isLimitsExceeded() || $this->isProducerShutDown()) {
                    break;
                }

                $this->waitDelay();

            }
        } catch (\Exception $ex) {
            $logger->warning($ex->getMessage());
        }

        $logger->info("Manager {$managerId} finish work {$workerClass}", [__NAMESPACE__]);
        $this->finishManagerLive();

        return true;
    }

    //endregion ********************************************************

    //region Worker ***********************************************************

    protected static $workerConfig;

    /**
     * @param null $key
     *
     * @return mixed
     */
    protected function getWorkerConfig($key = null)
    {
        /** @var AbstractWorker $class */
        if (null === static::$workerConfig) {
            $class = $this->workerClass;
            static::$workerConfig = $class::getConfig();
        }

        return (null === $key)
            ? static::$workerConfig
            : static::$workerConfig[$key];
    }

    //endregion ***********************************************************

    /**
     *
     */
    protected function waitDelay()
    {
        $sleep = $this->getWorkerConfig(AbstractWorker::CONFIG_P_MANAGER_JOB_DONE_DELAY);
        if ($sleep > 0) {
            $logger = $this->getLogger();
            $managerId = $this->getId();
            $logger->debug("Manager {$managerId} job-done-after-sleep : {$sleep} ...", [__NAMESPACE__]);
            sleep($sleep);
        }
    }

    /**
     * @return bool
     */
    protected function isProducerShutDown()
    {
        if (!$this->getProducer()->isStateStart()) {
            $logger = $this->getLogger();
            $managerId = $this->getId();

            $logger->info("Manager {$managerId} catch daemon  shutdown, finish listening queue", [__NAMESPACE__]);

            return true;
        }

        return false;
    }


    /**
     * @return bool
     */
    protected function isLimitsExceeded()
    {

        $events = $this->eventTriggerSet;
        $wc = $this->getWorkerConfig();
        $managerId = $this->getId();
        $logger = $this->getLogger();

        $jobsDoneCount
            = $events[self::EVENT_SUCCESS] + $events[self::EVENT_FAIL] + $events[self::EVENT_ERROR];

        $lim = $wc[AbstractWorker::CONFIG_P_JOBS_DONE_LIMIT];
        if ($lim > 0 && $lim === $jobsDoneCount) {
            $logger->debug("Manager {$managerId} done limit is equal {$jobsDoneCount}, so finish this manager ...",
                [__NAMESPACE__]);

            return true;
        }

        $lim = $wc[AbstractWorker::CONFIG_P_JOBS_SUCCESS_LIMIT];
        if ($lim > 0 && $lim === $events[self::EVENT_SUCCESS]) {

            $logger->debug("Manager {$managerId} success limit is equal {$events['isSuccess']}, so finish this manager ...",
                [__NAMESPACE__]);

            return true;
        }

        $lim = $wc[AbstractWorker::CONFIG_P_JOBS_FAIL_LIMIT];
        if ($lim > 0 && $lim === $events[self::EVENT_FAIL]) {

            $logger->debug("Manager {$managerId} fail limit is equal {$events['isFail']}, so finish this manager ...",
                [__NAMESPACE__]);

            return true;
        }

        $lim = $wc[AbstractWorker::CONFIG_P_JOBS_RETRY_LIMIT];
        if ($lim > 0 && $lim === $events[self::EVENT_RETRY]) {

            $logger->debug("Manager {$managerId} retry limit is equal {$events['isRetry']}, so finish this manager ...",
                [__NAMESPACE__]);

            return true;
        }

        $lim = $wc[AbstractWorker::CONFIG_P_JOBS_ERROR_LIMIT];
        if ($lim > 0 && $lim === $events[self::EVENT_ERROR]) {

            $logger->debug("Manager {$managerId} error limit is equal {$events['isError']}, so finish this manager ...",
                [__NAMESPACE__]);

            return true;
        }

        $lim = $wc[AbstractWorker::CONFIG_P_JOBS_REDIRECT_LIMIT];
        if ($lim > 0 && $lim === $events[self::EVENT_REDIRECT]) {

            $logger->debug("Manager {$managerId} redirect limit is equal {$events['isRedirect']}, so finish this manager ...",
                [__NAMESPACE__]);

            return true;
        }

        $lim = $wc[AbstractWorker::CONFIG_P_MANAGER_LIFETIME];
        if ($lim > 0 && $this->getStartOn() + $lim <= time()) {
            $logger->debug("Manager {$managerId} lifetime limit {$lim}, so finish this manager ...", [__NAMESPACE__]);

            return true;
        }

        return false;
    }


}

1			<?php
2
3			namespace Cronario;
4
5			use Cronario\Exception\JobException;
6			use Cronario\Exception\ResultException;
7
8
9			// IMPORTANT !!!
10			// use here 'static' not 'self' in child class of \Thread
11			// cause it is BUG - will get 'Segmentation Error' sometimes
12
13
14			class Manager extends \Thread
15			{
16
17			/**
18			* cant use cause
19			* pthreads detected an attempt to call private method Cronario\Manager::setOptions from outside the threading context
20			*
21			* use TraitOptions;
22			*/
23
24
25			const P_WORKER_CLASS = 'workerClass';
26			const P_ID = 'id';
27			const P_APP_ID = 'appId';
28			const P_BOOTSTRAP_FILE = 'bootstrapFile';
29
30			const REDIS_NS_LIVE = 'cronario@manager-live';
31			const REDIS_NS_STATS = 'cronario@manager-stats';
32
33			protected $appId;
34			protected $id;
35			protected $bootstrapFile;
36			protected $workerClass;
37			protected $startOn;
38
39			protected $eventTriggerSet = []; // need for speed enter for this values
40
41
42			//region INIT *******************************************************
43
44			/**
45			* @param $id
46			* @param $appId
47			* @param $workerClass
48			* @param $bootstrapFile
49			*/
50	1		public function __construct($id, $appId, $workerClass, $bootstrapFile)
51			{
52	1		$this->setId($id);
53	1		$this->setAppId($appId);
54	1		$this->setWorkerClass($workerClass);
55	1		$this->setBootstrapFile($bootstrapFile);
56
57	1		$this->startOn = time();
58
59	1		return $this->start(PTHREADS_INHERIT_NONE);
60			}
61
62			// endregion *******************************************************
63
64
65			/**
66			* @return mixed
67			*/
68			protected function getBootstrapFile()
69			{
70			return $this->bootstrapFile;
71			}
72
73			/**
74			* @param $bootstrapFile
75			*
76			* @return $this
77			*/
78	1		protected function setBootstrapFile($bootstrapFile)
79			{
80	1		$this->bootstrapFile = $bootstrapFile;
81
82	1		return $this;
83			}
84
85			/**
86			* @return mixed
87			*/
88			protected function getAppId()
89			{
90			return $this->appId;
91			}
92
93			/**
94			* @param $appId
95			*
96			* @return $this
97			*/
98	1		protected function setAppId($appId)
99			{
100	1		$this->appId = $appId;
101
102	1		return $this;
103			}
104
105			/**
106			* @return mixed
107			*/
108			protected function getId()
109			{
110			return $this->id;
111			}
112
113
114			/**
115			* @param $id
116			*
117			* @return $this
118			*/
119	1		protected function setId($id)
120			{
121	1		$this->id = $id;
122
123	1		return $this;
124			}
125
126			/**
127			* @return mixed
128			*/
129			protected function getWorkerClass()
130			{
131			return $this->workerClass;
132			}
133
134			/**
135			* @param $workerClass
136			*
137			* @return $this
138			*/
139	1		protected function setWorkerClass($workerClass)
140			{
141	1		$this->workerClass = $workerClass;
142
143	1		return $this;
144			}
145
146			/**
147			* @return mixed
148			*/
149			protected function getStartOn()
150			{
151			return $this->startOn;
152			}
153
154
155			//endregion *******************************************************
156
157
158			// region PRODUCER GOODS *************************************************
159
160			/**
161			* @var Producer
162			*/
163			protected static $producer;
164
165			/**
166			* @return Producer
167			* @throws Exception\FacadeException
168			*/
169			protected function getProducer()
170			{
171			if (null === static::$producer) {
172			static::$producer = Facade::getProducer($this->getAppId());
173			}
174
175			return static::$producer;
176			}
177
178			/**
179			* @var \Predis\Client
180			*/
181			protected static $redis;
182
183			/**
184			* @return \Predis\Client
185			* @throws Exception\FacadeException
186			*/
187			protected function getRedis()
188			{
189			if (null === static::$redis) {
190			static::$redis = static::getProducer()->getRedis();
191			}
192
193			return static::$redis;
194			}
195
196			/**
197			* @var \Psr\Log\LoggerInterface
198			*/
199			protected static $logger;
200
201			/**
202			* @return \Psr\Log\LoggerInterface
203			*/
204			protected function getLogger()
205			{
206			if (null === static::$logger) {
207			static::$logger = static::getProducer()->getLogger();
208			}
209
210			return static::$logger;
211			}
212
213			/**
214			* @var Queue
215			*/
216			protected static $queue;
217
218			/**
219			* @return Queue
220			*/
221			protected function getQueue()
222			{
223			if (null === static::$queue) {
224			static::$queue = static::getProducer()->getQueue();
225			}
226
227			return static::$queue;
228			}
229
230			// endregion *******************************************************
231
232
233			//region Stats ***********************************************************
234
235			const EVENT_SUCCESS = 'success';
236			const EVENT_FAIL = 'fail';
237			const EVENT_ERROR = 'error';
238			const EVENT_RETRY = 'retry';
239			const EVENT_REDIRECT = 'redirect';
240
241			/**
242			* @param $event
243			*/
244			protected function eventTrigger($event)
245			{
246			$this->eventTriggerSet[$event]++;
247
248			$key1 = $this->buildManagerStatKey();
249			$this->getRedis()->hincrby($key1, $event, 1);
250
251			$key2 = $this->buildManagerLiveKey($this->getId());
252			$this->getRedis()->hincrby($key2, $event, 1);
253			}
254
255
256			const P_STATS_KEY_NAMESPACE = 'namespace';
257			const P_STATS_KEY_APP_ID = 'appId';
258			const P_STATS_KEY_WORKER_CLASS = 'workerClass';
259
260			/**
261			* @param $key
262			*
263			* @return array
264			*/
265			public static function parseManagerStatKey($key)
266			{
267			list($namespace, $appId, $workerClass) = explode(':', $key);
268
269			return [
270			static::P_STATS_KEY_NAMESPACE => $namespace,
271			static::P_STATS_KEY_APP_ID => $appId,
272			static::P_STATS_KEY_WORKER_CLASS => $workerClass,
273			];
274			}
275
276			/**
277			* @return string
278			*/
279			protected function buildManagerStatKey()
280			{
281			return implode(':', [static::REDIS_NS_STATS, $this->getAppId(), $this->getWorkerClass(),]);
282			}
283
284			/**
285			* @param $id
286			*
287			* @return string
288			*/
289			protected function buildManagerLiveKey($id)
290			{
291			return implode(':', [static::REDIS_NS_LIVE, $this->getAppId(), $id]);
292			}
293
294
295			const P_LIVE_KEY_NAMESPACE = 'namespace';
296			const P_LIVE_KEY_APP_ID = 'appId';
297			const P_LIVE_KEY_WORKER_CLASS = 'workerClass';
298			const P_LIVE_KEY_MANAGER_ID = 'managerId';
299			const P_LIVE_KEY_STARTED_TIME = 'started';
300
301			/**
302			* @return $this
303			*/
304			protected function startManagerLive()
305			{
306			$key = $this->buildManagerLiveKey($this->getId());
307			$this->getRedis()->hmset($key, [
308			static::P_LIVE_KEY_APP_ID => $this->getAppId(),
309			static::P_LIVE_KEY_WORKER_CLASS => $this->getWorkerClass(),
310			static::P_LIVE_KEY_MANAGER_ID => $this->getId(),
311			static::P_LIVE_KEY_STARTED_TIME => $this->getStartOn(),
312			]);
313
314			return $this;
315			}
316
317			/**
318			* @return $this
319			*/
320			protected function finishManagerLive()
321			{
322			$key = $this->buildManagerLiveKey($this->getId());
323			$this->getRedis()->del($key);
324
325			return $this;
326			}
327
328			//endregion ***********************************************************
329
330
331			//region MainLoop ********************************************************
332
333			/**
334			* @return bool
335			*/
336			public function run()
337			{
338
339			$file = $this->getBootstrapFile();
340
341			require_once($file);
342
343			$this->startManagerLive();
344
345			$workerClass = $this->getWorkerClass();
346			$managerId = $this->getId();
347			$logger = $this->getLogger();
348			$queue = $this->getQueue();
349			$logger->info("Manager {$managerId} start work {$workerClass}", [__NAMESPACE__]);
350
351
352			try {
353			$worker = AbstractWorker::factory($workerClass);
354			} catch (Exception\WorkerException $ex) {
355
356			$queue->stop($workerClass);
357			$logger->warning($ex->getMessage(), [__NAMESPACE__]);
358			$logger->info("Manager {$managerId} Queue stop {$workerClass}", [__NAMESPACE__]);
359			$logger->info("Manager {$managerId} finish work {$workerClass}", [__NAMESPACE__]);
360
361			return false;
362			}
363
364			try {
365
366			$waitForJob
367			= $this->getWorkerConfig(AbstractWorker::CONFIG_P_MANAGER_IDLE_DIE_DELAY);
368
369			while ($jobId = $queue->reserveJob($workerClass, $waitForJob)) {
370
371			if ($jobId === false) {
372			$logger->debug("Manager {$managerId} queue is empty so {$workerClass}", [__NAMESPACE__]);
373			break;
374			}
375
376			$logger->debug("Manager {$managerId} reserve job {$jobId}", [__NAMESPACE__]);
377
378			$job = null;
379			try {
380			/** @var AbstractJob $job */
381			$job = Facade::getStorage($this->getAppId())->find($jobId);
382			} catch (JobException $ex) {
383			$logger->warning($ex->getMessage(), [__NAMESPACE__]);
384			$queue->deleteJob($jobId);
385			continue;
386			}
387
388			if ($job === null) {
389			$logger->error("Manager {$managerId} job is not instanceof of AbstractJob {$jobId}",
390			[__NAMESPACE__]);
391			$queue->deleteJob($jobId);
392			continue;
393			}
394
395			$logger->debug("Manager {$managerId} Worker start working at {$jobId}...", [__NAMESPACE__]);
396
397			// DO JOB!
398
399			/** @var ResultException $result */
400			$result = $worker($job);
401
402			$logger->debug("Manager {$managerId} Worker finish working at {$jobId} : {$result->getGlobalCode()}",
403			[__NAMESPACE__]);
404
405			if ($result instanceof ResultException) {
406
407			if ($result->isSuccess()) {
408
409			$queue->deleteJob($jobId);
410			$logger->debug("Manager {$managerId} Job ResultException isSuccess {$jobId}", [__NAMESPACE__]);
411			$this->eventTrigger(self::EVENT_SUCCESS);
412
413			} elseif ($result->isFailure()) {
414
415			$queue->deleteJob($jobId);
416			$logger->debug("Manager {$managerId} Job ResultException isFail {$jobId}", [__NAMESPACE__]);
417			$this->eventTrigger(self::EVENT_FAIL);
418
419			} elseif ($result->isError()) {
420
421			$queue->buryJob($jobId);
422			$logger->debug("Manager {$managerId} Job ResultException isError {$jobId}", [__NAMESPACE__]);
423			$this->eventTrigger(self::EVENT_ERROR);
424
425			} elseif ($result->isRetry()) {
426
427			$logger->debug("Manager {$managerId} Job ResultException isRetry {$jobId}", [__NAMESPACE__]);
428			$job->addAttempts();
429			$job->save(); // important this will saved result to job !!!
430
431			$attemptCount = $job->getAttempts();
432			$attemptDelay = $job->countAttemptQueueDelay();
433
434			// can be new worker class "like redirect in sms/alpha to sms"
435			$gatewayClass = $job->getWorkerClass();
436			$queue->deleteJob($jobId);
437
438			if ($job->hasAttempt()) {
439			$logger->debug("job {$jobId} has {$attemptCount} attempts (max:{$job->getAttemptsMax()}) and will be delayed {$attemptDelay}",
440			[__NAMESPACE__]);
441
442			$queue->putJob($gatewayClass, $jobId, $attemptDelay);
443			$this->eventTrigger(self::EVENT_RETRY);
444			} else {
445			$logger->debug("job {$jobId} has {$attemptCount} attempts (max:{$job->getAttemptsMax()}) and will have bad result",
446			[__NAMESPACE__]);
447
448			$job->setResult(new ResultException(ResultException::FAILURE_MAX_ATTEMPTS));
449			$job->save();
450			$this->eventTrigger(self::EVENT_FAIL);
451			}
452
453			} elseif ($result->isRedirect()) {
454
455			$queue->deleteJob($jobId);
456			$queue->putJob($job->getWorkerClass(), $jobId);
457
458			$logger->debug("Job ResultException isRedirect {$jobId} to {$job->getWorkerClass()}",
459			[__NAMESPACE__]);
460
461			$this->eventTrigger(self::EVENT_REDIRECT);
462
463			} else {
464			$queue->deleteJob($jobId);
465			$logger->error("Undefined result job id : {$jobId}", [__NAMESPACE__]);
466			}
467			} else {
468			$logger->error('job result is not type of AbstractResultException', [__NAMESPACE__]);
469			}
470
471
472			if ($job->getSchedule()) {
473			$newJob = clone $job;
474			$newJob();
475
476			$logger->debug("Manager {$managerId} catch daemon shutdown, finish listening queue",
477			[__NAMESPACE__]);
478			}
479
480			if ($this->isLimitsExceeded() \|\| $this->isProducerShutDown()) {
481			break;
482			}
483
484			$this->waitDelay();
485
486			}
487			} catch (\Exception $ex) {
488			$logger->warning($ex->getMessage());
489			}
490
491			$logger->info("Manager {$managerId} finish work {$workerClass}", [__NAMESPACE__]);
492			$this->finishManagerLive();
493
494			return true;
495			}
496
497			//endregion ********************************************************
498
499			//region Worker ***********************************************************
500
501			protected static $workerConfig;
502
503			/**
504			* @param null $key
505			*
506			* @return mixed
507			*/
508			protected function getWorkerConfig($key = null)
509			{
510			/** @var AbstractWorker $class */
511			if (null === static::$workerConfig) {
512			$class = $this->workerClass;
513			static::$workerConfig = $class::getConfig();
514			}
515
516			return (null === $key)
517			? static::$workerConfig
518			: static::$workerConfig[$key];
519			}
520
521			//endregion ***********************************************************
522
523			/**
524			*
525			*/
526			protected function waitDelay()
527			{
528			$sleep = $this->getWorkerConfig(AbstractWorker::CONFIG_P_MANAGER_JOB_DONE_DELAY);
529			if ($sleep > 0) {
530			$logger = $this->getLogger();
531			$managerId = $this->getId();
532			$logger->debug("Manager {$managerId} job-done-after-sleep : {$sleep} ...", [__NAMESPACE__]);
533			sleep($sleep);
534			}
535			}
536
537			/**
538			* @return bool
539			*/
540			protected function isProducerShutDown()
541			{
542			if (!$this->getProducer()->isStateStart()) {
543			$logger = $this->getLogger();
544			$managerId = $this->getId();
545
546			$logger->info("Manager {$managerId} catch daemon shutdown, finish listening queue", [__NAMESPACE__]);
547
548			return true;
549			}
550
551			return false;
552			}
553
554
555			/**
556			* @return bool
557			*/
558			protected function isLimitsExceeded()
559			{
560
561			$events = $this->eventTriggerSet;
562			$wc = $this->getWorkerConfig();
563			$managerId = $this->getId();
564			$logger = $this->getLogger();
565
566			$jobsDoneCount
567			= $events[self::EVENT_SUCCESS] + $events[self::EVENT_FAIL] + $events[self::EVENT_ERROR];
568
569			$lim = $wc[AbstractWorker::CONFIG_P_JOBS_DONE_LIMIT];
570			if ($lim > 0 && $lim === $jobsDoneCount) {
571			$logger->debug("Manager {$managerId} done limit is equal {$jobsDoneCount}, so finish this manager ...",
572			[__NAMESPACE__]);
573
574			return true;
575			}
576
577			$lim = $wc[AbstractWorker::CONFIG_P_JOBS_SUCCESS_LIMIT];
578		View Code Duplication	if ($lim > 0 && $lim === $events[self::EVENT_SUCCESS]) {
			0 ignored issues – show Duplication introduced 2016-06-02 21:44 UTC by Report Bug Copy Issue Report Show Similar Issues like this This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
579			$logger->debug("Manager {$managerId} success limit is equal {$events['isSuccess']}, so finish this manager ...",
580			[__NAMESPACE__]);
581
582			return true;
583			}
584
585			$lim = $wc[AbstractWorker::CONFIG_P_JOBS_FAIL_LIMIT];
586		View Code Duplication	if ($lim > 0 && $lim === $events[self::EVENT_FAIL]) {
			0 ignored issues – show Duplication introduced 2016-06-02 21:44 UTC by Report Bug Copy Issue Report Show Similar Issues like this This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
587			$logger->debug("Manager {$managerId} fail limit is equal {$events['isFail']}, so finish this manager ...",
588			[__NAMESPACE__]);
589
590			return true;
591			}
592
593			$lim = $wc[AbstractWorker::CONFIG_P_JOBS_RETRY_LIMIT];
594		View Code Duplication	if ($lim > 0 && $lim === $events[self::EVENT_RETRY]) {
			0 ignored issues – show Duplication introduced 2016-06-02 21:44 UTC by Report Bug Copy Issue Report Show Similar Issues like this This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
595			$logger->debug("Manager {$managerId} retry limit is equal {$events['isRetry']}, so finish this manager ...",
596			[__NAMESPACE__]);
597
598			return true;
599			}
600
601			$lim = $wc[AbstractWorker::CONFIG_P_JOBS_ERROR_LIMIT];
602		View Code Duplication	if ($lim > 0 && $lim === $events[self::EVENT_ERROR]) {
			0 ignored issues – show Duplication introduced 2016-06-02 21:44 UTC by Report Bug Copy Issue Report Show Similar Issues like this This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
603			$logger->debug("Manager {$managerId} error limit is equal {$events['isError']}, so finish this manager ...",
604			[__NAMESPACE__]);
605
606			return true;
607			}
608
609			$lim = $wc[AbstractWorker::CONFIG_P_JOBS_REDIRECT_LIMIT];
610		View Code Duplication	if ($lim > 0 && $lim === $events[self::EVENT_REDIRECT]) {
			0 ignored issues – show Duplication introduced 2016-06-02 21:44 UTC by Report Bug Copy Issue Report Show Similar Issues like this This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
611			$logger->debug("Manager {$managerId} redirect limit is equal {$events['isRedirect']}, so finish this manager ...",
612			[__NAMESPACE__]);
613
614			return true;
615			}
616
617			$lim = $wc[AbstractWorker::CONFIG_P_MANAGER_LIFETIME];
618			if ($lim > 0 && $this->getStartOn() + $lim <= time()) {
619			$logger->debug("Manager {$managerId} lifetime limit {$lim}, so finish this manager ...", [__NAMESPACE__]);
620
621			return true;
622			}
623
624			return false;
625			}
626
627
628			}

cronario / cronario

Issues (18)

Security Analysis no request data

src/Manager.php (5 issues)

Labels

Severity

Introduced By

Upgrade to new PHP Analysis Engine

Duplication Side-by-Side

Filter issues like