Queue::get() - Code Metrics - Inspection of "phpunit composer dep" - thomasvargiu/AMQPAL - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f55883...4285c1 )

by Thomas Mauro

created 2016-03-22 19:18 UTC

Queue::get() A

↳ Parent: Queue

Complexity

Conditions	2
Paths	2

Size

Total Lines	10
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	10
ccs	5
cts	5
cp	1
rs	9.4285
cc	2
eloc	5
nc	2
nop	1
crap	2

<?php

namespace AMQPAL\Adapter\PhpAmqpLib;

use PhpAmqpLib\Message\AMQPMessage;
use AMQPAL\Adapter\ChannelInterface;
use AMQPAL\Adapter\ConnectionInterface;
use AMQPAL\Adapter\ConsumerInterface;
use AMQPAL\Adapter\Exception;
use AMQPAL\Adapter\Message;
use AMQPAL\Adapter\QueueInterface;
use AMQPAL\Options;

/**
 * Class Queue
 *
 * @package AMQPAL\Adapter\PhpAmqpLib
 */
class Queue implements QueueInterface
{
    /**
     * @var Channel
     */
    protected $channel;
    /**
     * @var Options\QueueOptions
     */
    protected $options;
    /**
     * @var MessageMapper
     */
    protected $messageMapper;

    /**
     * Declare a new queue on the broker.
     *
     * @return integer the message count.
     */
    public function declareQueue()
    {
        $this->channel->getResource()->queue_declare(
            $this->options->getName(),
            $this->options->isPassive(),
            $this->options->isDurable(),
            $this->options->isExclusive(),
            $this->options->isAutoDelete(),
            $this->options->isNoWait(),
            $this->options->getArguments()
        );

        return $this;
    }

    /**
     * Bind the given queue to a routing key on an exchange.
     *
     * @param string $exchangeName Name of the exchange to bind to.
     * @param string $routingKey   Pattern or routing key to bind with.
     * @param bool   $noWait       No wait for a reply
     * @param array  $arguments    Additional binding arguments.
     *
     * @return boolean
     */
    public function bind($exchangeName, $routingKey = null, $noWait = false, array $arguments = [])
    {
        if (null === $routingKey) {
            $routingKey = '';
        }
        $queueName = $this->options->getName();
        $this->channel->getResource()->queue_bind($queueName, $exchangeName, $routingKey, $noWait, $arguments);

        return $this;
    }

    /**
     * Remove a routing key binding on an exchange from the given queue.
     *
     * @param string $exchangeName  The name of the exchange on which the
     *                              queue is bound.
     * @param string $routingKey    The binding routing key used by the
     *                              queue.
     * @param array  $arguments     Additional binding arguments.
     *
     * @return $this
     */
    public function unbind($exchangeName, $routingKey = null, array $arguments = [])
    {
        if (null === $routingKey) {
            $routingKey = '';
        }
        $queueName = $this->options->getName();
        $this->channel->getResource()->queue_unbind($queueName, $exchangeName, $routingKey, $arguments);

        return $this;
    }

    /**
     * Acknowledge the receipt of a message.
     *
     * @param string $deliveryTag   The message delivery tag of which to
     *                              acknowledge receipt.
     * @param bool   $multiple      Acknowledge all previous
     *                              unacked messages as well.
     *
     * @return $this
     */
    public function ack($deliveryTag, $multiple = false)
    {
        $this->channel->getResource()->basic_ack($deliveryTag, $multiple);

        return $this;
    }

    /**
     * Mark a message as explicitly not acknowledged.
     *
     * Mark the message identified by delivery_tag as explicitly not
     * acknowledged. This method can only be called on messages that have not
     * yet been acknowledged. When called, the broker will immediately put the
     * message back onto the queue, instead of waiting until the connection is
     * closed. This method is only supported by the RabbitMQ broker. The
     * behavior of calling this method while connected to any other broker is
     * undefined.
     *
     * @param string $deliveryTag   Delivery tag of last message to reject.
     * @param bool   $requeue       Requeue the message(s).
     * @param bool   $multiple      Mark as not acknowledge all previous
     *                              unacked messages as well.
     *
     * @return $this
     */
    public function nack($deliveryTag, $requeue = false, $multiple = false)
    {
        $this->channel->getResource()->basic_nack($deliveryTag, $multiple, $requeue);

        return $this;
    }

    /**
     * Mark one message as explicitly not acknowledged.
     *
     * Mark the message identified by delivery_tag as explicitly not
     * acknowledged. This method can only be called on messages that have not
     * yet been acknowledged.
     *
     * @param string $deliveryTag Delivery tag of the message to reject.
     * @param bool   $requeue     Requeue the message(s).
     *
     * @return $this
     */
    public function reject($deliveryTag, $requeue = false)
    {
        $this->channel->getResource()->basic_reject($deliveryTag, $requeue);

        return $this;
    }

