AMQPBackendDispatcher::getContext() - Code Metrics - Inspection of "Migrate to AMQP interop." - sonata-project/SonataNotificationBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — 3.x (#276)

by Maksim

created 2017-11-16 14:16 UTC

AMQPBackendDispatcher::getContext() B

↳ Parent: AMQPBackendDispatcher

Complexity

Conditions	6
Paths	5

Size

Total Lines	38
Code Lines	23

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	38
rs	8.439
c	0
b	0
f	0
cc	6
eloc	23
nc	5
nop	0

<?php

/*
 * 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\DelayStrategyAware;
use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
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 array  $settings
     * @param array  $queues
     * @param string $defaultQueue
     * @param array  $backends
     */
    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->setDelayStrategy(new RabbitMqDlxDelayStrategy());
            }

            $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);
    }
}


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

sonata-project / SonataNotificationBundle

Pull Request — 3.x (#276)

AMQPBackendDispatcher::getContext() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like