Completed
Pull Request — master (#276)
by Maksim
04:34
created

AMQPBackendDispatcher::getContext()   B

Complexity

Conditions 4
Paths 4

Size

Total Lines 26
Code Lines 15

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
dl 0
loc 26
rs 8.5806
c 0
b 0
f 0
cc 4
eloc 15
nc 4
nop 0
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\AmqpContext;
18
use Interop\Amqp\AmqpConnectionFactory;
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
     * @var AmqpConnectionFactory
39
     */
40
    protected $connectionFactory;
41
42
    /**
43
     * @deprecated since 3.2, will be removed in 4.x
44
     *
45
     * @var AMQPChannel
46
     */
47
    protected $channel;
48
49
    /**
50
     * @deprecated since 3.2, will be removed in 4.x
51
     *
52
     * @var AMQPConnection
53
     */
54
    protected $connection;
55
56
    /**
57
     * @var AmqpContext
58
     */
59
    protected $context;
60
61
    protected $backendsInitialized = false;
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
     * @return AmqpContext
78
     */
79
    public function getContext()
80
    {
81
        if (!$this->context) {
82
            if (!class_exists(\Enqueue\AmqpLib\AmqpConnectionFactory::class)) {
83
                throw new \RuntimeException('Please install enqueue/amqp-lib:^0.8 as a dependency');
84
            }
85
            /** @var AmqpConnectionFactory $factory */
86
            $factory = new \Enqueue\AmqpLib\AmqpConnectionFactory([
87
                'host' => $this->settings['host'],
88
                'port' => $this->settings['port'],
89
                'user' => $this->settings['user'],
90
                'pass' => $this->settings['pass'],
91
                'vhost' => $this->settings['vhost'],
92
            ]);
93
94
            if ($factory instanceof DelayStrategyAware) {
95
                $factory->setDelayStrategy(new RabbitMqDlxDelayStrategy());
96
            }
97
98
            $this->context = $this->connectionFactory->createContext();
99
100
            register_shutdown_function([$this, 'shutdown']);
101
        }
102
103
        return $this->context;
104
    }
105
106
    /**
107
     * {@inheritdoc}
108
     */
109
    public function getBackend($type)
110
    {
111
        if (!$this->backendsInitialized) {
112
            foreach ($this->backends as $backend) {
113
                $backend['backend']->initialize();
114
            }
115
            $this->backendsInitialized = true;
116
        }
117
118
        $default = null;
119
120
        if (0 === count($this->queues)) {
121
            foreach ($this->backends as $backend) {
122
                if ('default' === $backend['type']) {
123
                    return $backend['backend'];
124
                }
125
            }
126
        }
127
128
        foreach ($this->backends as $backend) {
129
            if ('all' === $type && '' === $backend['type']) {
130
                return $backend['backend'];
131
            }
132
133
            if ($backend['type'] === $type) {
134
                return $backend['backend'];
135
            }
136
137
            if ($backend['type'] === $this->defaultQueue) {
138
                $default = $backend['backend'];
139
            }
140
        }
141
142
        if (null === $default) {
143
            throw new BackendNotFoundException('Could not find a message backend for the type '.$type);
144
        }
145
146
        return $default;
147
    }
148
149
    /**
150
     * {@inheritdoc}
151
     */
152
    public function getIterator()
153
    {
154
        throw new \RuntimeException(
155
            'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
156
        );
157
    }
158
159
    /**
160
     * {@inheritdoc}
161
     */
162
    public function handle(MessageInterface $message, EventDispatcherInterface $dispatcher)
163
    {
164
        throw new \RuntimeException(
165
            'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
166
        );
167
    }
168
169
    /**
170
     * {@inheritdoc}
171
     */
172
    public function getStatus()
173
    {
174
        try {
175
            $this->getContext();
176
            $output = $this->getApiQueueStatus();
177
            $checked = 0;
178
            $missingConsumers = [];
179
180
            foreach ($this->queues as $queue) {
181
                foreach ($output as $q) {
182
                    if ($q['name'] === $queue['queue']) {
183
                        ++$checked;
184
                        if (0 === $q['consumers']) {
185
                            $missingConsumers[] = $queue['queue'];
186
                        }
187
                    }
188
                }
189
            }
190
191
            if ($checked !== count($this->queues)) {
192
                return new Failure(
193
                    'Not all queues for the available notification types registered in the rabbitmq broker. '
194
                    .'Are the consumer commands running?'
195
                );
196
            }
197
198
            if (count($missingConsumers) > 0) {
199
                return new Failure(
200
                    'There are no rabbitmq consumers running for the queues: '.implode(', ', $missingConsumers)
201
                );
202
            }
203
        } catch (\Exception $e) {
204
            return new Failure($e->getMessage());
205
        }
206
207
        return new Success('Channel is running (RabbitMQ) and consumers for all queues available.');
208
    }
209
210
    /**
211
     * {@inheritdoc}
212
     */
213
    public function cleanup()
214
    {
215
        throw new \RuntimeException(
216
            'You need to use a specific rabbitmq backend supporting the selected queue to run a consumer.'
217
        );
218
    }
219
220
    public function shutdown()
221
    {
222
        if ($this->context) {
223
            $this->context->close();
224
        }
225
    }
226
227
    /**
228
     * {@inheritdoc}
229
     */
230
    public function initialize()
231
    {
232
    }
233
234
    /**
235
     * @deprecated since 3.2, will be removed in 4.x
236
     *
237
     * @return AMQPChannel
238
     */
239
    public function getChannel()
240
    {
241
        @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 by
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...
242
243
        if (!$this->channel) {
0 ignored issues
show
Deprecated Code introduced by
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...
244
            // load context
245
            $this->getContext();
246
247
            /** @var \Enqueue\AmqpLib\AmqpContext $context */
248
            $context = $this->getContext();
249
250
            $this->channel = $context->getLibChannel();
0 ignored issues
show
Deprecated Code introduced by
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...
251
            $this->connection = $this->channel->getConnection();
0 ignored issues
show
Deprecated Code introduced by
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 by
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...
252
        }
253
254
        return $this->channel;
0 ignored issues
show
Deprecated Code introduced by
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...
255
    }
256
257
    /**
258
     * Calls the rabbitmq management api /api/<vhost>/queues endpoint to list the available queues.
259
     *
260
     * @see http://hg.rabbitmq.com/rabbitmq-management/raw-file/3646dee55e02/priv/www-api/help.html
261
     *
262
     * @return array
263
     */
264
    protected function getApiQueueStatus()
265
    {
266
        if (!class_exists(GuzzleClient::class)) {
267
            throw new \RuntimeException(
268
                'The guzzle http client library is required to run rabbitmq health checks. '
269
                .'Make sure to add guzzlehttp/guzzle to your composer.json.'
270
            );
271
        }
272
273
        $client = new GuzzleClient();
274
        $client->setConfig(['curl.options' => [CURLOPT_CONNECTTIMEOUT_MS => 3000]]);
275
        $request = $client->get(sprintf('%s/queues', $this->settings['console_url']));
276
        $request->setAuth($this->settings['user'], $this->settings['pass']);
277
278
        return json_decode($request->send()->getBody(true), true);
279
    }
280
}
281