PeclAmqpDriver::getChannel() - Code Metrics - Inspection of "fix: With PECL driver, consumer is not cancelled" - Evaneos/Burrow - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#30)

by Pol-Valentin

created 2019-07-17 09:02 UTC

PeclAmqpDriver::getChannel() A

↳ Parent: PeclAmqpDriver

Complexity

Conditions	2
Paths	2

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
nc	2
nop	0
dl	0
loc	9
rs	9.9666
c	0
b	0
f	0

<?php

namespace Burrow\Driver;

use Assert\AssertionFailedException;
use Burrow\Driver;
use Burrow\Exception\ConsumerException;
use Burrow\Exception\TimeoutException;
use Burrow\Message;

/**
 * @codeCoverageIgnore
 */
class PeclAmqpDriver implements Driver
{
    const DELIVERY_MODE = 'delivery_mode';
    const CONTENT_TYPE = 'content_type';
    const APPLICATION_HEADERS = 'headers';
    const CORRELATION_ID = 'correlation_id';
    const REPLY_TO = 'reply_to';

    /** @var \AMQPConnection */
    private $connection;

    /** @var \AMQPChannel */
    private $channel;

    /**
     * PeclAmqpDriver constructor.
     *
     * @param \AMQPConnection $connection
     */
    public function __construct(\AMQPConnection $connection)
    {
        $this->connection = $connection;
    }

    /**
     * Declare a persistent queue
     *
     * @param string $queueName
     * @param string $type
     *
     * @return string
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPQueueException
     */
    public function declareSimpleQueue($queueName = '', $type = self::QUEUE_DURABLE)
    {
        $flag = AMQP_DURABLE;
        if ($type === self::QUEUE_EXCLUSIVE) {
            $flag = AMQP_EXCLUSIVE;
        }

        $queue = $this->getQueue($queueName);
        $queue->setFlags($flag);
        $queue->declareQueue();

        return $queue->getName();
    }

    /**
     * Declare an exchange
     *
     * @param string $exchangeName
     * @param string $type
     *
     * @return string
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPExchangeException
     */
    public function declareExchange($exchangeName = '', $type = self::EXCHANGE_TYPE_FANOUT)
    {
        $exchange = $this->getExchange($exchangeName);
        $exchange->setType($type);
        $exchange->setFlags(AMQP_DURABLE);
        $exchange->declareExchange();

        return $exchange->getName();
    }

    /**
     * Bind an existing queue to an exchange
     *
     * @param string $exchange
     * @param string $queueName
     * @param string $routingKey
     *
     * @return void
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPQueueException
     */
    public function bindQueue($exchange, $queueName, $routingKey = '')
    {
        $queue = $this->getQueue($queueName);
        $queue->bind($exchange, $routingKey);
    }

    /**
     * Create a persisting queue and bind it to an exchange
     *
     * @param string $exchange
     * @param string $queueName
     * @param string $routingKey
     *
     * @return void
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPQueueException
     */
    public function declareAndBindQueue($exchange, $queueName, $routingKey = '')
    {
        $this->declareSimpleQueue($queueName);
        $this->bindQueue($exchange, $queueName, $routingKey);
    }

    /**
     * Delete a queue
     *
     * @param string $queueName
     *
     * @return void
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPQueueException
     */
    public function deleteQueue($queueName)
    {
        $this->getQueue($queueName)->delete();
    }

    /**
     * Delete an exchange
     *
     * @param string $exchangeName
     *
     * @return void
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPExchangeException
     */
    public function deleteExchange($exchangeName)
    {
        $this->getExchange($exchangeName)->delete();
    }

    /**
     * Publish a message in the exchange
     *
     * @param string $exchangeName
     * @param Message $message
     *
     * @return void
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPExchangeException
     */
    public function publish($exchangeName, Message $message)
    {
        $exchange = $this->getExchange($exchangeName);
        $exchange->publish(
            $message->getBody(),
            $message->getRoutingKey(),
            AMQP_NOPARAM,
            self::getMessageProperties($message)
        );
    }

