Consumer::bind() - Code Metrics - Inspection of "Merge pull request #373 from bernardphp/routers" - bernardphp/bernard - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( cb8968...43f345 )

by Márk

created 2018-02-18 22:15 UTC

Consumer::bind() A

↳ Parent: Consumer

Complexity

Conditions	2
Paths	2

Size

Total Lines	10
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

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

<?php

namespace Bernard;

use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use Bernard\Event\EnvelopeEvent;
use Bernard\Event\PingEvent;
use Bernard\Event\RejectEnvelopeEvent;

class Consumer
{
    protected $router;
    protected $dispatcher;
    protected $shutdown = false;
    protected $pause = false;
    protected $configured = false;
    protected $options = [
        'max-runtime' => PHP_INT_MAX,
        'max-messages' => null,
        'stop-when-empty' => false,
        'stop-on-error' => false,
    ];

    /**
     * @param Router                   $router
     * @param EventDispatcherInterface $dispatcher
     */
    public function __construct(Router $router, EventDispatcherInterface $dispatcher)
    {
        $this->router = $router;
        $this->dispatcher = $dispatcher;
    }

    /**
     * Starts an infinite loop calling Consumer::tick();.
     *
     * @param Queue $queue
     * @param array $options
     */
    public function consume(Queue $queue, array $options = [])
    {
        declare(ticks=1);

        $this->bind();

        while ($this->tick($queue, $options)) {

            // NO op
        }
    }

    /**
     * Returns true do indicate it should be run again or false to indicate
     * it should not be run again.
     *
     * @param Queue $queue
     * @param array $options
     *
     * @return bool
     */
    public function tick(Queue $queue, array $options = [])
    {
        $this->configure($options);

        if ($this->shutdown) {
            return false;
        }

        if (microtime(true) > $this->options['max-runtime']) {
            return false;
        }

        if ($this->pause) {
            return true;
        }

        $this->dispatcher->dispatch(BernardEvents::PING, new PingEvent($queue));

        if (!$envelope = $queue->dequeue()) {
            return !$this->options['stop-when-empty'];
        }

        $this->invoke($envelope, $queue);

        if (null === $this->options['max-messages']) {
            return true;
        }

        return (bool) --$this->options['max-messages'];
    }

    /**
     * Mark Consumer as shutdown.
     */
    public function shutdown()
    {
        $this->shutdown = true;
    }

    /**
     * Pause consuming.
     */
    public function pause()
    {
        $this->pause = true;
    }

    /**
     * Resume consuming.
     */
    public function resume()
    {
        $this->pause = false;
    }

    /**
     * Until there is a real extension point to doing invoked stuff, this can be used
     * by wrapping the invoke method.
     *
     * @param Envelope $envelope
     * @param Queue    $queue
     *
     * @throws \Exception
     * @throws \Throwable
     */
    public function invoke(Envelope $envelope, Queue $queue)
    {
        try {
            $this->dispatcher->dispatch(BernardEvents::INVOKE, new EnvelopeEvent($envelope, $queue));

            $receiver = $this->router->route($envelope);
            $receiver->receive($envelope->getMessage());

            // We successfully processed the message.
            $queue->acknowledge($envelope);

            $this->dispatcher->dispatch(BernardEvents::ACKNOWLEDGE, new EnvelopeEvent($envelope, $queue));
        } catch (\Throwable $error) {

            $this->rejectDispatch($error, $envelope, $queue);
        } catch (\Exception $exception) {
            $this->rejectDispatch($exception, $envelope, $queue);
        }
    }

    /**
     * @param array $options
     */
    protected function configure(array $options)
    {
        if ($this->configured) {
            return $this->options;
        }

        $this->options = array_filter($options) + $this->options;
        $this->options['max-runtime'] += microtime(true);
        $this->configured = true;
    }

