QueueController::actionConsume() - Code Metrics - hiqdev/hiapi - Measure and Improve Code Quality continuously with Scrutinizer

QueueController::actionConsume() A
last analyzed 2021-06-11 17:48 UTC

↳ Parent: QueueController

Complexity

Conditions	5
Paths	2

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	32
c	0
b	0
f	0
rs	9.0968
cc	5
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
     */
    protected $logger;
    /**
     * @var AMQPStreamConnection
     */
    protected $amqp;
    /**
     * @var AutoBusFactoryInterface
     */
    private $busFactory;

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

        parent::__construct($id, $module, $config);
    }

    /**
     * @return \PhpAmqpLib\Channel\AMQPChannel
     */
    protected 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(), ['amqpMessage' => $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', ['amqpMessage' => $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([
            'x-delayed-type' => 'direct',
        ]));
        $channel->queue_bind($queueName, $delayExchange);

        // Send message
        $delayDuration = 1000 * $exception->getSecondsBeforeRetry() * (int)($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 . 'ms', ['amqpMessage' => $msg, 'exception' => $exception]);
    }

    private function storeRejected(string $queueName, AMQPMessage $message, \Exception $exception): void
    {
        $channel = $this->amqp->channel();
        $failedExchange = "$queueName.failed";

        $channel->exchange_declare($failedExchange, 'fanout', false, true, false);
        $channel->basic_publish($message, $failedExchange);
    }
}


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			protected $logger;
29			/**
30			* @var AMQPStreamConnection
31			*/
32			protected $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			$this->logger = $logger;
47			$this->amqp = $amqp;
48			$this->busFactory = $busFactory;
49
50			parent::__construct($id, $module, $config);
51			}
52
53			/**
54			* @return \PhpAmqpLib\Channel\AMQPChannel
55			*/
56			protected 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']);
			0 ignored issues – show Deprecated Code introduced 2020-07-01 13:38 UTC by Report Bug Copy Issue Report The property `PhpAmqpLib\Message\AMQPMessage::$delivery_info` has been deprecated. 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...
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 2019-04-29 16:26 UTC by Report Bug Copy Issue Report The expression `$channel->callbacks` of type `callable[]` 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(), ['amqpMessage' => $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', ['amqpMessage' => $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([
157			'x-delayed-type' => 'direct',
158			]));
159			$channel->queue_bind($queueName, $delayExchange);
160
161			// Send message
162			$delayDuration = 1000 * $exception->getSecondsBeforeRetry() * (int)($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 . 'ms', ['amqpMessage' => $msg, 'exception' => $exception]);
171			}
172
173			private function storeRejected(string $queueName, AMQPMessage $message, \Exception $exception): void
174			{
175			$channel = $this->amqp->channel();
176			$failedExchange = "$queueName.failed";
177
178			$channel->exchange_declare($failedExchange, 'fanout', false, true, false);
179			$channel->basic_publish($message, $failedExchange);
180			}
181			}
182

hiqdev / hiapi

QueueController::actionConsume() A last analyzed 2021-06-11 17:48 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

QueueController::actionConsume() A
last analyzed 2021-06-11 17:48 UTC