Consumer::decodeMessage() - Code Metrics - Inspection of "Added optional delegation" - sunspikes/carrot - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 21d77e...3a0edd )

by Krishnaprasad

created 2016-08-07 20:58 UTC

Consumer::decodeMessage() A

↳ Parent: Consumer

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	4
rs	10
cc	1
eloc	2
nc	1
nop	1

<?php
/**
 * The MIT License (MIT)
 *
 * Copyright (c) 2016 Krishnaprasad MG <[email protected]>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

namespace Sunspikes\Carrot\Consumer;

use PhpAmqpLib\Message\AMQPMessage;
use Sunspikes\Carrot\CarrotConnectionTrait;
use Sunspikes\Carrot\ConfigAwareTrait;
use Sunspikes\Carrot\Exception\ConsumerException;
use PhpAmqpLib\Channel\AMQPChannel;

/**
 * Queue consumer
 */
class Consumer implements ConsumerInterface
{
    use CarrotConnectionTrait,
        ConfigAwareTrait;

    protected $channel;
    protected $exchange;

    /**
     * @param AMQPChannel $channel
     * @param string $exchange
     */
    public function __construct(AMQPChannel $channel, $exchange)
    {
        $this->channel = $channel;
        $this->exchange = $exchange;
    }

    /**
     * @inheritdoc
     */
    public function add($name, callable $callable)
    {
        try {
            $config = $this->config;
            $this->channel->queue_declare(
                $name,
                $config['queue']['passive'],
                $config['queue']['durable'],
                $config['queue']['exclusive'],
                $config['queue']['auto_delete'],
                $config['queue']['no_wait'],
                $config['queue']['arguments'],
                $config['queue']['ticket']
            );
            $this->channel->queue_bind(
                $name,
                $this->exchange,
                $name,
                $config['queue']['no_wait'],
                $config['queue']['arguments'],
                $config['queue']['ticket']
            );
            $this->channel->basic_consume(
                $name,
                $config['consumer']['consumer_tag'],
                $config['consumer']['no_local'],
                $config['consumer']['no_ack'],
                $config['consumer']['exclusive'],
                $config['consumer']['no_wait'],
                function (AMQPMessage $message) use ($callable, $config) {
                    $originalMessage = $message;

                    if (true === $config['delegate']) {
                        $message = $this->decodeMessage($originalMessage);
                    }

                    call_user_func($callable, $message);

                    if (true === $config['delegate']) {
                        $this->acknowledgeMessage($originalMessage);
                    }
                },
                $config['consumer']['ticket'],
                $config['consumer']['arguments']
            );
        } catch (\Exception $e) {
            throw new ConsumerException('Carrot consumer failed to add service: '. $e->getMessage());
        }
    }

    /**
     * Decode the message and return array
     *
     * @param AMQPMessage $message
     * @return array
     */
    public function decodeMessage(AMQPMessage $message)
    {
        return json_decode($message->body);
    }

    /**
     * Send ACK for the message
     *
     * @param AMQPMessage $message
     */
    public function acknowledgeMessage(AMQPMessage $message)
    {
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
    }

    /**
     * @param AMQPMessage $message
     * @param bool $requeue
     */
    public function rejectMessage(AMQPMessage $message, $requeue = false)
    {
        $message->delivery_info['channel']->basic_reject($message->delivery_info['delivery_tag'], $requeue);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
    }

    /**
     * @inheritdoc
     */
    public function listen()
    {
        try {
            while(count($this->channel->callbacks)) {
                if (! $this->channel->getConnection()->isConnected()) {
                    throw new ConsumerException('Carrot consumer disconnected');
                }
                $this->channel->wait(null, true);
            }

            $this->closeConnection();
        } catch (\Exception $e) {
            // consumer failed to listen or disconnected
            throw new ConsumerException("Carrot lost connection with RabbitMQ server");
        }
    }
}