    /**
     * Setup signal handlers for unix signals.
     *
     * If the process control extension does not exist (e.g. on Windows), ignore the signal handlers.
     * The difference is that when terminating the consumer, running processes will not stop gracefully
     * and will terminate immediately.
     */
    protected function bind()
    {
        if (function_exists('pcntl_signal')) {
            pcntl_signal(SIGTERM, [$this, 'shutdown']);
            pcntl_signal(SIGINT, [$this, 'shutdown']);
            pcntl_signal(SIGQUIT, [$this, 'shutdown']);
            pcntl_signal(SIGUSR2, [$this, 'pause']);
            pcntl_signal(SIGCONT, [$this, 'resume']);
        }
    }

    /**
     * @param \Throwable|\Exception $exception note that the type-hint is missing due to PHP 5.x compat
     * @param Envelope              $envelope
     * @param Queue                 $queue
     *
     * @throws \Exception
     * @throws \Throwable
     */
    private function rejectDispatch($exception, Envelope $envelope, Queue $queue)
    {
        // Make sure the exception is not interfering.
        // Previously failing jobs handling have been moved to a middleware.
        //
        // Emit an event to let others log that exception
        $this->dispatcher->dispatch(BernardEvents::REJECT, new RejectEnvelopeEvent($envelope, $queue, $exception));

        if ($this->options['stop-on-error']) {
            throw $exception;
        }
    }
}