    /**
     * Consume the queue
     *
     * @param string $queueName
     * @param callable $callback Must return false if you want to consume only one message
     * @param int $timeout
     * @param bool $autoAck
     *
     * @return void
     *
     * @throws ConsumerException
     * @throws TimeoutException
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPQueueException
     * @throws \InvalidArgumentException
     * @throws AssertionFailedException
     */
    public function consume($queueName, callable $callback, $timeout = 0, $autoAck = true)
    {
        $this->connection->setReadTimeout($timeout);
        $this->getChannel()->setPrefetchCount(1);
        $queue = $this->getQueue($queueName);
        $flags = $autoAck ? AMQP_AUTOACK : AMQP_NOPARAM;
        $consumerTag = self::generateConsumerTag();
        try {
            $queue->consume(function (\AMQPEnvelope $message) use ($callback, $queueName) {

                $burrowMessage = new Message(
                    $message->getBody(),
                    $message->getRoutingKey(),
                    $message->getHeaders(),
                    $message->getCorrelationId(),
                    $message->getReplyTo()
                );
                $burrowMessage->setDeliveryTag($message->getDeliveryTag());
                $burrowMessage->setQueue($queueName);

                return $callback($burrowMessage);
            }, $flags, $consumerTag);
        } catch (\AMQPQueueException $e) {
            if ($e->getMessage() === 'Consumer timeout exceed') {
                throw TimeoutException::build($e, $timeout);
            }
            throw ConsumerException::build($e);
        } finally {
            $queue->cancel($consumerTag);
        }
    }

    /**
     * Acknowledge the reception of the message.
     *
     * @param Message $message
     *
     * @return void
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPQueueException
     */
    public function ack(Message $message)
    {
        $queue = $this->getQueue($message->getQueue());
        $queue->ack($message->getDeliveryTag());
    }

    /**
     * Acknowledge an error during the consumption of the message
     *
     * @param Message $message
     * @param bool $requeue
     *
     * @return void
     *
     * @throws \AMQPConnectionException
     * @throws \AMQPChannelException
     * @throws \AMQPQueueException
     */
    public function nack(Message $message, $requeue = true)
    {
        $queue = $this->getQueue($message->getQueue());
        $queue->nack($message->getDeliveryTag(), $requeue ? AMQP_REQUEUE : AMQP_NOPARAM);
    }

    /**
     * Close the connection
     *
     * @return void
     */
    public function close()
    {
        $this->connection->disconnect();
    }

    /**
     * @return \AMQPChannel
     *
     * @throws \AMQPConnectionException
     */
    private function getChannel()
    {
        if (null === $this->channel) {
            $this->connection->connect();
            $this->channel = new \AMQPChannel($this->connection);
        }

        return $this->channel;
    }

    /**
     * @param string $queueName
     *
     * @return \AMQPQueue
     *
     * @throws \AMQPQueueException
     * @throws \AMQPConnectionException
     */
    private function getQueue($queueName)
    {
        $queue = new \AMQPQueue($this->getChannel());
        if ($queueName) {
            $queue->setName($queueName);
        }

        return $queue;
    }

    /**
     * @param string $exchangeName
     *
     * @return \AMQPExchange
     *
     * @throws \AMQPExchangeException
     * @throws \AMQPConnectionException
     */
    private function getExchange($exchangeName)
    {
        $exchange = new \AMQPExchange($this->getChannel());
        if ($exchangeName) {
            $exchange->setName($exchangeName);
        }

        return $exchange;
    }

    /**
     * Returns the message parameters
     *
     * @param Message $message
     *
     * @return array
     */
    private static function getMessageProperties(Message $message)
    {
        $properties = [
            self::DELIVERY_MODE => 2,
            self::CONTENT_TYPE => 'text/plain',
            self::APPLICATION_HEADERS => $message->getHeaders(),
        ];

        if ($message->getCorrelationId() !== null) {
            $properties[self::CORRELATION_ID] = $message->getCorrelationId();
        }

        if ($message->getReplyTo() !== null) {
            $properties[self::REPLY_TO] = $message->getReplyTo();
        }

        return $properties;
    }

    private static function generateConsumerTag(
        $prefix = 'consumer-tags',
        $length = 10,
        $chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    ) {
        return $prefix . substr(str_shuffle(str_repeat($chars, ceil($length / strlen($chars)))), 1, $length);
    }
}


