QueueController::actionConsume() - Code Metrics - Inspection of "Merge branch 'v2'" - hiqdev/hiapi - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 4bcd21...430cdc )

by Dmitry

created 2018-05-14 09:57 UTC

QueueController::actionConsume() B

↳ Parent: QueueController

Complexity

Conditions	5
Paths	2

Size

Total Lines	32
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	32
rs	8.439
cc	5
eloc	20
nc	2
nop	2

<?php

namespace hiapi\console;

use hiapi\exceptions\NotProcessableException;
use hiqdev\yii2\autobus\components\AutoBusFactoryInterface;
use hiqdev\yii2\autobus\components\AutoBusInterface;
use hiqdev\yii2\autobus\exceptions\WrongCommandException;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use Psr\Log\LoggerInterface;
use yii\base\Module;
use yii\console\ExitCode;
use yii\helpers\Console;

/**
 * Class QueueController
 *
 * @author Dmytro Naumenko <[email protected]>
 */
class QueueController extends \yii\console\Controller
{
    /**
     * @var LoggerInterface
     */
    private $logger;
    /**
     * @var AMQPStreamConnection
     */
    private $amqp;
    /**
     * @var AutoBusFactoryInterface
     */
    private $busFactory;

    public function __construct(
        $id,
        Module $module,
        AMQPStreamConnection $amqp,
        LoggerInterface $logger,
        AutoBusFactoryInterface $busFactory,
        array $config = []
    ) {
        parent::__construct($id, $module, $config);

        $this->logger = $logger;
        $this->amqp = $amqp;
        $this->busFactory = $busFactory;
    }

    /**
     * @return \PhpAmqpLib\Channel\AMQPChannel
     */
    private function createChannel(string $queue): AMQPChannel
    {
        $channel = $this->amqp->channel();
        $channel->queue_declare($queue, false, true, false, false);

        return $channel;
    }

    /**
     * @param string $queueName
     * @param int $messagesCount
     * @return int
     */
    public function actionConsume(string $queueName, $messagesCount = 100)
    {
        $channel = $this->createChannel($queueName);
        $bus = $this->busFactory->get($queueName);

        Console::output(' [*] Waiting for messages. To exit press CTRL+C');

        $callback = function (AMQPMessage $msg) use (&$messagesCount, $queueName, $channel, $bus) {
            Console::output(' [x] Received ' . $msg->body);
            $channel->basic_ack($msg->delivery_info['delivery_tag']);
            $messagesCount--;

            try {
                $this->handle($bus, $msg);
            } catch (NotProcessableException $e) {
                $this->requeue($queueName, $msg, $e);
            } catch (\Exception $e) {
                $this->handleError($queueName, $msg, $e);
            }
        };

        $channel->basic_qos(null, 1, null);
function acceptsInteger($int) { }

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

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
        $channel->basic_consume($queueName, '', false, false, false, false, $callback);

        while ($channel->callbacks && $messagesCount > 0) {

            $channel->wait();
        }

        Console::output(' [x] Reached consumed messages limit. Stopping process.');

        return ExitCode::OK;
    }

    private function handleError(string $queueName, AMQPMessage $message, \Exception $exception)
    {
        Console::error(' [E] Error: ' . $exception->getMessage());
        $this->logger->warning('Failed to handle message: ' . $exception->getMessage(), ['message' => $message, 'exception' => $exception]);
        $this->storeRejected($queueName, $message, $exception);
    }

    /**
     * Decodes AMQP message and sends it to the handler
     * // TODO: move to separate class?
     *
     * @param AMQPMessage $msg
     * @throws WrongCommandException
     */
    protected function handle(AutoBusInterface $bus, AMQPMessage $msg): void
    {
        if ($msg->get_properties()['content_type'] !== 'application/json') {
            throw new \RuntimeException('Do not know how to decode ' . $msg->getContentEncoding());
        }

        $body = json_decode($msg->getBody(), true);
        if (!isset($body['name'])) {
            throw new WrongCommandException('Message must have a name');
        }
        $parts = explode('\\', $body['name']);
        $name = array_pop($parts);

        $bus->runCommand($name, $body);
    }