1			<?php
2			/**
3			* The MIT License (MIT)
4			*
5			* Copyright (c) 2016 Krishnaprasad MG <[email protected]>
6			*
7			* Permission is hereby granted, free of charge, to any person obtaining a copy
8			* of this software and associated documentation files (the "Software"), to deal
9			* in the Software without restriction, including without limitation the rights
10			* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11			* copies of the Software, and to permit persons to whom the Software is
12			* furnished to do so, subject to the following conditions:
13			*
14			* The above copyright notice and this permission notice shall be included in all
15			* copies or substantial portions of the Software.
16			*
17			* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18			* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19			* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20			* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21			* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22			* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23			* SOFTWARE.
24			*/
25
26			namespace Sunspikes\Carrot\Consumer;
27
28			use PhpAmqpLib\Message\AMQPMessage;
29			use Sunspikes\Carrot\CarrotConnectionTrait;
30			use Sunspikes\Carrot\ConfigAwareTrait;
31			use Sunspikes\Carrot\Exception\ConsumerException;
32			use PhpAmqpLib\Channel\AMQPChannel;
33
34			/**
35			* Queue consumer
36			*/
37			class Consumer implements ConsumerInterface
38			{
39			use CarrotConnectionTrait,
40			ConfigAwareTrait;
41
42			protected $channel;
43			protected $exchange;
44
45			/**
46			* @param AMQPChannel $channel
47			* @param string $exchange
48			*/
49			public function __construct(AMQPChannel $channel, $exchange)
50			{
51			$this->channel = $channel;
52			$this->exchange = $exchange;
53			}
54
55			/**
56			* @inheritdoc
57			*/
58			public function add($name, callable $callable)
59			{
60			try {
61			$config = $this->config;
62			$this->channel->queue_declare(
63			$name,
64			$config['queue']['passive'],
65			$config['queue']['durable'],
66			$config['queue']['exclusive'],
67			$config['queue']['auto_delete'],
68			$config['queue']['no_wait'],
69			$config['queue']['arguments'],
70			$config['queue']['ticket']
71			);
72			$this->channel->queue_bind(
73			$name,
74			$this->exchange,
75			$name,
76			$config['queue']['no_wait'],
77			$config['queue']['arguments'],
78			$config['queue']['ticket']
79			);
80			$this->channel->basic_consume(
81			$name,
82			$config['consumer']['consumer_tag'],
83			$config['consumer']['no_local'],
84			$config['consumer']['no_ack'],
85			$config['consumer']['exclusive'],
86			$config['consumer']['no_wait'],
87			function (AMQPMessage $message) use ($callable, $config) {
88			$originalMessage = $message;
89
90			if (true === $config['delegate']) {
91			$message = $this->decodeMessage($originalMessage);
92			}
93
94			call_user_func($callable, $message);
95
96			if (true === $config['delegate']) {
97			$this->acknowledgeMessage($originalMessage);
98			}
99			},
100			$config['consumer']['ticket'],
101			$config['consumer']['arguments']
102			);
103			} catch (\Exception $e) {
104			throw new ConsumerException('Carrot consumer failed to add service: '. $e->getMessage());
105			}
106			}
107
108			/**
109			* Decode the message and return array
110			*
111			* @param AMQPMessage $message
112			* @return array
113			*/
114			public function decodeMessage(AMQPMessage $message)
115			{
116			return json_decode($message->body);
117			}
118
119			/**
120			* Send ACK for the message
121			*
122			* @param AMQPMessage $message
123			*/
124			public function acknowledgeMessage(AMQPMessage $message)
125			{
126			$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
			0 ignored issues – show Documentation introduced 2016-08-07 21:02 UTC by Report Bug Copy Issue Report `$message->delivery_info['delivery_tag']` is of type `object<PhpAmqpLib\Channel\AMQPChannel>`, but the function expects a `string`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
127			}
128
129			/**
130			* @param AMQPMessage $message
131			* @param bool $requeue
132			*/
133			public function rejectMessage(AMQPMessage $message, $requeue = false)
134			{
135			$message->delivery_info['channel']->basic_reject($message->delivery_info['delivery_tag'], $requeue);
			0 ignored issues – show Documentation introduced 2016-08-07 21:02 UTC by Report Bug Copy Issue Report `$message->delivery_info['delivery_tag']` is of type `object<PhpAmqpLib\Channel\AMQPChannel>`, but the function expects a `string`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
136			}
137
138			/**
139			* @inheritdoc
140			*/
141			public function listen()
142			{
143			try {
144			while(count($this->channel->callbacks)) {
145			if (! $this->channel->getConnection()->isConnected()) {
146			throw new ConsumerException('Carrot consumer disconnected');
147			}
148			$this->channel->wait(null, true);
149			}
150
151			$this->closeConnection();
152			} catch (\Exception $e) {
153			// consumer failed to listen or disconnected
154			throw new ConsumerException("Carrot lost connection with RabbitMQ server");
155			}
156			}
157			}
158
159

sunspikes / carrot

Push — master ( 21d77e...3a0edd )

Consumer::decodeMessage() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like