1		<?php
2
3		namespace Bernard;
4
5		use Symfony\Component\EventDispatcher\EventDispatcherInterface;
6		use Bernard\Event\EnvelopeEvent;
7		use Bernard\Event\PingEvent;
8		use Bernard\Event\RejectEnvelopeEvent;
9
10		class Consumer
11		{
12		protected $router;
13		protected $dispatcher;
14		protected $shutdown = false;
15		protected $pause = false;
16		protected $configured = false;
17		protected $options = [
18		'max-runtime' => PHP_INT_MAX,
19		'max-messages' => null,
20		'stop-when-empty' => false,
21		'stop-on-error' => false,
22		];
23
24		/**
25		* @param Router $router
26		* @param EventDispatcherInterface $dispatcher
27		*/
28	11	public function __construct(Router $router, EventDispatcherInterface $dispatcher)
29		{
30	11	$this->router = $router;
31	11	$this->dispatcher = $dispatcher;
32	11	}
33
34		/**
35		* Starts an infinite loop calling Consumer::tick();.
36		*
37		* @param Queue $queue
38		* @param array $options
39		*/
40		public function consume(Queue $queue, array $options = [])
41		{
42		declare(ticks=1);
43
44		$this->bind();
45
46		while ($this->tick($queue, $options)) {
		0 ignored issues – show Unused Code introduced 2018-02-15 06:37 UTC by Report Bug Copy Issue Report This `while` loop is empty and can be removed. This check looks for `while` loops that have no statements or where all statements have been commented out. This may be the result of changes for debugging or the code may simply be obsolete. Consider removing the loop. Loading history...
47		// NO op
48		}
49		}
50
51		/**
52		* Returns true do indicate it should be run again or false to indicate
53		* it should not be run again.
54		*
55		* @param Queue $queue
56		* @param array $options
57		*
58		* @return bool
59		*/
60	10	public function tick(Queue $queue, array $options = [])
61		{
62	10	$this->configure($options);
63
64	10	if ($this->shutdown) {
65	1	return false;
66		}
67
68	9	if (microtime(true) > $this->options['max-runtime']) {
69	1	return false;
70		}
71
72	8	if ($this->pause) {
73	1	return true;
74		}
75
76	8	$this->dispatcher->dispatch(BernardEvents::PING, new PingEvent($queue));
77
78	8	if (!$envelope = $queue->dequeue()) {
79	2	return !$this->options['stop-when-empty'];
80		}
81
82	7	$this->invoke($envelope, $queue);
83
84	6	if (null === $this->options['max-messages']) {
85	5	return true;
86		}
87
88	1	return (bool) --$this->options['max-messages'];
89		}
90
91		/**
92		* Mark Consumer as shutdown.
93		*/
94	1	public function shutdown()
95		{
96	1	$this->shutdown = true;
97	1	}
98
99		/**
100		* Pause consuming.
101		*/
102	1	public function pause()
103		{
104	1	$this->pause = true;
105	1	}
106
107		/**
108		* Resume consuming.
109		*/
110	1	public function resume()
111		{
112	1	$this->pause = false;
113	1	}
114
115		/**
116		* Until there is a real extension point to doing invoked stuff, this can be used
117		* by wrapping the invoke method.
118		*
119		* @param Envelope $envelope
120		* @param Queue $queue
121		*
122		* @throws \Exception
123		* @throws \Throwable
124		*/
125	8	public function invoke(Envelope $envelope, Queue $queue)
126		{
127		try {
128	8	$this->dispatcher->dispatch(BernardEvents::INVOKE, new EnvelopeEvent($envelope, $queue));
129
130	8	$receiver = $this->router->route($envelope);
131	7	$receiver->receive($envelope->getMessage());
132
133		// We successfully processed the message.
134	6	$queue->acknowledge($envelope);
135
136	6	$this->dispatcher->dispatch(BernardEvents::ACKNOWLEDGE, new EnvelopeEvent($envelope, $queue));
137	8	} catch (\Throwable $error) {
		0 ignored issues – show Bug introduced 2016-03-07 06:41 UTC by Report Bug Copy Issue Report The class `Throwable` does not exist. Did you forget a USE statement, or did you not list all dependencies? Scrutinizer analyzes your `composer.json`/`composer.lock` file if available to determine the classes, and functions that are defined by your dependencies. It seems like the listed class was neither found in your dependencies, nor was it found in the analyzed files in your repository. If you are using some other form of dependency management, you might want to disable this analysis. Loading history...
138		$this->rejectDispatch($error, $envelope, $queue);
139	2	} catch (\Exception $exception) {
140	2	$this->rejectDispatch($exception, $envelope, $queue);
141		}
142	7	}
143
144		/**
145		* @param array $options
146		*/
147	10	protected function configure(array $options)
148		{
149	10	if ($this->configured) {
150	3	return $this->options;
151		}
152
153	10	$this->options = array_filter($options) + $this->options;
154	10	$this->options['max-runtime'] += microtime(true);
155	10	$this->configured = true;
156	10	}
157
158		/**
159		* Setup signal handlers for unix signals.
160		*
161		* If the process control extension does not exist (e.g. on Windows), ignore the signal handlers.
162		* The difference is that when terminating the consumer, running processes will not stop gracefully
163		* and will terminate immediately.
164		*/
165		protected function bind()
166		{
167		if (function_exists('pcntl_signal')) {
168		pcntl_signal(SIGTERM, [$this, 'shutdown']);
169		pcntl_signal(SIGINT, [$this, 'shutdown']);
170		pcntl_signal(SIGQUIT, [$this, 'shutdown']);
171		pcntl_signal(SIGUSR2, [$this, 'pause']);
172		pcntl_signal(SIGCONT, [$this, 'resume']);
173		}
174		}
175
176		/**
177		* @param \Throwable\|\Exception $exception note that the type-hint is missing due to PHP 5.x compat
178		* @param Envelope $envelope
179		* @param Queue $queue
180		*
181		* @throws \Exception
182		* @throws \Throwable
183		*/
184	2	private function rejectDispatch($exception, Envelope $envelope, Queue $queue)
185		{
186		// Make sure the exception is not interfering.
187		// Previously failing jobs handling have been moved to a middleware.
188		//
189		// Emit an event to let others log that exception
190	2	$this->dispatcher->dispatch(BernardEvents::REJECT, new RejectEnvelopeEvent($envelope, $queue, $exception));
191
192	2	if ($this->options['stop-on-error']) {
193	1	throw $exception;
194		}
195	1	}
196		}
197

bernardphp / bernard

GitHub Access Token became invalid

Push — master ( cb8968...43f345 )

Consumer::bind() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like