Consumer::processMessages() - Code Metrics - Inspection of "[WIP] Experimental 3.0" - eMAGTechLabs/RabbitMqBundle - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed
Pull Request — master (#39)

by Aleksandr
created 2020-12-15 00:23 UTC
Consumer::processMessages() C

↳ Parent: Consumer
Complexity

Conditions	15
Paths	80
Size

Total Lines	58
Code Lines	39
Duplication

Lines	0
Ratio	0 %
Code Coverage

Tests	14
CRAP Score	21.075
Importance

Changes	2
Bugs	0	Features	0
Metric	Value
cc	15
eloc	39
c	2
b	0
f	0
nc	80
nop	2
dl	0
loc	58
ccs	14
cts	20
cp	0.7
crap	21.075
rs	5.9166
How to fix Long Method Complexity

<?php

namespace OldSound\RabbitMqBundle\RabbitMq;

use http\Exception\InvalidArgumentException;
use OldSound\RabbitMqBundle\Declarations\BatchConsumeOptions;
use OldSound\RabbitMqBundle\Declarations\ConsumeOptions;
use OldSound\RabbitMqBundle\Declarations\RpcConsumeOptions;
use OldSound\RabbitMqBundle\Event\AfterProcessingMessageEvent;
filter:
    dependency_paths: ["lib/*"]
use OldSound\RabbitMqBundle\Event\AfterProcessingMessagesEvent;
use OldSound\RabbitMqBundle\Event\AMQPEvent;
filter:
    dependency_paths: ["lib/*"]
use OldSound\RabbitMqBundle\Event\BeforeProcessingMessageEvent;
filter:
    dependency_paths: ["lib/*"]
use OldSound\RabbitMqBundle\Event\BeforeProcessingMessagesEvent;
use OldSound\RabbitMqBundle\Event\OnConsumeEvent;
use OldSound\RabbitMqBundle\Event\OnIdleEvent;
use OldSound\RabbitMqBundle\EventDispatcherAwareTrait;
use OldSound\RabbitMqBundle\ExecuteReceiverStrategy\BatchExecuteReceiverStrategy;
use OldSound\RabbitMqBundle\ExecuteReceiverStrategy\ExecuteReceiverStrategyInterface;
use OldSound\RabbitMqBundle\ExecuteReceiverStrategy\CallbackReceiverExecutorInterface;
filter:
    dependency_paths: ["lib/*"]
use OldSound\RabbitMqBundle\ExecuteReceiverStrategy\MessagesProcessorInterface;
filter:
    dependency_paths: ["lib/*"]
use OldSound\RabbitMqBundle\ExecuteReceiverStrategy\SingleExecuteReceiverStrategy;
use OldSound\RabbitMqBundle\Logger\ExtraContextLogger;
use OldSound\RabbitMqBundle\ReceiverExecutor\BatchReceiverExecutor;
use OldSound\RabbitMqBundle\ReceiverExecutor\ReceiverExecutorDecorator;
use OldSound\RabbitMqBundle\ReceiverExecutor\ReceiverExecutorInterface;
use OldSound\RabbitMqBundle\ReceiverExecutor\ReplyReceiverExecutor;
use OldSound\RabbitMqBundle\ReceiverExecutor\SingleReceiverExecutor;
use OldSound\RabbitMqBundle\Producer\ProducerInterface;
use OldSound\RabbitMqBundle\Receiver\NotReadyReceiveException;
use OldSound\RabbitMqBundle\Receiver\ReceiverException;
use OldSound\RabbitMqBundle\Receiver\ReceiverInterface;
use OldSound\RabbitMqBundle\Receiver\ReplyReceiverInterface;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Exception\AMQPRuntimeException;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use PhpAmqpLib\Message\AMQPMessage;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\NullLogger;
use Symfony\Component\Serializer\SerializerInterface;

class Consumer
{
    use LoggerAwareTrait;
    use EventDispatcherAwareTrait;

    /** @var AMQPChannel */
    protected $channel;
    /** @var Consuming[] */
    protected $consumings = [];
    /** @var string[] */
    protected $consumerTags = [];
    /** @var int|null */
    protected $target;
    /** @var int */
    protected $consumed = 0;
    /** @var bool */
    protected $forceStop = false;
    /**
     * Importrant! If true - then channel can not be used from somewhere else
     * @var bool
     */
    public $multiAck = false;
    /**
     * @var \DateTime|null DateTime after which the consumer will gracefully exit. "Gracefully" means, that
     *      any currently running consumption will not be interrupted.
     */
    public $gracefulMaxExecutionDateTime;
    /** @var int Exit code used, when consumer is closed by the Graceful Max Execution Timeout feature. */
    public $gracefulMaxExecutionTimeoutExitCode = 0;
    /** @var int|null */
    public $timeoutWait;
    /** @var int */
    public $idleTimeout = 0;
    /** @var int */
    public $idleTimeoutExitCode;
    /** @var \DateTime|null */
    public $lastActivityDateTime;

    public function __construct(AMQPChannel $channel)
    {
        $this->channel = $channel;
        $this->logger = new NullLogger();
    }

    public function getChannel(): AMQPChannel
    {
        return $this->channel;
    }

    protected function setup(): Consumer
    {
        foreach($this->consumings as $index => $consuming) {
            $this->channel->basic_qos($consuming->options->qosPrefetchSize, $consuming->options->qosPrefetchCount, false);

            $options = $consuming->options;
            $consuming->consumerTag ? $consuming->consumerTag : sprintf("PHPPROCESS_%s_%s_%s", gethostname(), getmypid(), $index);
            $consumerTag = $this->channel->basic_consume(
                $options->queue,
                $consuming->consumerTag,
                $options->noLocal,
                $options->noAck,
                $options->exclusive,
                false,
                function (AMQPMessage $message) use ($consuming) {
                    $consuming->executeReceiverStrategy->onConsumeCallback($message);
                    $this->maybeStopConsumer();
                });

            $consuming->consumerTag = $consumerTag;
        }

        return $this;
    }

    /**
     * @param iterable|ConsumeOptions[] $queueConsumings
     */
    public function consumeQueues(iterable $queueConsumings)
    {
        foreach ($queueConsumings as $queueConsuming) {
            $this->consumeQueue($queueConsuming);
                   consumeQueue($queueConsuming);
        }
    }

    private function createStrategyByOptions(ConsumeOptions $consumeOptions): ExecuteReceiverStrategyInterface
    {
        if ($consumeOptions instanceof BatchConsumeOptions) {
            return new BatchExecuteReceiverStrategy($consumeOptions->batchCount);
        }
        return new SingleExecuteReceiverStrategy();
    }

    private function createReceiverExecutorByOptions(ConsumeOptions $consumeOptions): ReceiverExecutorInterface
    {
        if ($consumeOptions instanceof BatchConsumeOptions) {
            $receiverExecutor = new BatchReceiverExecutor();
        } else if ($consumeOptions instanceof RpcConsumeOptions) {
            $receiverExecutor = new ReplyReceiverExecutor($consumeOptions);
        } else {
            $receiverExecutor = new SingleReceiverExecutor();
        }

        return $receiverExecutor;
    }

    public function consumeQueue(ConsumeOptions $consumerOptions, $receiver): Consumer
    {
        $executeReceiverStrategy = $this->createStrategyByOptions($consumerOptions);
        $receiverExecutor = $this->createReceiverExecutorByOptions($consumerOptions);
        if (!$receiverExecutor->support($receiver)) {
            throw new \InvalidArgumentException('sdfs');
        }

        $consuming = new Consuming($consumerOptions, $executeReceiverStrategy, $receiverExecutor, $receiver);
        $executeReceiverStrategy->setMessagesProcessor(
            new ReceiverExecutorDecorator($consuming, $this->logger)
        );

        $this->consumings[] = $consuming;

        return $this;
    }

    /**
     * @return ConsumeOptions[]
     */
    public function getConsumings(): array
    {
        return $this->consumings;

    }

    /**
     * Consume the message
     * @param   int     $msgAmount
     * @return  int
     *
     * @throws  AMQPTimeoutException
     */
    public function startConsume(int $msgAmount = null)
    {
        $this->target = $msgAmount;
        $this->consumed = 0;

        $this->setup();

        $this->lastActivityDateTime = new \DateTime();
        while ($this->channel->is_consuming()) {
            $this->dispatchEvent(OnConsumeEvent::NAME, new OnConsumeEvent($this));
            $this->maybeStopConsumer();

            if ($this->forceStop) {
                break;
            }
            /*
             * Be careful not to trigger ::wait() with 0 or less seconds, when
             * graceful max execution timeout is being used.
             */
            $waitTimeout = $this->chooseWaitTimeout();
            if ($this->gracefulMaxExecutionDateTime && $waitTimeout < 1) {
                return $this->gracefulMaxExecutionTimeoutExitCode;
            }

            try {
                $this->channel->wait(null, false, $waitTimeout);
                $this->lastActivityDateTime = new \DateTime();
                if ($this->forceStop) {
                    break;
                }
            } catch (AMQPTimeoutException $e) {
                foreach($this->executeReceiverStrategies as $executeReceiverStrategy) {

                    $executeReceiverStrategy->onCatchTimeout($e);
                }
                $now = new \DateTime();
                if ($this->gracefulMaxExecutionDateTime && $this->gracefulMaxExecutionDateTime <= $now) {
                    return $this->gracefulMaxExecutionTimeoutExitCode;
                }

                if ($this->idleTimeout && ($this->lastActivityDateTime->getTimestamp() + $this->idleTimeout <= $now->getTimestamp())) {
                    $idleEvent = new OnIdleEvent($this);
                    $this->dispatchEvent(OnIdleEvent::NAME, $idleEvent);

                    if ($idleEvent->isForceStop()) {
                        if (null !== $this->idleTimeoutExitCode) {
                            return $this->idleTimeoutExitCode;
                        } else {
                            throw $e;
                        }
                    }
                }
            }
        }

        return 0;
    }

    /**
     * @param AMQPMessage[] $messages
     * @param ConsumeOptions $queueConsuming
     */
    protected function processMessages(array $messages, ConsumeOptions $queueConsuming)
    {
        if (count($messages) === 0) {
            throw new \InvalidArgumentException('Messages can not be empty');
        }

        $canPrecessMultiMessages = $this->getExecuteReceiverStrategy($queueConsuming)->canPrecessMultiMessages();
        $canPrecessMultiMessages = $this->/** @scrutinizer ignore-call */ getExecuteReceiverStrategy($queueConsuming)->canPrecessMultiMessages();
        if (!$canPrecessMultiMessages && count($messages) !== 1) {
            throw new \InvalidArgumentException('Strategy is not supported process of multi messages');
        }

        /** @var int[]|int $flags */
        $flags = [];
        try {
            if ($queueConsuming->receiver instanceof ReceiverInterface) {
                $flags = $queueConsuming->receiver->execute($messages[0]);
            } else if ($queueConsuming->receiver instanceof BatchReceiverInterface) {
filter:
    dependency_paths: ["lib/*"]
                $flags = $queueConsuming->receiver->batchExecute($messages);
            } else if ($queueConsuming->receiver instanceof ReplyReceiverInterface) {
                $reply = $queueConsuming->receiver->execute($messages[0]);
                $isRpcCall = $messages[0]->has('reply_to') && $messages[0]->has('correlation_id');
                if ($isRpcCall) {
                    $this->sendRpcReply($messages[0], $reply);
                           sendRpcReply($messages[0], $reply);
                    $flags = ReceiverInterface::MSG_ACK;
                } else {
                    $flags = ReceiverInterface::MSG_REJECT;
                    // logging
                }
            } else {
                throw new \InvalidArgumentException('TODO');
            }
        } catch (ReceiverException $exception) {
            $flags = $exception->getCode();
        } catch (NotReadyReceiveException $exception) {
            // TODO
            $this->forceStop = true;
            return;
        }

        if (!is_array($flags)) { // spread handle flag for each delivery tag
            $flag = $flags;
            $flags = [];
            foreach ($messages as $message) {
                $flags[$message->getDeliveryTag()] = $flag;
            }
        } else if (count($flags) !== count($messages)) {
            throw new AMQPRuntimeException(
                'Method batchExecute() should return an array with elements equal with the number of messages processed'
            );
        }

        if (!$queueConsuming->noAck) {
            $messages = array_combine(
                array_map(fn ($message) => $message->getDeliveryTag(), $messages),
                $messages
            );

            $this->handleProcessMessages($messages, $flags, $queueConsuming);
            $this->handleProcessMessages(/** @scrutinizer ignore-type */ $messages, $flags, $queueConsuming);
        }
    }

    /**
     * @param AMQPMessage[] $messages
     * @param int[]|RpcReponse[]|RpcResponseException[]|bool[] $replies
     */
    private function handleProcessMessages($messages, array $replies, ConsumeOptions $queueConsuming)
    {
        $executeReceiverStrategy = $this->getExecuteReceiverStrategy($queueConsuming);

        $ack = !array_search(fn ($reply) => $reply !== ReceiverInterface::MSG_ACK, $replies, true);
        if ($this->multiAck && count($messages) > 1 && $ack) {
            $channels = array_map(fn ($message) => $message->getChannel(), $messages);
            if (count($channels) !== array_unique($channels)) { // all messages have same channel
                throw new InvalidArgumentException('Messages can not be processed as multi ack with different channels');
            }

            $lastDeliveryTag = array_key_last($replies);

            $this->channel->basic_ack($lastDeliveryTag, true);
            $this->consumed = $this->consumed + count($messages);
            foreach ($messages as $message) {
                $executeReceiverStrategy->onMessageProcessed($message);
            }
        } else {
            foreach ($replies as $deliveryTag => $reply) {
                $message = $messages[$deliveryTag] ?? null;
                if (null === $message) {
                    throw new AMQPRuntimeException(sprintf('Unknown delivery_tag %d!', $deliveryTag));
                }

                $channel = $message->getChannel();
                $processFlag = $reply;
                if ($processFlag === ReceiverInterface::MSG_REJECT_REQUEUE || false === $processFlag) {
                    $channel->basic_reject($deliveryTag, true); // Reject and requeue message to RabbitMQ
                } else if ($processFlag === ReceiverInterface::MSG_SINGLE_NACK_REQUEUE) {
                    $channel->basic_nack($deliveryTag, false, true); // NACK and requeue message to RabbitMQ
                } else if ($processFlag === ReceiverInterface::MSG_REJECT) {
                    $channel->basic_reject($deliveryTag, false); // Reject and drop
                } else if ($processFlag !== ReceiverInterface::MSG_ACK_SENT) {
                    $channel->basic_ack($deliveryTag); // Remove message from queue only if callback return not false
                }

                $this->consumed++;

                $executeReceiverStrategy->onMessageProcessed($message);
            }
        }
    }

    protected function maybeStopConsumer()
    {
        if ($this->forceStop || ($this->target && $this->consumed == $this->target)) {
0   == false // true
0   == null  // true
123 == false // false
123 == null  // false

// It is often better to use strict comparison
0 === false // false
0 === null  // false
            $this->stopConsuming();
        }
    }

    public function forceStopConsumer()
    {
        $this->forceStop = true;
    }

    public function stopConsuming($immediately = false)
    {
        if (false === $immediately) {
            foreach ($this->executeReceiverStrategies as $executeReceiverStrategy) {

                $executeReceiverStrategy->onStopConsuming();
            }
        }

        foreach ($this->consumerTags as $consumerTag) {
            $this->channel->basic_cancel($consumerTag, false, true);
        }

        $this->consumerTags = [];
    }

    /**
     * @param int $secondsInTheFuture
     */
    public function setGracefulMaxExecutionDateTimeFromSecondsInTheFuture($secondsInTheFuture)
    {
        $this->gracefulMaxExecutionDateTime = new \DateTime("+{$secondsInTheFuture} seconds");
    }

    /**
     * Choose the timeout wait (in seconds) to use for the $this->getChannel()->wait() method.
     */
    private function chooseWaitTimeout(): int
    {
        if ($this->gracefulMaxExecutionDateTime) {
            $allowedExecutionSeconds = $this->gracefulMaxExecutionDateTime->getTimestamp() - time();

            /*
             * Respect the idle timeout if it's set and if it's less than
             * the remaining allowed execution.
             */
            $waitTimeout = $this->idleTimeout && $this->idleTimeout < $allowedExecutionSeconds
                ? $this->idleTimeout
                : $allowedExecutionSeconds;
        } else {
            $waitTimeout = $this->idleTimeout;
        }

        if (!is_null($this->timeoutWait) && $this->timeoutWait > 0) {
            $waitTimeout = min($waitTimeout, $this->timeoutWait);
        }
        return $waitTimeout;
    }
}

eMAGTechLabs / RabbitMqBundle

Pull Request — master (#39)

Consumer::processMessages() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Duplication Side-by-Side

Filter issues like