    /**
     * Purge the contents of a queue.
     *
     * @return $this
     */
    public function purge()
    {
        $this->channel->getResource()->queue_purge($this->options->getName());

        return $this;
    }

    /**
     * Cancel a queue that is already bound to an exchange and routing key.
     *
     * @param string $consumerTag  The queue name to cancel, if the queue
     *                             object is not already representative of
     *                             a queue.
     *
     * @return $this
     */
    public function cancel($consumerTag = '')
    {
        $this->channel->getResource()->basic_cancel($consumerTag);

        return $this;
    }

    /**
     * Delete a queue from the broker.
     *
     * This includes its entire contents of unread or unacknowledged messages.
     *
     * @param bool $ifUnused        Optionally $ifUnused can be specified
     *                              to indicate the queue should not be
     *                              deleted until no clients are connected to
     *                              it.
     * @param bool $ifEmpty         Optionally $ifUnused can be specified
     *                              to indicate the queue should not be
     *                              deleted until it's empty
     * @param bool $noWait          No wait for a reply
     *
     * @return $this
     * @throws Exception\InvalidArgumentException
     */
    public function delete($ifUnused = false, $ifEmpty = false, $noWait = false)
    {
        $this->channel->getResource()->queue_delete($this->options->getName(), $ifUnused, $ifEmpty, $noWait);

        return $this;
    }

    /**
     * Retrieve the next message from the queue.
     *
     * @param bool $autoAck
     * @return null|Message
     * @throws \OutOfBoundsException
     */
    public function get($autoAck = false)
    {
        /** @var AMQPMessage $message */
        $message = $this->channel->getResource()->basic_get($this->getOptions()->getName(), !$autoAck);
        if (!$message) {
            return null;
        }

        return $this->getMessageMapper()->toMessage($message);
    }

    /**
     * @return Options\QueueOptions
     */
    public function getOptions()
    {
        return $this->options;
    }

    /**
     * @param Options\QueueOptions $options
     * @return $this
     */
    public function setOptions(Options\QueueOptions $options)
    {
        $this->options = $options;
        return $this;
    }

    /**
     * @return MessageMapper
     */
    public function getMessageMapper()
    {
        if (!$this->messageMapper) {
            $this->messageMapper = new MessageMapper();
        }
        return $this->messageMapper;
    }

    /**
     * @param MessageMapper $messageMapper
     * @return $this
     */
    public function setMessageMapper(MessageMapper $messageMapper)
    {
        $this->messageMapper = $messageMapper;
        return $this;
    }

    /**
     * Consume messages from a queue.
     *
     * @param string                          $consumerTag  A string describing this consumer. Used
     *                                                      for canceling subscriptions with cancel().
     * @param bool                            $noLocal
     * @param bool                            $autoAck
     * @param bool                            $exclusive
     * @param bool                            $nowait       No wait for a reply.
     * @param callback|ConsumerInterface|null $callback     A callback function to which the
     *                                                      consumed message will be passed.
     * @return $this
     */
    public function consume(
        $consumerTag = null,
        $noLocal = false,
        $autoAck = false,
        $exclusive = false,
        $nowait = false,
        callable $callback = null
    ) {
        if (null === $consumerTag) {
            $consumerTag = '';
        }

        $queue = $this->getOptions()->getName();

        $consumerCallback = null;
        if ($callback) {
            $consumerCallback = new ConsumerCallback($callback, $this);
            $consumerCallback->setMessageMapper($this->getMessageMapper());
        }

        $this->channel->getResource()
            ->basic_consume($queue, $consumerTag, $noLocal, !$autoAck, $exclusive, $nowait, $consumerCallback);

        return $this;
    }

    /**
     * Get the Channel object in use
     *
     * @return ChannelInterface
     */
    public function getChannel()
    {
        return $this->channel;
    }

    /**
     * @param Channel $channel
     * @return $this
     */
    public function setChannel(Channel $channel)
    {
        $this->channel = $channel;
        return $this;
    }

    /**
     * Get the Connection object in use
     *
     * @return ConnectionInterface
     */
    public function getConnection()
    {
        return $this->channel->getConnection();
    }
}