    /**
     * Resends message to queue with a delay
     *
     * @param string $queueName
     * @param AMQPMessage $msg
     * @param NotProcessableException $exception
     */
    private function requeue(string $queueName, AMQPMessage $msg, NotProcessableException $exception): void
    {
        $tries = 0;
        $headers = $msg->get_properties()['application_headers'];
        if ($headers instanceof AMQPTable) {
            $tries = $headers->getNativeData()['x-number-of-tries'] ?? 0;
        }

        if ($exception->getMaxTries() !== null && $tries >= $exception->getMaxTries()) {
            $this->logger->debug('No tries left for message. Marking it as an error', ['message' => $msg, 'exception' => $exception]);
            $this->handleError($queueName, $msg, $exception);
            return;
        }

        // Init delay exchange
        $channel = $this->amqp->channel();
        $delayExchange = "$queueName.delayed";
        $channel->exchange_declare($delayExchange, 'x-delayed-message', false, true, true, false, false, new AMQPTable([
function acceptsInteger($int) { }

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

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
            'x-delayed-type' => 'direct',
        ]));
        $channel->queue_bind($queueName, $delayExchange);

        // Send message
        $delayDuration = 1000 * $exception->getSecondsBeforeRetry() * (int)pow($exception->getProgressionMultiplier(), $tries);
        $delayMessage = new AMQPMessage($msg->getBody(), array_merge($msg->get_properties(), [
            'application_headers' => new AMQPTable([
                'x-delay' => $delayDuration,
                'x-number-of-tries' => $tries + 1,
            ]),
        ]));
        $channel->basic_publish($delayMessage, $delayExchange, '');
        $this->logger->debug('Delayed message for ' . $delayDuration, ['message' => $msg, 'exception' => $exception]);
    }

    private function storeRejected(string $queueName, AMQPMessage $message, \Exception $exception): void

    {
        // TODO: store $exception as well
        $channel = $this->createChannel("$queueName.failed");
        $channel->basic_publish($message, "$queueName.failed");
    }
}


