Worker::raiseExceptionOccurredJobEvent() - Code Metrics - yiicod/yii2-jobqueue - Measure and Improve Code Quality continuously with Scrutinizer

Worker::raiseExceptionOccurredJobEvent() A
last analyzed 2018-04-27 17:36 UTC

↳ Parent: Worker

Complexity

Conditions	1
Paths	1

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	6
rs	9.4285
c	0
b	0
f	0
cc	1
eloc	3
nc	1
nop	3

<?php

namespace yiicod\jobqueue;

use Exception;
use Illuminate\Contracts\Debug\ExceptionHandler;
use Illuminate\Queue\Failed\FailedJobProviderInterface;
use Illuminate\Queue\Jobs\Job;
use Illuminate\Queue\MaxAttemptsExceededException;
use Illuminate\Queue\QueueManager;
use Illuminate\Queue\WorkerOptions;
use Throwable;
use Yii;
use yii\base\Event;
use yiicod\jobqueue\base\FatalThrowableError;
use yiicod\jobqueue\events\JobExceptionOccurredEvent;
use yiicod\jobqueue\events\JobFailedEvent;
use yiicod\jobqueue\events\JobProcessedEvent;
use yiicod\jobqueue\events\JobProcessingEvent;
use yiicod\jobqueue\events\WorkerStoppingEvent;
use yiicod\jobqueue\queues\MongoThreadQueue;

/**
 * Worker for laravel queues
 *
 * @author Orlov Alexey <[email protected]>
 * @author Virchenko Maksim <[email protected]>
 */
class Worker
{
    /**
     * Events
     */
    const EVENT_RAISE_BEFORE_JOB = 'raiseBeforeJobEvent';
    const EVENT_RAISE_AFTER_JOB = 'raiseAfterJobEvent';
    const EVENT_RAISE_EXCEPTION_OCCURED_JOB = 'raiseExceptionOccurredJobEvent';
    const EVENT_RAISE_FAILED_JOB = 'raiseFailedJobEvent';
    const EVENT_STOP = 'stop';
    /**
     * The queue manager instance.
     *
     * @var \Illuminate\Queue\QueueManager
     */
    protected $manager;

    /**
     * Failer instance
     *
     * @var FailedJobProviderInterface
     */
    protected $failer;

    /**
     * The exception handler instance.
     *
     * @var \Illuminate\Foundation\Exceptions\Handler
     */
    protected $exceptions;

    protected $shouldQuit = false;

    /**
     * Create a new queue worker.
     *
     * @param QueueManager $manager
     * @param FailedJobProviderInterface $failer
     * @param ExceptionHandler $exceptions
     */
    public function __construct(QueueManager $manager,
                                FailedJobProviderInterface $failer,
                                ExceptionHandler $exceptions)
    {
        $this->manager = $manager;
        $this->failer = $failer;
        $this->exceptions = $exceptions;

    }

    /**
     * Listen to the given queue in a loop.
     *
     * @param  string $connectionName
     * @param  string $queue
     * @param WorkerOptions $options
     *
     * @return array
     */
    public function daemon($connectionName, $queue, WorkerOptions $options)
    {
        while (true) {
            if ($this->shouldQuit) {
                $this->kill();
            }

            if (false === $this->runNextJob($connectionName, $queue, $options)) {
                $this->sleep($options->sleep);
            }

            if ($this->memoryExceeded($options->memory)) {
                $this->stop();
            }
        }
    }

    /**
     * Process the next job on the queue.
     *
     * @param  string $connectionName
     * @param  string $queue
     * @param  \Illuminate\Queue\WorkerOptions $options
     */
    public function runNextJob($connectionName, $queue, WorkerOptions $options)

    {
        /** @var MongoThreadQueue|Queue $connection */
        $connection = $this->manager->connection($connectionName);
        $job = $this->getNextJob($connection, $queue);

        // If we're able to pull a job off of the stack, we will process it and then return
        // from this method. If there is no job on the queue, we will "sleep" the worker
        // for the specified number of seconds, then keep processing jobs after sleep.
        if ($job instanceof Job && $connection->canRunJob($job)) {
            // If job can be run, then markJobAsReserved and run process
            $connection->markJobAsReserved($job);

            $this->runInBackground($job, $connectionName);

            return true;
        }

        return false;
    }

