Consumer::__construct() - Code Metrics - Inspection of "Extract consumer call_user_func to protected metho..." - bernardphp/bernard - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#324)

unknown

created 2017-08-02 08:12 UTC

Consumer::__construct() A

↳ Parent: Consumer

Complexity

Conditions	1
Paths	1

Size

Total Lines	5
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

<?php

namespace Bernard;

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

/**
 * @package Consumer
 */
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 (boolean) --$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));

            $this->call($envelope);

            // 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 Envelope $envelope
     */
    protected function call(Envelope $envelope)
    {
        // for 5.3 support where a function name is not a callable
        call_user_func($this->router->map($envelope), $envelope->getMessage());
    }

    /**
     * @param array $options
     *
     * @return void
     */
    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			/**
11			* @package Consumer
12			*/
13			class Consumer
14			{
15			protected $router;
16			protected $dispatcher;
17			protected $shutdown = false;
18			protected $pause = false;
19			protected $configured = false;
20			protected $options = [
21			'max-runtime' => PHP_INT_MAX,
22			'max-messages' => null,
23			'stop-when-empty' => false,
24			'stop-on-error' => false,
25			];
26
27			/**
28			* @param Router $router
29			* @param EventDispatcherInterface $dispatcher
30			*/
31			public function __construct(Router $router, EventDispatcherInterface $dispatcher)
32			{
33			$this->router = $router;
34			$this->dispatcher = $dispatcher;
35			}
36
37			/**
38			* Starts an infinite loop calling Consumer::tick();
39			*
40			* @param Queue $queue
41			* @param array $options
42			*/
43			public function consume(Queue $queue, array $options = [])
44			{
45			declare (ticks = 1);
46
47			$this->bind();
48
49			while ($this->tick($queue, $options)) {
50			// NO op
51			}
52			}
53
54			/**
55			* Returns true do indicate it should be run again or false to indicate
56			* it should not be run again.
57			*
58			* @param Queue $queue
59			* @param array $options
60			*
61			* @return bool
62			*/
63			public function tick(Queue $queue, array $options = [])
64			{
65			$this->configure($options);
66
67			if ($this->shutdown) {
68			return false;
69			}
70
71			if (microtime(true) > $this->options['max-runtime']) {
72			return false;
73			}
74
75			if ($this->pause) {
76			return true;
77			}
78
79			$this->dispatcher->dispatch(BernardEvents::PING, new PingEvent($queue));
80
81			if (!$envelope = $queue->dequeue()) {
82			return !$this->options['stop-when-empty'];
83			}
84
85			$this->invoke($envelope, $queue);
86
87			if (null === $this->options['max-messages']) {
88			return true;
89			}
90
91			return (boolean) --$this->options['max-messages'];
92			}
93
94			/**
95			* Mark Consumer as shutdown
96			*/
97			public function shutdown()
98			{
99			$this->shutdown = true;
100			}
101
102			/**
103			* Pause consuming
104			*/
105			public function pause()
106			{
107			$this->pause = true;
108			}
109
110			/**
111			* Resume consuming
112			*/
113			public function resume()
114			{
115			$this->pause = false;
116			}
117
118			/**
119			* Until there is a real extension point to doing invoked stuff, this can be used
120			* by wrapping the invoke method.
121			*
122			* @param Envelope $envelope
123			* @param Queue $queue
124			*
125			* @throws \Exception
126			* @throws \Throwable
127			*/
128			public function invoke(Envelope $envelope, Queue $queue)
129			{
130			try {
131			$this->dispatcher->dispatch(BernardEvents::INVOKE, new EnvelopeEvent($envelope, $queue));
132
133			$this->call($envelope);
134
135			// We successfully processed the message.
136			$queue->acknowledge($envelope);
137
138			$this->dispatcher->dispatch(BernardEvents::ACKNOWLEDGE, new EnvelopeEvent($envelope, $queue));
139			} 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...
140			$this->rejectDispatch($error, $envelope, $queue);
141			} catch (\Exception $exception) {
142			$this->rejectDispatch($exception, $envelope, $queue);
143			}
144			}
145
146			/**
147			* @param Envelope $envelope
148			*/
149			protected function call(Envelope $envelope)
150			{
151			// for 5.3 support where a function name is not a callable
152			call_user_func($this->router->map($envelope), $envelope->getMessage());
153			}
154
155			/**
156			* @param array $options
157			*
158			* @return void
159			*/
160			protected function configure(array $options)
161			{
162			if ($this->configured) {
163			return $this->options;
164			}
165
166			$this->options = array_filter($options) + $this->options;
167			$this->options['max-runtime'] += microtime(true);
168			$this->configured = true;
169			}
170
171			/**
172			* Setup signal handlers for unix signals.
173			*
174			* If the process control extension does not exist (e.g. on Windows), ignore the signal handlers.
175			* The difference is that when terminating the consumer, running processes will not stop gracefully
176			* and will terminate immediately.
177			*/
178			protected function bind()
179			{
180			if (function_exists('pcntl_signal')) {
181			pcntl_signal(SIGTERM, [$this, 'shutdown']);
182			pcntl_signal(SIGINT, [$this, 'shutdown']);
			0 ignored issues – show Coding Style introduced 2016-02-12 14:02 UTC by Report Bug Copy Issue Report Expected 1 space instead of 2 after comma in function call. Loading history...
183			pcntl_signal(SIGQUIT, [$this, 'shutdown']);
184			pcntl_signal(SIGUSR2, [$this, 'pause']);
185			pcntl_signal(SIGCONT, [$this, 'resume']);
186			}
187			}
188
189			/**
190			* @param \Throwable\|\Exception $exception note that the type-hint is missing due to PHP 5.x compat
191			*
192			* @param Envelope $envelope
193			* @param Queue $queue
194			*
195			* @throws \Exception
196			* @throws \Throwable
197			*/
198			private function rejectDispatch($exception, Envelope $envelope, Queue $queue)
199			{
200			// Make sure the exception is not interfering.
201			// Previously failing jobs handling have been moved to a middleware.
202			//
203			// Emit an event to let others log that exception
204			$this->dispatcher->dispatch(BernardEvents::REJECT, new RejectEnvelopeEvent($envelope, $queue, $exception));
205
206			if ($this->options['stop-on-error']) {
207			throw $exception;
208			}
209			}
210			}
211

bernardphp / bernard

GitHub Access Token became invalid

Pull Request — master (#324)

Consumer::__construct() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like