1			<?php
2
3			namespace hiapi\console;
4
5			use hiapi\exceptions\NotProcessableException;
6			use hiqdev\yii2\autobus\components\AutoBusFactoryInterface;
7			use hiqdev\yii2\autobus\components\AutoBusInterface;
8			use hiqdev\yii2\autobus\exceptions\WrongCommandException;
9			use PhpAmqpLib\Channel\AMQPChannel;
10			use PhpAmqpLib\Connection\AMQPStreamConnection;
11			use PhpAmqpLib\Message\AMQPMessage;
12			use PhpAmqpLib\Wire\AMQPTable;
13			use Psr\Log\LoggerInterface;
14			use yii\base\Module;
15			use yii\console\ExitCode;
16			use yii\helpers\Console;
17
18			/**
19			* Class QueueController
20			*
21			* @author Dmytro Naumenko <[email protected]>
22			*/
23			class QueueController extends \yii\console\Controller
24			{
25			/**
26			* @var LoggerInterface
27			*/
28			private $logger;
29			/**
30			* @var AMQPStreamConnection
31			*/
32			private $amqp;
33			/**
34			* @var AutoBusFactoryInterface
35			*/
36			private $busFactory;
37
38			public function __construct(
39			$id,
40			Module $module,
41			AMQPStreamConnection $amqp,
42			LoggerInterface $logger,
43			AutoBusFactoryInterface $busFactory,
44			array $config = []
45			) {
46			parent::__construct($id, $module, $config);
47
48			$this->logger = $logger;
49			$this->amqp = $amqp;
50			$this->busFactory = $busFactory;
51			}
52
53			/**
54			* @return \PhpAmqpLib\Channel\AMQPChannel
55			*/
56			private function createChannel(string $queue): AMQPChannel
57			{
58			$channel = $this->amqp->channel();
59			$channel->queue_declare($queue, false, true, false, false);
60
61			return $channel;
62			}
63
64			/**
65			* @param string $queueName
66			* @param int $messagesCount
67			* @return int
68			*/
69			public function actionConsume(string $queueName, $messagesCount = 100)
70			{
71			$channel = $this->createChannel($queueName);
72			$bus = $this->busFactory->get($queueName);
73
74			Console::output(' [*] Waiting for messages. To exit press CTRL+C');
75
76			$callback = function (AMQPMessage $msg) use (&$messagesCount, $queueName, $channel, $bus) {
77			Console::output(' [x] Received ' . $msg->body);
78			$channel->basic_ack($msg->delivery_info['delivery_tag']);
79			$messagesCount--;
80
81			try {
82			$this->handle($bus, $msg);
83			} catch (NotProcessableException $e) {
84			$this->requeue($queueName, $msg, $e);
85			} catch (\Exception $e) {
86			$this->handleError($queueName, $msg, $e);
87			}
88			};
89
90			$channel->basic_qos(null, 1, null);
			0 ignored issues – show Documentation introduced 2018-05-04 13:47 UTC by Report Bug Copy Issue Report `null` is of type `null`, but the function expects a `boolean`. 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...
91			$channel->basic_consume($queueName, '', false, false, false, false, $callback);
92
93			while ($channel->callbacks && $messagesCount > 0) {
			0 ignored issues – show Bug Best Practice introduced 2018-05-04 13:47 UTC by Report Bug Copy Issue Report The expression `$channel->callbacks` of type `array` is implicitly converted to a boolean; are you sure this is intended? If so, consider using `! empty($expr)` instead to make it clear that you intend to check for an array without elements. This check marks implicit conversions of arrays to boolean values in a comparison. While in PHP an empty array is considered to be equal (but not identical) to false, this is not always apparent. Consider making the comparison explicit by using `empty(..)` or `! empty(...)` instead. Loading history...
94			$channel->wait();
95			}
96
97			Console::output(' [x] Reached consumed messages limit. Stopping process.');
98
99			return ExitCode::OK;
100			}
101
102			private function handleError(string $queueName, AMQPMessage $message, \Exception $exception)
103			{
104			Console::error(' [E] Error: ' . $exception->getMessage());
105			$this->logger->warning('Failed to handle message: ' . $exception->getMessage(), ['message' => $message, 'exception' => $exception]);
106			$this->storeRejected($queueName, $message, $exception);
107			}
108
109			/**
110			* Decodes AMQP message and sends it to the handler
111			* // TODO: move to separate class?
112			*
113			* @param AMQPMessage $msg
114			* @throws WrongCommandException
115			*/
116			protected function handle(AutoBusInterface $bus, AMQPMessage $msg): void
117			{
118			if ($msg->get_properties()['content_type'] !== 'application/json') {
119			throw new \RuntimeException('Do not know how to decode ' . $msg->getContentEncoding());
120			}
121
122			$body = json_decode($msg->getBody(), true);
123			if (!isset($body['name'])) {
124			throw new WrongCommandException('Message must have a name');
125			}
126			$parts = explode('\\', $body['name']);
127			$name = array_pop($parts);
128
129			$bus->runCommand($name, $body);
130			}
131
132			/**
133			* Resends message to queue with a delay
134			*
135			* @param string $queueName
136			* @param AMQPMessage $msg
137			* @param NotProcessableException $exception
138			*/
139			private function requeue(string $queueName, AMQPMessage $msg, NotProcessableException $exception): void
140			{
141			$tries = 0;
142			$headers = $msg->get_properties()['application_headers'];
143			if ($headers instanceof AMQPTable) {
144			$tries = $headers->getNativeData()['x-number-of-tries'] ?? 0;
145			}
146
147			if ($exception->getMaxTries() !== null && $tries >= $exception->getMaxTries()) {
148			$this->logger->debug('No tries left for message. Marking it as an error', ['message' => $msg, 'exception' => $exception]);
149			$this->handleError($queueName, $msg, $exception);
150			return;
151			}
152
153			// Init delay exchange
154			$channel = $this->amqp->channel();
155			$delayExchange = "$queueName.delayed";
156			$channel->exchange_declare($delayExchange, 'x-delayed-message', false, true, true, false, false, new AMQPTable([
			0 ignored issues – show Documentation introduced 2018-05-14 10:08 UTC by Report Bug Copy Issue Report `new \PhpAmqpLib\Wire\AMQ...yed-type' => 'direct'))` is of type `object<PhpAmqpLib\Wire\AMQPTable>`, but the function expects a `array`. 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...
157			'x-delayed-type' => 'direct',
158			]));
159			$channel->queue_bind($queueName, $delayExchange);
160
161			// Send message
162			$delayDuration = 1000 * $exception->getSecondsBeforeRetry() * (int)pow($exception->getProgressionMultiplier(), $tries);
163			$delayMessage = new AMQPMessage($msg->getBody(), array_merge($msg->get_properties(), [
164			'application_headers' => new AMQPTable([
165			'x-delay' => $delayDuration,
166			'x-number-of-tries' => $tries + 1,
167			]),
168			]));
169			$channel->basic_publish($delayMessage, $delayExchange, '');
170			$this->logger->debug('Delayed message for ' . $delayDuration, ['message' => $msg, 'exception' => $exception]);
171			}
172
173			private function storeRejected(string $queueName, AMQPMessage $message, \Exception $exception): void
			0 ignored issues – show Unused Code introduced 2018-05-14 10:08 UTC by Report Bug Copy Issue Report The parameter `$exception` is not used and could be removed. This check looks from parameters that have been defined for a function or method, but which are not used in the method body. Loading history...
174			{
175			// TODO: store $exception as well
176			$channel = $this->createChannel("$queueName.failed");
177			$channel->basic_publish($message, "$queueName.failed");
178			}
179			}
180

hiqdev / hiapi

Push — master ( 4bcd21...430cdc )

QueueController::actionConsume() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like