    /**
     * Make a Process for the Artisan command for the job id.
     *
     * @param Job $job
     * @param string $connectionName
     */
    protected function runInBackground(Job $job, string $connectionName)
    {
        $process = Yii::$container->get(JobProcess::class)->getProcess($job, $connectionName);

        $process->run();
    }

    /**
     * Process the given job from the queue.
     *
     * @param  string $connectionName
     * @param  \Illuminate\Contracts\Queue\Job $job
     * @param  \Illuminate\Queue\WorkerOptions $options
     *
     * @throws \Throwable
     */
    public function process($connectionName, $job, WorkerOptions $options)
    {
        try {
            // First we will raise the before job event and determine if the job has already ran
            // over the its maximum attempt limit, which could primarily happen if the job is
            // continually timing out and not actually throwing any exceptions from itself.
            $this->raiseBeforeJobEvent($connectionName, $job);

            $this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
                $connectionName, $job, (int)$options->maxTries
            );

            // Here we will fire off the job and let it process. We will catch any exceptions so
            // they can be reported to the developers logs, etc. Once the job is finished the
            // proper events will be fired to let any listeners know this job has finished.
            $job->fire();

            $this->raiseAfterJobEvent($connectionName, $job);
        } catch (Exception $e) {
            $this->handleJobException($connectionName, $job, $options, $e);
        } catch (Throwable $e) {

            $this->handleJobException(
                $connectionName, $job, $options, new FatalThrowableError($e)
            );
        }
    }

    /**
     * Handle an exception that occurred while the job was running.
     *
     * @param string $connectionName
     * @param \Illuminate\Contracts\Queue\Job $job
     * @param WorkerOptions $options
     * @param \Exception $e
     *
     * @throws \Exception
     */
    protected function handleJobException($connectionName, $job, WorkerOptions $options, $e)
    {
        try {
            // First, we will go ahead and mark the job as failed if it will exceed the maximum
            // attempts it is allowed to run the next time we process it. If so we will just
            // go ahead and mark it as failed now so we do not have to release this again.
            if (!$job->hasFailed()) {

                $this->markJobAsFailedIfWillExceedMaxAttempts(
                    $connectionName, $job, (int)$options->maxTries, $e
                );
            }

            $this->raiseExceptionOccurredJobEvent(
                $connectionName, $job, $e
            );
        } finally {
            // If we catch an exception, we will attempt to release the job back onto the queue
            // so it is not lost entirely. This'll let the job be retried at a later time by
            // another listener (or this same one). We will re-throw this exception after.
            if (!$job->isDeleted() && !$job->isReleased() && !$job->hasFailed()) {

                $job->release($options->delay);
            }
        }

        throw $e;
    }

    /**
     * Mark the given job as failed if it has exceeded the maximum allowed attempts.
     *
     * This will likely be because the job previously exceeded a timeout.
     *
     * @param  string $connectionName
     * @param  \Illuminate\Contracts\Queue\Job $job
     * @param  int $maxTries
     */
    protected function markJobAsFailedIfAlreadyExceedsMaxAttempts($connectionName, $job, $maxTries)
    {
        $maxTries = !is_null($job->maxTries()) ? $job->maxTries() : $maxTries;

        if (0 === $maxTries || $job->attempts() <= $maxTries) {
            return;
        }

        $this->failJob($connectionName, $job, $e = new MaxAttemptsExceededException(
            'A queued job has been attempted too many times. The job may have previously timed out.'
        ));

        throw $e;
    }

    /**
     * Mark the given job as failed if it has exceeded the maximum allowed attempts.
     *
     * @param  string $connectionName
     * @param  \Illuminate\Contracts\Queue\Job $job
     * @param  int $maxTries
     * @param  \Exception $e
     */
    protected function markJobAsFailedIfWillExceedMaxAttempts($connectionName, $job, $maxTries, $e)
    {
        $maxTries = !is_null($job->maxTries()) ? $job->maxTries() : $maxTries;

        if ($maxTries > 0 && $job->attempts() >= $maxTries) {
            $this->failJob($connectionName, $job, $e);
        }
    }

    /**
     * Get the next job from the queue connection.
     *
     * @param  \Illuminate\Contracts\Queue\Queue $connection
     * @param  string $queue
     *
     * @return \Illuminate\Contracts\Queue\Job|null
     */
    protected function getNextJob($connection, $queue)
    {
        try {
            foreach (explode(',', $queue) as $queue) {
                if (!is_null($job = $connection->pop($queue))) {
                    return $job;
                }
            }
        } catch (Exception $e) {
            $this->exceptions->report($e);
        } catch (Throwable $e) {

            $this->exceptions->report($e = new FatalThrowableError($e));
        }
    }

    /**
     * Kill the process.
     *
     * @param  int $status
     */
    public function kill($status = 0)
    {
        if (extension_loaded('posix')) {
            posix_kill(getmypid(), SIGKILL);
        }

        exit($status);
    }

    /**
     * Sleep the script for a given number of seconds.
     *
     * @param int $seconds
     */
    public function sleep($seconds)
    {
        sleep($seconds);
    }

    /**
     * Process the next job on the queue.
     *
     * @param  string $connectionName
     * @param $id
     * @param  \Illuminate\Queue\WorkerOptions $options
     */
    public function runJobById($connectionName, $id, WorkerOptions $options)
    {
        try {
            $job = $this->manager->connection($connectionName)->getJobById($id);
interface User
{
    /** @return string */
    public function getPassword();
}

class MyUser implements User
{
    public function getPassword()
    {
        // return something
    }

    public function getDisplayName()
    {
        // return some name.
    }
}

class AuthSystem
{
    public function authenticate(User $user)
    {
        $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName()));
        // do something.
    }
}

            // If we're able to pull a job off of the stack, we will process it and then return
            // from this method. If there is no job on the queue, we will "sleep" the worker
            // for the specified number of seconds, then keep processing jobs after sleep.
            if ($job) {
                return $this->process($connectionName, $job, $options);
            }
        } catch (Exception $e) {
            $this->exceptions->report($e);
        } catch (Throwable $e) {

            $this->exceptions->report(new FatalThrowableError($e));
        }

        $this->sleep($options->sleep);
    }

    /**
     * Mark the given job as failed and raise the relevant event.
     *
     * @param  string $connectionName
     * @param  \Illuminate\Contracts\Queue\Job $job
     * @param  \Exception $e
     */
    protected function failJob($connectionName, $job, $e)
    {
        if ($job->isDeleted()) {
            return;
        }

        try {
            // If the job has failed, we will delete it, call the "failed" method and then call
            // an event indicating the job has failed so it can be logged if needed. This is
            // to allow every developer to better keep monitor of their failed queue jobs.
            $job->delete();

            $job->failed($e);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
        } finally {
            $this->failer->log($connectionName, $job->getQueue(), $job->getRawBody(), $e);
            $this->raiseFailedJobEvent($connectionName, $job, $e);
        }
    }

    /**
     * Determine if the memory limit has been exceeded.
     *
     * @param  int $memoryLimit
     *
     * @return bool
     */
    public function memoryExceeded($memoryLimit)
    {
        return (memory_get_usage() / 1024 / 1024) >= $memoryLimit;
    }

    /**
     * Stop the worker if we have lost connection to a database.
     *
     * @param  \Exception $e
     */
    protected function stopWorkerIfLostConnection($e)
    {
        $message = $e->getMessage();
        $contains = [
            'server has gone away',
            'no connection to the server',
            'Lost connection',
            'is dead or not enabled',
            'Error while sending',
            'decryption failed or bad record mac',
            'server closed the connection unexpectedly',
            'SSL connection has been closed unexpectedly',
            'Error writing data to the connection',
            'Resource deadlock avoided',
            'Transaction() on null',
        ];

        foreach ($contains as $contain) {
            if (strpos($message, $contain)) {
                $this->shouldQuit = true;
            }
        }
    }

    /**
     * Raise the before queue job event.
     *
     * @param string $connectionName
     * @param \Illuminate\Contracts\Queue\Job $job
     */
    protected function raiseBeforeJobEvent($connectionName, $job)
    {
        Event::trigger(self::class, self::EVENT_RAISE_BEFORE_JOB, new JobProcessingEvent($connectionName, $job));
    }

    /**
     * Raise the after queue job event.
     *
     * @param string $connectionName
     * @param \Illuminate\Contracts\Queue\Job $job
     */
    protected function raiseAfterJobEvent($connectionName, $job)
    {
        Event::trigger(self::class, self::EVENT_RAISE_AFTER_JOB, new JobProcessedEvent($connectionName, $job));
    }

    /**
     * Raise the exception occurred queue job event.
     *
     * @param string $connectionName
     * @param \Illuminate\Contracts\Queue\Job $job
     * @param \Exception $e
     */
    protected function raiseExceptionOccurredJobEvent($connectionName, $job, $e)
    {
        Event::trigger(self::class, self::EVENT_RAISE_EXCEPTION_OCCURED_JOB, new JobExceptionOccurredEvent(
            $connectionName, $job, $e
        ));
    }

    /**
     * Raise the failed queue job event.
     *
     * @param  string $connectionName
     * @param  \Illuminate\Contracts\Queue\Job $job
     * @param  \Exception $e
     */
    protected function raiseFailedJobEvent($connectionName, $job, $e)
    {
        Event::trigger(self::class, self::EVENT_RAISE_FAILED_JOB, new JobFailedEvent(
            $connectionName, $job, $e
        ));
    }

    /**
     * Stop listening and bail out of the script.
     */
    public function stop($status = 0)

    {
        Event::trigger(self::class, self::EVENT_STOP, new WorkerStoppingEvent());

        Yii::$app->end();
    }
}