1			<?php
2
3			namespace Burrow\Driver;
4
5			use Assert\AssertionFailedException;
6			use Burrow\Driver;
7			use Burrow\Exception\ConsumerException;
8			use Burrow\Exception\TimeoutException;
9			use Burrow\Message;
10
11			/**
12			* @codeCoverageIgnore
13			*/
14			class PeclAmqpDriver implements Driver
15			{
16			const DELIVERY_MODE = 'delivery_mode';
17			const CONTENT_TYPE = 'content_type';
18			const APPLICATION_HEADERS = 'headers';
19			const CORRELATION_ID = 'correlation_id';
20			const REPLY_TO = 'reply_to';
21
22			/** @var \AMQPConnection */
23			private $connection;
24
25			/** @var \AMQPChannel */
26			private $channel;
27
28			/**
29			* PeclAmqpDriver constructor.
30			*
31			* @param \AMQPConnection $connection
32			*/
33			public function __construct(\AMQPConnection $connection)
34			{
35			$this->connection = $connection;
36			}
37
38			/**
39			* Declare a persistent queue
40			*
41			* @param string $queueName
42			* @param string $type
43			*
44			* @return string
45			*
46			* @throws \AMQPConnectionException
47			* @throws \AMQPChannelException
48			* @throws \AMQPQueueException
49			*/
50			public function declareSimpleQueue($queueName = '', $type = self::QUEUE_DURABLE)
51			{
52			$flag = AMQP_DURABLE;
53			if ($type === self::QUEUE_EXCLUSIVE) {
54			$flag = AMQP_EXCLUSIVE;
55			}
56
57			$queue = $this->getQueue($queueName);
58			$queue->setFlags($flag);
59			$queue->declareQueue();
60
61			return $queue->getName();
62			}
63
64			/**
65			* Declare an exchange
66			*
67			* @param string $exchangeName
68			* @param string $type
69			*
70			* @return string
71			*
72			* @throws \AMQPConnectionException
73			* @throws \AMQPChannelException
74			* @throws \AMQPExchangeException
75			*/
76			public function declareExchange($exchangeName = '', $type = self::EXCHANGE_TYPE_FANOUT)
77			{
78			$exchange = $this->getExchange($exchangeName);
79			$exchange->setType($type);
80			$exchange->setFlags(AMQP_DURABLE);
81			$exchange->declareExchange();
82
83			return $exchange->getName();
84			}
85
86			/**
87			* Bind an existing queue to an exchange
88			*
89			* @param string $exchange
90			* @param string $queueName
91			* @param string $routingKey
92			*
93			* @return void
94			*
95			* @throws \AMQPConnectionException
96			* @throws \AMQPChannelException
97			* @throws \AMQPQueueException
98			*/
99			public function bindQueue($exchange, $queueName, $routingKey = '')
100			{
101			$queue = $this->getQueue($queueName);
102			$queue->bind($exchange, $routingKey);
103			}
104
105			/**
106			* Create a persisting queue and bind it to an exchange
107			*
108			* @param string $exchange
109			* @param string $queueName
110			* @param string $routingKey
111			*
112			* @return void
113			*
114			* @throws \AMQPConnectionException
115			* @throws \AMQPChannelException
116			* @throws \AMQPQueueException
117			*/
118			public function declareAndBindQueue($exchange, $queueName, $routingKey = '')
119			{
120			$this->declareSimpleQueue($queueName);
121			$this->bindQueue($exchange, $queueName, $routingKey);
122			}
123
124			/**
125			* Delete a queue
126			*
127			* @param string $queueName
128			*
129			* @return void
130			*
131			* @throws \AMQPConnectionException
132			* @throws \AMQPChannelException
133			* @throws \AMQPQueueException
134			*/
135			public function deleteQueue($queueName)
136			{
137			$this->getQueue($queueName)->delete();
138			}
139
140			/**
141			* Delete an exchange
142			*
143			* @param string $exchangeName
144			*
145			* @return void
146			*
147			* @throws \AMQPConnectionException
148			* @throws \AMQPChannelException
149			* @throws \AMQPExchangeException
150			*/
151			public function deleteExchange($exchangeName)
152			{
153			$this->getExchange($exchangeName)->delete();
154			}
155
156			/**
157			* Publish a message in the exchange
158			*
159			* @param string $exchangeName
160			* @param Message $message
161			*
162			* @return void
163			*
164			* @throws \AMQPConnectionException
165			* @throws \AMQPChannelException
166			* @throws \AMQPExchangeException
167			*/
168			public function publish($exchangeName, Message $message)
169			{
170			$exchange = $this->getExchange($exchangeName);
171			$exchange->publish(
172			$message->getBody(),
173			$message->getRoutingKey(),
174			AMQP_NOPARAM,
175			self::getMessageProperties($message)
176			);
177			}
178
179			/**
180			* Consume the queue
181			*
182			* @param string $queueName
183			* @param callable $callback Must return false if you want to consume only one message
184			* @param int $timeout
185			* @param bool $autoAck
186			*
187			* @return void
188			*
189			* @throws ConsumerException
190			* @throws TimeoutException
191			* @throws \AMQPConnectionException
192			* @throws \AMQPChannelException
193			* @throws \AMQPQueueException
194			* @throws \InvalidArgumentException
195			* @throws AssertionFailedException
196			*/
197			public function consume($queueName, callable $callback, $timeout = 0, $autoAck = true)
198			{
199			$this->connection->setReadTimeout($timeout);
200			$this->getChannel()->setPrefetchCount(1);
201			$queue = $this->getQueue($queueName);
202			$flags = $autoAck ? AMQP_AUTOACK : AMQP_NOPARAM;
203			$consumerTag = self::generateConsumerTag();
204			try {
205			$queue->consume(function (\AMQPEnvelope $message) use ($callback, $queueName) {
206
207			$burrowMessage = new Message(
208			$message->getBody(),
209			$message->getRoutingKey(),
210			$message->getHeaders(),
211			$message->getCorrelationId(),
212			$message->getReplyTo()
213			);
214			$burrowMessage->setDeliveryTag($message->getDeliveryTag());
215			$burrowMessage->setQueue($queueName);
216
217			return $callback($burrowMessage);
218			}, $flags, $consumerTag);
219			} catch (\AMQPQueueException $e) {
220			if ($e->getMessage() === 'Consumer timeout exceed') {
221			throw TimeoutException::build($e, $timeout);
222			}
223			throw ConsumerException::build($e);
224			} finally {
225			$queue->cancel($consumerTag);
226			}
227			}
228
229			/**
230			* Acknowledge the reception of the message.
231			*
232			* @param Message $message
233			*
234			* @return void
235			*
236			* @throws \AMQPConnectionException
237			* @throws \AMQPChannelException
238			* @throws \AMQPQueueException
239			*/
240			public function ack(Message $message)
241			{
242			$queue = $this->getQueue($message->getQueue());
243			$queue->ack($message->getDeliveryTag());
244			}
245
246			/**
247			* Acknowledge an error during the consumption of the message
248			*
249			* @param Message $message
250			* @param bool $requeue
251			*
252			* @return void
253			*
254			* @throws \AMQPConnectionException
255			* @throws \AMQPChannelException
256			* @throws \AMQPQueueException
257			*/
258			public function nack(Message $message, $requeue = true)
259			{
260			$queue = $this->getQueue($message->getQueue());
261			$queue->nack($message->getDeliveryTag(), $requeue ? AMQP_REQUEUE : AMQP_NOPARAM);
262			}
263
264			/**
265			* Close the connection
266			*
267			* @return void
268			*/
269			public function close()
270			{
271			$this->connection->disconnect();
272			}
273
274			/**
275			* @return \AMQPChannel
276			*
277			* @throws \AMQPConnectionException
278			*/
279			private function getChannel()
280			{
281			if (null === $this->channel) {
282			$this->connection->connect();
283			$this->channel = new \AMQPChannel($this->connection);
284			}
285
286			return $this->channel;
287			}
288
289			/**
290			* @param string $queueName
291			*
292			* @return \AMQPQueue
293			*
294			* @throws \AMQPQueueException
295			* @throws \AMQPConnectionException
296			*/
297			private function getQueue($queueName)
298			{
299			$queue = new \AMQPQueue($this->getChannel());
300			if ($queueName) {
301			$queue->setName($queueName);
302			}
303
304			return $queue;
305			}
306
307			/**
308			* @param string $exchangeName
309			*
310			* @return \AMQPExchange
311			*
312			* @throws \AMQPExchangeException
313			* @throws \AMQPConnectionException
314			*/
315			private function getExchange($exchangeName)
316			{
317			$exchange = new \AMQPExchange($this->getChannel());
318			if ($exchangeName) {
319			$exchange->setName($exchangeName);
320			}
321
322			return $exchange;
323			}
324
325			/**
326			* Returns the message parameters
327			*
328			* @param Message $message
329			*
330			* @return array
331			*/
332			private static function getMessageProperties(Message $message)
333			{
334			$properties = [
335			self::DELIVERY_MODE => 2,
336			self::CONTENT_TYPE => 'text/plain',
337			self::APPLICATION_HEADERS => $message->getHeaders(),
338			];
339
340			if ($message->getCorrelationId() !== null) {
341			$properties[self::CORRELATION_ID] = $message->getCorrelationId();
342			}
343
344			if ($message->getReplyTo() !== null) {
345			$properties[self::REPLY_TO] = $message->getReplyTo();
346			}
347
348			return $properties;
349			}
350
351			private static function generateConsumerTag(
352			$prefix = 'consumer-tags',
353			$length = 10,
354			$chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
355			) {
356			return $prefix . substr(str_shuffle(str_repeat($chars, ceil($length / strlen($chars)))), 1, $length);
357			}
358			}
359

Evaneos / Burrow

Pull Request — master (#30)

PeclAmqpDriver::getChannel() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like