AMQPBackendDispatcher::resolveDelayStrategy() - Code Metrics - Inspection of "Change DelayStrategy as a configuration option" - sonata-project/SonataNotificationBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — 3.x (#387)

unknown

created 2019-07-11 11:44 UTC

AMQPBackendDispatcher::resolveDelayStrategy() A

↳ Parent: AMQPBackendDispatcher

Complexity

Conditions	4
Paths	2

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	19
rs	9.6333
c	0
b	0
f	0
cc	4
nc	2
nop	1

<?php

declare(strict_types=1);

/*
 * This file is part of the Sonata Project package.
 *
 * (c) Thomas Rabaix <[email protected]>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Sonata\NotificationBundle\Backend;

use Enqueue\AmqpTools\DelayStrategy;
use Enqueue\AmqpTools\DelayStrategyAware;
use Guzzle\Http\Client as GuzzleClient;
use Interop\Amqp\AmqpConnectionFactory;
use Interop\Amqp\AmqpContext;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPConnection;
use Sonata\NotificationBundle\Exception\BackendNotFoundException;
use Sonata\NotificationBundle\Model\MessageInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use ZendDiagnostics\Result\Failure;
use ZendDiagnostics\Result\Success;

/**
 * Producer side of the rabbitmq backend.
 */
class AMQPBackendDispatcher extends QueueBackendDispatcher
{
    /**
     * @var array
     */
    protected $settings;

    /**
     * @deprecated since 3.2, will be removed in 4.x
     *
     * @var AMQPChannel
     */
    protected $channel;

    /**
     * @deprecated since 3.2, will be removed in 4.x
     *
     * @var AMQPConnection
     */
    protected $connection;

    protected $backendsInitialized = false;

    /**
     * @var AmqpConnectionFactory
     */
    private $connectionFactory;

    /**
     * @var AmqpContext
     */
    private $context;

    /**
     * @param string $defaultQueue
     */
    public function __construct(array $settings, array $queues, $defaultQueue, array $backends)
    {
        parent::__construct($queues, $defaultQueue, $backends);

        $this->settings = $settings;
    }

    /**
     * @deprecated since 3.2, will be removed in 4.x
     *
     * @return AMQPChannel
     */
    public function getChannel()
    {
        @trigger_error(sprintf('The method %s is deprecated since version 3.3 and will be removed in 4.0. Use %s::getContext() instead.', __METHOD__, __CLASS__), E_USER_DEPRECATED);
// For example instead of
@mkdir($dir);

// Better use
if (@mkdir($dir) === false) {
    throw new \RuntimeException('The directory '.$dir.' could not be created.');
}

        if (!$this->channel) {

            if (!$this->context instanceof \Enqueue\AmqpLib\AmqpContext) {
                throw new \LogicException('The BC layer works only if enqueue/amqp-lib lib is being used.');
            }

            // load context
            $this->getContext();

            /** @var \Enqueue\AmqpLib\AmqpContext $context */
            $context = $this->getContext();

            $this->channel = $context->getLibChannel();

            $this->connection = $this->channel->getConnection();

        }

        return $this->channel;

    }

    /**
     * @return AmqpContext
     */
    final public function getContext()
    {
        if (!$this->context) {
            if (!\array_key_exists('factory_class', $this->settings)) {
                throw new \LogicException('The factory_class option is missing though it is required.');
            }
            $factoryClass = $this->settings['factory_class'];
            if (
                !class_exists($factoryClass) ||
                !(new \ReflectionClass($factoryClass))->implementsInterface(AmqpConnectionFactory::class)
            ) {
                throw new \LogicException(sprintf(
                    'The factory_class option "%s" has to be valid class that implements "%s"',
                    $factoryClass,
                    AmqpConnectionFactory::class
                ));
            }

            /* @var AmqpConnectionFactory $factory */
            $this->connectionFactory = $factory = new $factoryClass([
                'host' => $this->settings['host'],
                'port' => $this->settings['port'],
                'user' => $this->settings['user'],
                'pass' => $this->settings['pass'],
                'vhost' => $this->settings['vhost'],
            ]);

            if ($factory instanceof DelayStrategyAware) {
                $factory = $this->resolveDelayStrategy($factory);
            }

            $this->context = $factory->createContext();

            register_shutdown_function([$this, 'shutdown']);
        }

        return $this->context;
    }

    /**
     * {@inheritdoc}
     */
    public function getBackend($type)
    {
        if (!$this->backendsInitialized) {
            foreach ($this->backends as $backend) {
                $backend['backend']->initialize();
            }
            $this->backendsInitialized = true;
        }

        $default = null;

        if (0 === \count($this->queues)) {
            foreach ($this->backends as $backend) {
                if ('default' === $backend['type']) {
                    return $backend['backend'];
                }
            }
        }

        foreach ($this->backends as $backend) {
            if ('all' === $type && '' === $backend['type']) {
                return $backend['backend'];
            }

            if ($backend['type'] === $type) {
                return $backend['backend'];
            }

            if ($backend['type'] === $this->defaultQueue) {
                $default = $backend['backend'];
            }
        }

        if (null === $default) {
            throw new BackendNotFoundException('Could not find a message backend for the type '.$type);
        }

        return $default;
    }

    /**
     * {@inheritdoc}
     */
    public function getIterator()
    {
        throw new \RuntimeException(
            'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
        );
    }

    /**
     * {@inheritdoc}
     */
    public function handle(MessageInterface $message, EventDispatcherInterface $dispatcher)
    {
        throw new \RuntimeException(
            'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
        );
    }

    /**
     * {@inheritdoc}
     */
    public function getStatus()
    {
        try {
            $this->getContext();
            $output = $this->getApiQueueStatus();
            $checked = 0;
            $missingConsumers = [];

            foreach ($this->queues as $queue) {
                foreach ($output as $q) {
                    if ($q['name'] === $queue['queue']) {
                        ++$checked;
                        if (0 === $q['consumers']) {
                            $missingConsumers[] = $queue['queue'];
                        }
                    }
                }
            }

            if ($checked !== \count($this->queues)) {
                return new Failure(
                    'Not all queues for the available notification types registered in the rabbitmq broker. '
                    .'Are the consumer commands running?'
                );
            }

            if (\count($missingConsumers) > 0) {
                return new Failure(
                    'There are no rabbitmq consumers running for the queues: '.implode(', ', $missingConsumers)
                );
            }
        } catch (\Exception $e) {
            return new Failure($e->getMessage());
        }

        return new Success('Channel is running (RabbitMQ) and consumers for all queues available.');
    }

    /**
     * {@inheritdoc}
     */
    public function cleanup()
    {
        throw new \RuntimeException(
            'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
        );
    }

    public function shutdown()
    {
        if ($this->context) {
            $this->context->close();
        }
    }

    /**
     * {@inheritdoc}
     */
    public function initialize()
    {
    }

    /**
     * Calls the rabbitmq management api /api/<vhost>/queues endpoint to list the available queues.
     *
     * @see http://hg.rabbitmq.com/rabbitmq-management/raw-file/3646dee55e02/priv/www-api/help.html
     *
     * @return array
     */
    protected function getApiQueueStatus()
    {
        if (!class_exists(GuzzleClient::class)) {
            throw new \RuntimeException(
                'The guzzle http client library is required to run rabbitmq health checks. '
                .'Make sure to add guzzlehttp/guzzle to your composer.json.'
            );
        }

        $client = new GuzzleClient();
        $client->setConfig(['curl.options' => [CURLOPT_CONNECTTIMEOUT_MS => 3000]]);
        $request = $client->get(sprintf('%s/queues', $this->settings['console_url']));
        $request->setAuth($this->settings['user'], $this->settings['pass']);

        return json_decode($request->send()->getBody(true), true);
    }

    protected function resolveDelayStrategy(DelayStrategyAware $factory): DelayStrategyAware
    {
        $delayStrategy = $this->settings['delay_strategy_class'];
        if (
            !empty($delayStrategy) &&
            (
                !class_exists($delayStrategy) ||
                !(new \ReflectionClass($delayStrategy))->implementsInterface(DelayStrategy::class)
            )
        ) {
            throw new \LogicException(sprintf(
                'The delay_strategy_class option "%s" has to be a valid class that implements "%s" or null to ignore',
                $delayStrategy,
                DelayStrategy::class
            ));
        }

        return $factory->setDelayStrategy(new $delayStrategy());
    }
}


1			<?php
2
3			declare(strict_types=1);
4
5			/*
6			* This file is part of the Sonata Project package.
7			*
8			* (c) Thomas Rabaix <[email protected]>
9			*
10			* For the full copyright and license information, please view the LICENSE
11			* file that was distributed with this source code.
12			*/
13
14			namespace Sonata\NotificationBundle\Backend;
15
16			use Enqueue\AmqpTools\DelayStrategy;
17			use Enqueue\AmqpTools\DelayStrategyAware;
18			use Guzzle\Http\Client as GuzzleClient;
19			use Interop\Amqp\AmqpConnectionFactory;
20			use Interop\Amqp\AmqpContext;
21			use PhpAmqpLib\Channel\AMQPChannel;
22			use PhpAmqpLib\Connection\AMQPConnection;
23			use Sonata\NotificationBundle\Exception\BackendNotFoundException;
24			use Sonata\NotificationBundle\Model\MessageInterface;
25			use Symfony\Component\EventDispatcher\EventDispatcherInterface;
26			use ZendDiagnostics\Result\Failure;
27			use ZendDiagnostics\Result\Success;
28
29			/**
30			* Producer side of the rabbitmq backend.
31			*/
32			class AMQPBackendDispatcher extends QueueBackendDispatcher
33			{
34			/**
35			* @var array
36			*/
37			protected $settings;
38
39			/**
40			* @deprecated since 3.2, will be removed in 4.x
41			*
42			* @var AMQPChannel
43			*/
44			protected $channel;
45
46			/**
47			* @deprecated since 3.2, will be removed in 4.x
48			*
49			* @var AMQPConnection
50			*/
51			protected $connection;
52
53			protected $backendsInitialized = false;
54
55			/**
56			* @var AmqpConnectionFactory
57			*/
58			private $connectionFactory;
59
60			/**
61			* @var AmqpContext
62			*/
63			private $context;
64
65			/**
66			* @param string $defaultQueue
67			*/
68			public function __construct(array $settings, array $queues, $defaultQueue, array $backends)
69			{
70			parent::__construct($queues, $defaultQueue, $backends);
71
72			$this->settings = $settings;
73			}
74
75			/**
76			* @deprecated since 3.2, will be removed in 4.x
77			*
78			* @return AMQPChannel
79			*/
80			public function getChannel()
81			{
82			@trigger_error(sprintf('The method %s is deprecated since version 3.3 and will be removed in 4.0. Use %s::getContext() instead.', __METHOD__, __CLASS__), E_USER_DEPRECATED);
			0 ignored issues – show Security Best Practice introduced 2017-11-13 13:25 UTC by Report Bug Copy Issue Report It seems like you do not handle an error condition here. This can introduce security issues, and is generally not recommended. If you suppress an error, we recommend checking for the error condition explicitly: // For example instead of @mkdir($dir); // Better use if (@mkdir($dir) === false) { throw new \RuntimeException('The directory '.$dir.' could not be created.'); } Loading history...
83
84			if (!$this->channel) {
			0 ignored issues – show Deprecated Code introduced 2017-11-13 13:25 UTC by Report Bug Copy Issue Report The property `Sonata\NotificationBundl...endDispatcher::$channel` has been deprecated with message: since 3.2, will be removed in 4.x This property has been deprecated. The supplier of the class has supplied an explanatory message. The explanatory message should give you some clue as to whether and when the property will be removed from the class and what other property to use instead. Loading history...
85			if (!$this->context instanceof \Enqueue\AmqpLib\AmqpContext) {
86			throw new \LogicException('The BC layer works only if enqueue/amqp-lib lib is being used.');
87			}
88
89			// load context
90			$this->getContext();
91
92			/** @var \Enqueue\AmqpLib\AmqpContext $context */
93			$context = $this->getContext();
94
95			$this->channel = $context->getLibChannel();
			0 ignored issues – show Deprecated Code introduced 2017-11-13 13:25 UTC by Report Bug Copy Issue Report The property `Sonata\NotificationBundl...endDispatcher::$channel` has been deprecated with message: since 3.2, will be removed in 4.x This property has been deprecated. The supplier of the class has supplied an explanatory message. The explanatory message should give you some clue as to whether and when the property will be removed from the class and what other property to use instead. Loading history...
96			$this->connection = $this->channel->getConnection();
			0 ignored issues – show Deprecated Code introduced 2017-11-13 13:25 UTC by Report Bug Copy Issue Report The property `Sonata\NotificationBundl...Dispatcher::$connection` has been deprecated with message: since 3.2, will be removed in 4.x This property has been deprecated. The supplier of the class has supplied an explanatory message. The explanatory message should give you some clue as to whether and when the property will be removed from the class and what other property to use instead. Loading history... Deprecated Code introduced 2017-11-13 13:25 UTC by Report Bug Copy Issue Report The property `Sonata\NotificationBundl...endDispatcher::$channel` has been deprecated with message: since 3.2, will be removed in 4.x This property has been deprecated. The supplier of the class has supplied an explanatory message. The explanatory message should give you some clue as to whether and when the property will be removed from the class and what other property to use instead. Loading history...
97			}
98
99			return $this->channel;
			0 ignored issues – show Deprecated Code introduced 2017-11-13 13:25 UTC by Report Bug Copy Issue Report The property `Sonata\NotificationBundl...endDispatcher::$channel` has been deprecated with message: since 3.2, will be removed in 4.x This property has been deprecated. The supplier of the class has supplied an explanatory message. The explanatory message should give you some clue as to whether and when the property will be removed from the class and what other property to use instead. Loading history...
100			}
101
102			/**
103			* @return AmqpContext
104			*/
105			final public function getContext()
106			{
107			if (!$this->context) {
108			if (!\array_key_exists('factory_class', $this->settings)) {
109			throw new \LogicException('The factory_class option is missing though it is required.');
110			}
111			$factoryClass = $this->settings['factory_class'];
112			if (
113			!class_exists($factoryClass) \|\|
114			!(new \ReflectionClass($factoryClass))->implementsInterface(AmqpConnectionFactory::class)
115			) {
116			throw new \LogicException(sprintf(
117			'The factory_class option "%s" has to be valid class that implements "%s"',
118			$factoryClass,
119			AmqpConnectionFactory::class
120			));
121			}
122
123			/* @var AmqpConnectionFactory $factory */
124			$this->connectionFactory = $factory = new $factoryClass([
125			'host' => $this->settings['host'],
126			'port' => $this->settings['port'],
127			'user' => $this->settings['user'],
128			'pass' => $this->settings['pass'],
129			'vhost' => $this->settings['vhost'],
130			]);
131
132			if ($factory instanceof DelayStrategyAware) {
133			$factory = $this->resolveDelayStrategy($factory);
134			}
135
136			$this->context = $factory->createContext();
137
138			register_shutdown_function([$this, 'shutdown']);
139			}
140
141			return $this->context;
142			}
143
144			/**
145			* {@inheritdoc}
146			*/
147			public function getBackend($type)
148			{
149			if (!$this->backendsInitialized) {
150			foreach ($this->backends as $backend) {
151			$backend['backend']->initialize();
152			}
153			$this->backendsInitialized = true;
154			}
155
156			$default = null;
157
158			if (0 === \count($this->queues)) {
159			foreach ($this->backends as $backend) {
160			if ('default' === $backend['type']) {
161			return $backend['backend'];
162			}
163			}
164			}
165
166			foreach ($this->backends as $backend) {
167			if ('all' === $type && '' === $backend['type']) {
168			return $backend['backend'];
169			}
170
171			if ($backend['type'] === $type) {
172			return $backend['backend'];
173			}
174
175			if ($backend['type'] === $this->defaultQueue) {
176			$default = $backend['backend'];
177			}
178			}
179
180			if (null === $default) {
181			throw new BackendNotFoundException('Could not find a message backend for the type '.$type);
182			}
183
184			return $default;
185			}
186
187			/**
188			* {@inheritdoc}
189			*/
190			public function getIterator()
191			{
192			throw new \RuntimeException(
193			'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
194			);
195			}
196
197			/**
198			* {@inheritdoc}
199			*/
200			public function handle(MessageInterface $message, EventDispatcherInterface $dispatcher)
201			{
202			throw new \RuntimeException(
203			'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
204			);
205			}
206
207			/**
208			* {@inheritdoc}
209			*/
210			public function getStatus()
211			{
212			try {
213			$this->getContext();
214			$output = $this->getApiQueueStatus();
215			$checked = 0;
216			$missingConsumers = [];
217
218			foreach ($this->queues as $queue) {
219			foreach ($output as $q) {
220			if ($q['name'] === $queue['queue']) {
221			++$checked;
222			if (0 === $q['consumers']) {
223			$missingConsumers[] = $queue['queue'];
224			}
225			}
226			}
227			}
228
229			if ($checked !== \count($this->queues)) {
230			return new Failure(
231			'Not all queues for the available notification types registered in the rabbitmq broker. '
232			.'Are the consumer commands running?'
233			);
234			}
235
236			if (\count($missingConsumers) > 0) {
237			return new Failure(
238			'There are no rabbitmq consumers running for the queues: '.implode(', ', $missingConsumers)
239			);
240			}
241			} catch (\Exception $e) {
242			return new Failure($e->getMessage());
243			}
244
245			return new Success('Channel is running (RabbitMQ) and consumers for all queues available.');
246			}
247
248			/**
249			* {@inheritdoc}
250			*/
251			public function cleanup()
252			{
253			throw new \RuntimeException(
254			'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
255			);
256			}
257
258			public function shutdown()
259			{
260			if ($this->context) {
261			$this->context->close();
262			}
263			}
264
265			/**
266			* {@inheritdoc}
267			*/
268			public function initialize()
269			{
270			}
271
272			/**
273			* Calls the rabbitmq management api /api/<vhost>/queues endpoint to list the available queues.
274			*
275			* @see http://hg.rabbitmq.com/rabbitmq-management/raw-file/3646dee55e02/priv/www-api/help.html
276			*
277			* @return array
278			*/
279			protected function getApiQueueStatus()
280			{
281			if (!class_exists(GuzzleClient::class)) {
282			throw new \RuntimeException(
283			'The guzzle http client library is required to run rabbitmq health checks. '
284			.'Make sure to add guzzlehttp/guzzle to your composer.json.'
285			);
286			}
287
288			$client = new GuzzleClient();
289			$client->setConfig(['curl.options' => [CURLOPT_CONNECTTIMEOUT_MS => 3000]]);
290			$request = $client->get(sprintf('%s/queues', $this->settings['console_url']));
291			$request->setAuth($this->settings['user'], $this->settings['pass']);
292
293			return json_decode($request->send()->getBody(true), true);
294			}
295
296			protected function resolveDelayStrategy(DelayStrategyAware $factory): DelayStrategyAware
297			{
298			$delayStrategy = $this->settings['delay_strategy_class'];
299			if (
300			!empty($delayStrategy) &&
301			(
302			!class_exists($delayStrategy) \|\|
303			!(new \ReflectionClass($delayStrategy))->implementsInterface(DelayStrategy::class)
304			)
305			) {
306			throw new \LogicException(sprintf(
307			'The delay_strategy_class option "%s" has to be a valid class that implements "%s" or null to ignore',
308			$delayStrategy,
309			DelayStrategy::class
310			));
311			}
312
313			return $factory->setDelayStrategy(new $delayStrategy());
314			}
315			}
316

sonata-project / SonataNotificationBundle

Pull Request — 3.x (#387)

AMQPBackendDispatcher::resolveDelayStrategy() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like