1			<?php
2
3			namespace yiicod\jobqueue;
4
5			use Exception;
6			use Illuminate\Contracts\Debug\ExceptionHandler;
7			use Illuminate\Queue\Failed\FailedJobProviderInterface;
8			use Illuminate\Queue\Jobs\Job;
9			use Illuminate\Queue\MaxAttemptsExceededException;
10			use Illuminate\Queue\QueueManager;
11			use Illuminate\Queue\WorkerOptions;
12			use Throwable;
13			use Yii;
14			use yii\base\Event;
15			use yiicod\jobqueue\base\FatalThrowableError;
16			use yiicod\jobqueue\events\JobExceptionOccurredEvent;
17			use yiicod\jobqueue\events\JobFailedEvent;
18			use yiicod\jobqueue\events\JobProcessedEvent;
19			use yiicod\jobqueue\events\JobProcessingEvent;
20			use yiicod\jobqueue\events\WorkerStoppingEvent;
21			use yiicod\jobqueue\queues\MongoThreadQueue;
22
23			/**
24			* Worker for laravel queues
25			*
26			* @author Orlov Alexey <[email protected]>
27			* @author Virchenko Maksim <[email protected]>
28			*/
29			class Worker
30			{
31			/**
32			* Events
33			*/
34			const EVENT_RAISE_BEFORE_JOB = 'raiseBeforeJobEvent';
35			const EVENT_RAISE_AFTER_JOB = 'raiseAfterJobEvent';
36			const EVENT_RAISE_EXCEPTION_OCCURED_JOB = 'raiseExceptionOccurredJobEvent';
37			const EVENT_RAISE_FAILED_JOB = 'raiseFailedJobEvent';
38			const EVENT_STOP = 'stop';
39			/**
40			* The queue manager instance.
41			*
42			* @var \Illuminate\Queue\QueueManager
43			*/
44			protected $manager;
45
46			/**
47			* Failer instance
48			*
49			* @var FailedJobProviderInterface
50			*/
51			protected $failer;
52
53			/**
54			* The exception handler instance.
55			*
56			* @var \Illuminate\Foundation\Exceptions\Handler
57			*/
58			protected $exceptions;
59
60			protected $shouldQuit = false;
61
62			/**
63			* Create a new queue worker.
64			*
65			* @param QueueManager $manager
66			* @param FailedJobProviderInterface $failer
67			* @param ExceptionHandler $exceptions
68			*/
69			public function __construct(QueueManager $manager,
70			FailedJobProviderInterface $failer,
71			ExceptionHandler $exceptions)
72			{
73			$this->manager = $manager;
74			$this->failer = $failer;
75			$this->exceptions = $exceptions;
			0 ignored issues – show Documentation Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report It seems like `$exceptions` of type `object<Illuminate\Contra...Debug\ExceptionHandler>` is incompatible with the declared type `object<Illuminate\Foundation\Exceptions\Handler>` of property `$exceptions`. Our type inference engine has found an assignment to a property that is incompatible with the declared type of that property. Either this assignment is in error or the assigned type should be added to the documentation/type hint for that property.. Loading history...
76			}
77
78			/**
79			* Listen to the given queue in a loop.
80			*
81			* @param string $connectionName
82			* @param string $queue
83			* @param WorkerOptions $options
84			*
85			* @return array
86			*/
87			public function daemon($connectionName, $queue, WorkerOptions $options)
88			{
89			while (true) {
90			if ($this->shouldQuit) {
91			$this->kill();
92			}
93
94			if (false === $this->runNextJob($connectionName, $queue, $options)) {
95			$this->sleep($options->sleep);
96			}
97
98			if ($this->memoryExceeded($options->memory)) {
99			$this->stop();
100			}
101			}
102			}
103
104			/**
105			* Process the next job on the queue.
106			*
107			* @param string $connectionName
108			* @param string $queue
109			* @param \Illuminate\Queue\WorkerOptions $options
110			*/
111			public function runNextJob($connectionName, $queue, WorkerOptions $options)
			0 ignored issues – show Unused Code introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report The parameter `$options` is not used and could be removed. This check looks from parameters that have been defined for a function or method, but which are not used in the method body. Loading history...
112			{
113			/** @var MongoThreadQueue\|Queue $connection */
114			$connection = $this->manager->connection($connectionName);
115			$job = $this->getNextJob($connection, $queue);
116
117			// If we're able to pull a job off of the stack, we will process it and then return
118			// from this method. If there is no job on the queue, we will "sleep" the worker
119			// for the specified number of seconds, then keep processing jobs after sleep.
120			if ($job instanceof Job && $connection->canRunJob($job)) {
121			// If job can be run, then markJobAsReserved and run process
122			$connection->markJobAsReserved($job);
123
124			$this->runInBackground($job, $connectionName);
125
126			return true;
127			}
128
129			return false;
130			}
131
132			/**
133			* Make a Process for the Artisan command for the job id.
134			*
135			* @param Job $job
136			* @param string $connectionName
137			*/
138			protected function runInBackground(Job $job, string $connectionName)
139			{
140			$process = Yii::$container->get(JobProcess::class)->getProcess($job, $connectionName);
141
142			$process->run();
143			}
144
145			/**
146			* Process the given job from the queue.
147			*
148			* @param string $connectionName
149			* @param \Illuminate\Contracts\Queue\Job $job
150			* @param \Illuminate\Queue\WorkerOptions $options
151			*
152			* @throws \Throwable
153			*/
154			public function process($connectionName, $job, WorkerOptions $options)
155			{
156			try {
157			// First we will raise the before job event and determine if the job has already ran
158			// over the its maximum attempt limit, which could primarily happen if the job is
159			// continually timing out and not actually throwing any exceptions from itself.
160			$this->raiseBeforeJobEvent($connectionName, $job);
161
162			$this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
163			$connectionName, $job, (int)$options->maxTries
164			);
165
166			// Here we will fire off the job and let it process. We will catch any exceptions so
167			// they can be reported to the developers logs, etc. Once the job is finished the
168			// proper events will be fired to let any listeners know this job has finished.
169			$job->fire();
170
171			$this->raiseAfterJobEvent($connectionName, $job);
172			} catch (Exception $e) {
173			$this->handleJobException($connectionName, $job, $options, $e);
174			} catch (Throwable $e) {
			0 ignored issues – show Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report 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? Loading history...
175			$this->handleJobException(
176			$connectionName, $job, $options, new FatalThrowableError($e)
177			);
178			}
179			}
180
181			/**
182			* Handle an exception that occurred while the job was running.
183			*
184			* @param string $connectionName
185			* @param \Illuminate\Contracts\Queue\Job $job
186			* @param WorkerOptions $options
187			* @param \Exception $e
188			*
189			* @throws \Exception
190			*/
191			protected function handleJobException($connectionName, $job, WorkerOptions $options, $e)
192			{
193			try {
194			// First, we will go ahead and mark the job as failed if it will exceed the maximum
195			// attempts it is allowed to run the next time we process it. If so we will just
196			// go ahead and mark it as failed now so we do not have to release this again.
197			if (!$job->hasFailed()) {
			0 ignored issues – show Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report The method `hasFailed()` does not exist on `Illuminate\Contracts\Queue\Job`. Did you maybe mean `failed()`? This check marks calls to methods that do not seem to exist on an object. This is most likely the result of a method being renamed without all references to it being renamed likewise. Loading history...
198			$this->markJobAsFailedIfWillExceedMaxAttempts(
199			$connectionName, $job, (int)$options->maxTries, $e
200			);
201			}
202
203			$this->raiseExceptionOccurredJobEvent(
204			$connectionName, $job, $e
205			);
206			} finally {
207			// If we catch an exception, we will attempt to release the job back onto the queue
208			// so it is not lost entirely. This'll let the job be retried at a later time by
209			// another listener (or this same one). We will re-throw this exception after.
210			if (!$job->isDeleted() && !$job->isReleased() && !$job->hasFailed()) {
			0 ignored issues – show Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report The method `isReleased()` does not exist on `Illuminate\Contracts\Queue\Job`. Did you maybe mean `release()`? This check marks calls to methods that do not seem to exist on an object. This is most likely the result of a method being renamed without all references to it being renamed likewise. Loading history... Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report The method `hasFailed()` does not exist on `Illuminate\Contracts\Queue\Job`. Did you maybe mean `failed()`? This check marks calls to methods that do not seem to exist on an object. This is most likely the result of a method being renamed without all references to it being renamed likewise. Loading history...
211			$job->release($options->delay);
212			}
213			}
214
215			throw $e;
216			}
217
218			/**
219			* Mark the given job as failed if it has exceeded the maximum allowed attempts.
220			*
221			* This will likely be because the job previously exceeded a timeout.
222			*
223			* @param string $connectionName
224			* @param \Illuminate\Contracts\Queue\Job $job
225			* @param int $maxTries
226			*/
227			protected function markJobAsFailedIfAlreadyExceedsMaxAttempts($connectionName, $job, $maxTries)
228			{
229			$maxTries = !is_null($job->maxTries()) ? $job->maxTries() : $maxTries;
230
231			if (0 === $maxTries \|\| $job->attempts() <= $maxTries) {
232			return;
233			}
234
235			$this->failJob($connectionName, $job, $e = new MaxAttemptsExceededException(
236			'A queued job has been attempted too many times. The job may have previously timed out.'
237			));
238
239			throw $e;
240			}
241
242			/**
243			* Mark the given job as failed if it has exceeded the maximum allowed attempts.
244			*
245			* @param string $connectionName
246			* @param \Illuminate\Contracts\Queue\Job $job
247			* @param int $maxTries
248			* @param \Exception $e
249			*/
250			protected function markJobAsFailedIfWillExceedMaxAttempts($connectionName, $job, $maxTries, $e)
251			{
252			$maxTries = !is_null($job->maxTries()) ? $job->maxTries() : $maxTries;
253
254			if ($maxTries > 0 && $job->attempts() >= $maxTries) {
255			$this->failJob($connectionName, $job, $e);
256			}
257			}
258
259			/**
260			* Get the next job from the queue connection.
261			*
262			* @param \Illuminate\Contracts\Queue\Queue $connection
263			* @param string $queue
264			*
265			* @return \Illuminate\Contracts\Queue\Job\|null
266			*/
267			protected function getNextJob($connection, $queue)
268			{
269			try {
270			foreach (explode(',', $queue) as $queue) {
271			if (!is_null($job = $connection->pop($queue))) {
272			return $job;
273			}
274			}
275			} catch (Exception $e) {
276			$this->exceptions->report($e);
277			} catch (Throwable $e) {
			0 ignored issues – show Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report 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? Loading history...
278			$this->exceptions->report($e = new FatalThrowableError($e));
279			}
280			}
281
282			/**
283			* Kill the process.
284			*
285			* @param int $status
286			*/
287			public function kill($status = 0)
288			{
289			if (extension_loaded('posix')) {
290			posix_kill(getmypid(), SIGKILL);
291			}
292
293			exit($status);
294			}
295
296			/**
297			* Sleep the script for a given number of seconds.
298			*
299			* @param int $seconds
300			*/
301			public function sleep($seconds)
302			{
303			sleep($seconds);
304			}
305
306			/**
307			* Process the next job on the queue.
308			*
309			* @param string $connectionName
310			* @param $id
311			* @param \Illuminate\Queue\WorkerOptions $options
312			*/
313			public function runJobById($connectionName, $id, WorkerOptions $options)
314			{
315			try {
316			$job = $this->manager->connection($connectionName)->getJobById($id);
			0 ignored issues – show Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report It seems like you code against a concrete implementation and not the interface `Illuminate\Contracts\Queue\Queue` as the method `getJobById()` does only exist in the following implementations of said interface: `yiicod\jobqueue\queues\MongoThreadQueue`. Let’s take a look at an example: interface User { /** @return string / public function getPassword(); } class MyUser implements User { public function getPassword() { // return something } public function getDisplayName() { // return some name. } } class AuthSystem { public function authenticate(User $user) { $this->logger->info(sprintf('Authenticating %s.', $user->getDisplayName())); // do something. } } In the above example, the authenticate() method works fine as long as you just pass instances of MyUser. However, if you now also want to pass a different implementation of User which does not have a getDisplayName() method, the code will break. Available Fixes Change the type-hint for the parameter: class AuthSystem { public function authenticate(MyUser $user) { / ... / } } Add an additional type-check: class AuthSystem { public function authenticate(User $user) { if ($user instanceof MyUser) { $this->logger->info(/* ... /); } // or alternatively if ( ! $user instanceof MyUser) { throw new \LogicException( '$user must be an instance of MyUser, ' .'other instances are not supported.' ); } } } Note:* PHP Analyzer uses reverse abstract interpretation to narrow down the types inside the if block in such a case. Add the method to the interface: interface User { /** @return string / public function getPassword(); /* @return string */ public function getDisplayName(); } Loading history...
317
318			// If we're able to pull a job off of the stack, we will process it and then return
319			// from this method. If there is no job on the queue, we will "sleep" the worker
320			// for the specified number of seconds, then keep processing jobs after sleep.
321			if ($job) {
322			return $this->process($connectionName, $job, $options);
323			}
324			} catch (Exception $e) {
325			$this->exceptions->report($e);
326			} catch (Throwable $e) {
			0 ignored issues – show Bug introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report 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? Loading history...
327			$this->exceptions->report(new FatalThrowableError($e));
328			}
329
330			$this->sleep($options->sleep);
331			}
332
333			/**
334			* Mark the given job as failed and raise the relevant event.
335			*
336			* @param string $connectionName
337			* @param \Illuminate\Contracts\Queue\Job $job
338			* @param \Exception $e
339			*/
340			protected function failJob($connectionName, $job, $e)
341			{
342			if ($job->isDeleted()) {
343			return;
344			}
345
346			try {
347			// If the job has failed, we will delete it, call the "failed" method and then call
348			// an event indicating the job has failed so it can be logged if needed. This is
349			// to allow every developer to better keep monitor of their failed queue jobs.
350			$job->delete();
351
352			$job->failed($e);
			0 ignored issues – show Documentation introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report `$e` is of type `object<Exception>`, but the function expects a `object<Throwable>`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
353			} finally {
354			$this->failer->log($connectionName, $job->getQueue(), $job->getRawBody(), $e);
355			$this->raiseFailedJobEvent($connectionName, $job, $e);
356			}
357			}
358
359			/**
360			* Determine if the memory limit has been exceeded.
361			*
362			* @param int $memoryLimit
363			*
364			* @return bool
365			*/
366			public function memoryExceeded($memoryLimit)
367			{
368			return (memory_get_usage() / 1024 / 1024) >= $memoryLimit;
369			}
370
371			/**
372			* Stop the worker if we have lost connection to a database.
373			*
374			* @param \Exception $e
375			*/
376			protected function stopWorkerIfLostConnection($e)
377			{
378			$message = $e->getMessage();
379			$contains = [
380			'server has gone away',
381			'no connection to the server',
382			'Lost connection',
383			'is dead or not enabled',
384			'Error while sending',
385			'decryption failed or bad record mac',
386			'server closed the connection unexpectedly',
387			'SSL connection has been closed unexpectedly',
388			'Error writing data to the connection',
389			'Resource deadlock avoided',
390			'Transaction() on null',
391			];
392
393			foreach ($contains as $contain) {
394			if (strpos($message, $contain)) {
395			$this->shouldQuit = true;
396			}
397			}
398			}
399
400			/**
401			* Raise the before queue job event.
402			*
403			* @param string $connectionName
404			* @param \Illuminate\Contracts\Queue\Job $job
405			*/
406			protected function raiseBeforeJobEvent($connectionName, $job)
407			{
408			Event::trigger(self::class, self::EVENT_RAISE_BEFORE_JOB, new JobProcessingEvent($connectionName, $job));
409			}
410
411			/**
412			* Raise the after queue job event.
413			*
414			* @param string $connectionName
415			* @param \Illuminate\Contracts\Queue\Job $job
416			*/
417			protected function raiseAfterJobEvent($connectionName, $job)
418			{
419			Event::trigger(self::class, self::EVENT_RAISE_AFTER_JOB, new JobProcessedEvent($connectionName, $job));
420			}
421
422			/**
423			* Raise the exception occurred queue job event.
424			*
425			* @param string $connectionName
426			* @param \Illuminate\Contracts\Queue\Job $job
427			* @param \Exception $e
428			*/
429			protected function raiseExceptionOccurredJobEvent($connectionName, $job, $e)
430			{
431			Event::trigger(self::class, self::EVENT_RAISE_EXCEPTION_OCCURED_JOB, new JobExceptionOccurredEvent(
432			$connectionName, $job, $e
433			));
434			}
435
436			/**
437			* Raise the failed queue job event.
438			*
439			* @param string $connectionName
440			* @param \Illuminate\Contracts\Queue\Job $job
441			* @param \Exception $e
442			*/
443			protected function raiseFailedJobEvent($connectionName, $job, $e)
444			{
445			Event::trigger(self::class, self::EVENT_RAISE_FAILED_JOB, new JobFailedEvent(
446			$connectionName, $job, $e
447			));
448			}
449
450			/**
451			* Stop listening and bail out of the script.
452			*/
453			public function stop($status = 0)
			0 ignored issues – show Unused Code introduced 2018-02-18 20:08 UTC by Report Bug Copy Issue Report The parameter `$status` is not used and could be removed. This check looks from parameters that have been defined for a function or method, but which are not used in the method body. Loading history...
454			{
455			Event::trigger(self::class, self::EVENT_STOP, new WorkerStoppingEvent());
456
457			Yii::$app->end();
458			}
459			}
460

yiicod / yii2-jobqueue

Worker::raiseExceptionOccurredJobEvent() A last analyzed 2018-04-27 17:36 UTC

Complexity

Size

Duplication

Importance

Available Fixes

Duplication Side-by-Side

Filter issues like

Worker::raiseExceptionOccurredJobEvent() A
last analyzed 2018-04-27 17:36 UTC