OldSound\RabbitMqBundle\RabbitMq\Consumer

Complexity

Total Complexity

73

Size/Duplication

Total Lines	366
Duplicated Lines	0 %

Test Coverage

Coverage

66.67%

Importance

Changes	22
Bugs	0	Features	3

16 Methods

Rating	Name	Duplication	Size	Complexity
A	consumeQueue()	0	16	2
A	setGracefulMaxExecutionDateTimeFromSecondsInTheFuture()	0	3	1
A	__construct()	0	4	1
C	handleProcessMessages()	0	40	13
A	consumeQueues()	0	4	2
A	getChannel()	0	3	1
A	getConsumings()	0	3	1
A	setup()	0	23	3
C	processMessages()	0	58	15
A	createStrategyByOptions()	0	6	2
C	startConsume()	0	55	14
A	maybeStopConsumer()	0	4	4
A	forceStopConsumer()	0	3	1
A	stopConsuming()	0	13	4
A	createReceiverExecutorByOptions()	0	11	3
A	chooseWaitTimeout()	0	20	6

<?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;

The type OldSound\RabbitMqBundle\Event\AfterProcessingMessageEvent was not found. Maybe you did not declare it correctly or list all dependencies?

use OldSound\RabbitMqBundle\Event\AfterProcessingMessagesEvent;
use OldSound\RabbitMqBundle\Event\AMQPEvent;

The type OldSound\RabbitMqBundle\Event\AMQPEvent was not found. Maybe you did not declare it correctly or list all dependencies?

use OldSound\RabbitMqBundle\Event\BeforeProcessingMessageEvent;

The type OldSound\RabbitMqBundle\Event\BeforeProcessingMessageEvent was not found. Maybe you did not declare it correctly or list all dependencies?

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;

The type OldSound\RabbitMqBundle\ExecuteReceiverStrategy\CallbackReceiverExecutorInterface was not found. Maybe you did not declare it correctly or list all dependencies?

use OldSound\RabbitMqBundle\ExecuteReceiverStrategy\MessagesProcessorInterface;

The type OldSound\RabbitMqBundle\ExecuteReceiverStrategy\MessagesProcessorInterface was not found. Maybe you did not declare it correctly or list all dependencies?

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);

The call to OldSound\RabbitMqBundle\RabbitMq\Consumer::consumeQueue() has too few arguments starting with receiver.

        }
    }

    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;

The expression return $this->consumings returns the type OldSound\RabbitMqBundle\RabbitMq\Consuming[] which is incompatible with the documented return type OldSound\RabbitMqBundle\Declarations\ConsumeOptions[].

    }

    /**
     * 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) {

The property executeReceiverStrategies does not exist on OldSound\RabbitMqBundle\RabbitMq\Consumer. Did you maybe forget to declare it?

                    $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();

The method getExecuteReceiverStrategy() does not exist on OldSound\RabbitMqBundle\RabbitMq\Consumer.

        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) {

The type OldSound\RabbitMqBundle\RabbitMq\BatchReceiverInterface was not found. Maybe you did not declare it correctly or list all dependencies?

                $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);

The method sendRpcReply() does not exist on OldSound\RabbitMqBundle\RabbitMq\Consumer.

                    $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);

It seems like $messages can also be of type false; however, parameter $messages of OldSound\RabbitMqBundle\RabbitMq\Consumer::handleProcessMessages() does only seem to accept PhpAmqpLib\Message\AMQPMessage[], maybe add an additional type check?

        }
    }

    /**
     * @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)) {

The expression $this->target of type integer|null is loosely compared to true; this is ambiguous if the integer can be 0. You might want to explicitly use !== null instead.

            $this->stopConsuming();
        }
    }

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

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

The property executeReceiverStrategies does not exist on OldSound\RabbitMqBundle\RabbitMq\Consumer. Did you maybe forget to declare it?

                $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;
    }
}