1		<?php
2
3		namespace AMQPAL\Adapter\PhpAmqpLib;
4
5		use PhpAmqpLib\Message\AMQPMessage;
6		use AMQPAL\Adapter\ChannelInterface;
7		use AMQPAL\Adapter\ConnectionInterface;
8		use AMQPAL\Adapter\ConsumerInterface;
9		use AMQPAL\Adapter\Exception;
10		use AMQPAL\Adapter\Message;
11		use AMQPAL\Adapter\QueueInterface;
12		use AMQPAL\Options;
13
14		/**
15		* Class Queue
16		*
17		* @package AMQPAL\Adapter\PhpAmqpLib
18		*/
19		class Queue implements QueueInterface
20		{
21		/**
22		* @var Channel
23		*/
24		protected $channel;
25		/**
26		* @var Options\QueueOptions
27		*/
28		protected $options;
29		/**
30		* @var MessageMapper
31		*/
32		protected $messageMapper;
33
34		/**
35		* Declare a new queue on the broker.
36		*
37		* @return integer the message count.
38		*/
39	1	public function declareQueue()
40		{
41	1	$this->channel->getResource()->queue_declare(
42	1	$this->options->getName(),
43	1	$this->options->isPassive(),
44	1	$this->options->isDurable(),
45	1	$this->options->isExclusive(),
46	1	$this->options->isAutoDelete(),
47	1	$this->options->isNoWait(),
48	1	$this->options->getArguments()
49		);
50
51	1	return $this;
52		}
53
54		/**
55		* Bind the given queue to a routing key on an exchange.
56		*
57		* @param string $exchangeName Name of the exchange to bind to.
58		* @param string $routingKey Pattern or routing key to bind with.
59		* @param bool $noWait No wait for a reply
60		* @param array $arguments Additional binding arguments.
61		*
62		* @return boolean
63		*/
64	2	public function bind($exchangeName, $routingKey = null, $noWait = false, array $arguments = [])
65		{
66	2	if (null === $routingKey) {
67	1	$routingKey = '';
68		}
69	2	$queueName = $this->options->getName();
70	2	$this->channel->getResource()->queue_bind($queueName, $exchangeName, $routingKey, $noWait, $arguments);
71
72	2	return $this;
73		}
74
75		/**
76		* Remove a routing key binding on an exchange from the given queue.
77		*
78		* @param string $exchangeName The name of the exchange on which the
79		* queue is bound.
80		* @param string $routingKey The binding routing key used by the
81		* queue.
82		* @param array $arguments Additional binding arguments.
83		*
84		* @return $this
85		*/
86	2	public function unbind($exchangeName, $routingKey = null, array $arguments = [])
87		{
88	2	if (null === $routingKey) {
89	1	$routingKey = '';
90		}
91	2	$queueName = $this->options->getName();
92	2	$this->channel->getResource()->queue_unbind($queueName, $exchangeName, $routingKey, $arguments);
93
94	2	return $this;
95		}
96
97		/**
98		* Acknowledge the receipt of a message.
99		*
100		* @param string $deliveryTag The message delivery tag of which to
101		* acknowledge receipt.
102		* @param bool $multiple Acknowledge all previous
103		* unacked messages as well.
104		*
105		* @return $this
106		*/
107	2	public function ack($deliveryTag, $multiple = false)
108		{
109	2	$this->channel->getResource()->basic_ack($deliveryTag, $multiple);
110
111	2	return $this;
112		}
113
114		/**
115		* Mark a message as explicitly not acknowledged.
116		*
117		* Mark the message identified by delivery_tag as explicitly not
118		* acknowledged. This method can only be called on messages that have not
119		* yet been acknowledged. When called, the broker will immediately put the
120		* message back onto the queue, instead of waiting until the connection is
121		* closed. This method is only supported by the RabbitMQ broker. The
122		* behavior of calling this method while connected to any other broker is
123		* undefined.
124		*
125		* @param string $deliveryTag Delivery tag of last message to reject.
126		* @param bool $requeue Requeue the message(s).
127		* @param bool $multiple Mark as not acknowledge all previous
128		* unacked messages as well.
129		*
130		* @return $this
131		*/
132	4	public function nack($deliveryTag, $requeue = false, $multiple = false)
133		{
134	4	$this->channel->getResource()->basic_nack($deliveryTag, $multiple, $requeue);
135
136	4	return $this;
137		}
138
139		/**
140		* Mark one message as explicitly not acknowledged.
141		*
142		* Mark the message identified by delivery_tag as explicitly not
143		* acknowledged. This method can only be called on messages that have not
144		* yet been acknowledged.
145		*
146		* @param string $deliveryTag Delivery tag of the message to reject.
147		* @param bool $requeue Requeue the message(s).
148		*
149		* @return $this
150		*/
151	2	public function reject($deliveryTag, $requeue = false)
152		{
153	2	$this->channel->getResource()->basic_reject($deliveryTag, $requeue);
154
155	2	return $this;
156		}
157
158		/**
159		* Purge the contents of a queue.
160		*
161		* @return $this
162		*/
163	1	public function purge()
164		{
165	1	$this->channel->getResource()->queue_purge($this->options->getName());
166
167	1	return $this;
168		}
169
170		/**
171		* Cancel a queue that is already bound to an exchange and routing key.
172		*
173		* @param string $consumerTag The queue name to cancel, if the queue
174		* object is not already representative of
175		* a queue.
176		*
177		* @return $this
178		*/
179	1	public function cancel($consumerTag = '')
180		{
181	1	$this->channel->getResource()->basic_cancel($consumerTag);
182
183	1	return $this;
184		}
185
186		/**
187		* Delete a queue from the broker.
188		*
189		* This includes its entire contents of unread or unacknowledged messages.
190		*
191		* @param bool $ifUnused Optionally $ifUnused can be specified
192		* to indicate the queue should not be
193		* deleted until no clients are connected to
194		* it.
195		* @param bool $ifEmpty Optionally $ifUnused can be specified
196		* to indicate the queue should not be
197		* deleted until it's empty
198		* @param bool $noWait No wait for a reply
199		*
200		* @return $this
201		* @throws Exception\InvalidArgumentException
202		*/
203	8	public function delete($ifUnused = false, $ifEmpty = false, $noWait = false)
204		{
205	8	$this->channel->getResource()->queue_delete($this->options->getName(), $ifUnused, $ifEmpty, $noWait);
206
207	8	return $this;
208		}
209
210		/**
211		* Retrieve the next message from the queue.
212		*
213		* @param bool $autoAck
214		* @return null\|Message
215		* @throws \OutOfBoundsException
216		*/
217	2	public function get($autoAck = false)
218		{
219		/** @var AMQPMessage $message */
220	2	$message = $this->channel->getResource()->basic_get($this->getOptions()->getName(), !$autoAck);
221	2	if (!$message) {
222	1	return null;
223		}
224
225	1	return $this->getMessageMapper()->toMessage($message);
226		}
227
228		/**
229		* @return Options\QueueOptions
230		*/
231	6	public function getOptions()
232		{
233	6	return $this->options;
234		}
235
236		/**
237		* @param Options\QueueOptions $options
238		* @return $this
239		*/
240	29	public function setOptions(Options\QueueOptions $options)
241		{
242	29	$this->options = $options;
243	29	return $this;
244		}
245
246		/**
247		* @return MessageMapper
248		*/
249	5	public function getMessageMapper()
250		{
251	5	if (!$this->messageMapper) {
252	1	$this->messageMapper = new MessageMapper();
253		}
254	5	return $this->messageMapper;
255		}
256
257		/**
258		* @param MessageMapper $messageMapper
259		* @return $this
260		*/
261	5	public function setMessageMapper(MessageMapper $messageMapper)
262		{
263	5	$this->messageMapper = $messageMapper;
264	5	return $this;
265		}
266
267		/**
268		* Consume messages from a queue.
269		*
270		* @param string $consumerTag A string describing this consumer. Used
271		* for canceling subscriptions with cancel().
272		* @param bool $noLocal
273		* @param bool $autoAck
274		* @param bool $exclusive
275		* @param bool $nowait No wait for a reply.
276		* @param callback\|ConsumerInterface\|null $callback A callback function to which the
277		* consumed message will be passed.
278		* @return $this
279		*/
280	3	public function consume(
281		$consumerTag = null,
282		$noLocal = false,
283		$autoAck = false,
284		$exclusive = false,
285		$nowait = false,
286		callable $callback = null
287		) {
288	3	if (null === $consumerTag) {
289	1	$consumerTag = '';
290		}
291
292	3	$queue = $this->getOptions()->getName();
293
294	3	$consumerCallback = null;
295	3	if ($callback) {
296	3	$consumerCallback = new ConsumerCallback($callback, $this);
297	3	$consumerCallback->setMessageMapper($this->getMessageMapper());
298		}
299
300	3	$this->channel->getResource()
301	3	->basic_consume($queue, $consumerTag, $noLocal, !$autoAck, $exclusive, $nowait, $consumerCallback);
302
303	3	return $this;
304		}
305
306		/**
307		* Get the Channel object in use
308		*
309		* @return ChannelInterface
310		*/
311	1	public function getChannel()
312		{
313	1	return $this->channel;
314		}
315
316		/**
317		* @param Channel $channel
318		* @return $this
319		*/
320	30	public function setChannel(Channel $channel)
321		{
322	30	$this->channel = $channel;
323	30	return $this;
324		}
325
326		/**
327		* Get the Connection object in use
328		*
329		* @return ConnectionInterface
330		*/
331	1	public function getConnection()
332		{
333	1	return $this->channel->getConnection();
334		}
335		}
336

thomasvargiu / AMQPAL

Push — master ( f55883...4285c1 )

Queue::get() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like