OldSound\RabbitMqBundle\Command\GroupConsumerCommand

Complexity

Total Complexity

21

Size/Duplication

Total Lines	135
Duplicated Lines	0 %

Test Coverage

Coverage

0%

Importance

Changes	1
Bugs	0	Features	1

3 Methods

Rating	Name	Duplication	Size	Complexity
F	execute()	0	84	19
A	__construct()	0	14	1
A	configure()	0	17	1

<?php

namespace OldSound\RabbitMqBundle\Command;

use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer;
use OldSound\RabbitMqBundle\RabbitMq\Consumer;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Exception\InvalidArgumentException;
use Symfony\Component\Console\Exception\InvalidOptionException;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;

class GroupConsumerCommand extends BaseRabbitMqCommand
{
    /** @var iterable|Consumer[][] */
    protected $consumers;

    public function __construct(
        iterable $consumers,
        iterable $multiConsumers,
        iterable $dynamicConsumers,
        iterable $batchConsumers,
        iterable $anonConsumers
    ) {
        parent::__construct();
        $this->consumers = [

It seems like array($consumers, $multiConsumers, $dynamicConsumers, $batchConsumers, $anonConsumers) of type array<integer,iterable> is incompatible with the declared type array<mixed,OldSound\RabbitMqBundle\RabbitMq\Consumer[]>|iterable of property $consumers.

            $consumers,
            $multiConsumers,
            $dynamicConsumers,
            $batchConsumers,
            $anonConsumers
        ];
    }

    protected function configure()
    {
        $this
            ->setName('rabbitmq:group:consumer')
            ->setDescription('Synchronous execute combined multiple consumers')
            ->addArgument('name', InputArgument::OPTIONAL, 'Consumer group name', 'default')
            // TODO ?! ->addOption('messages', 'm', InputOption::VALUE_OPTIONAL, 'Messages to consume', 0)
            // ->addOption('route', 'r', InputOption::VALUE_OPTIONAL, 'Routing Key', '')
            ->addOption('memory-limit', 'l', InputOption::VALUE_OPTIONAL, 'Allowed memory for this process (MB)', null)
            ->addOption('debug', 'd', InputOption::VALUE_NONE, 'Enable Debugging')
            ->addOption('without-signals', 'w', InputOption::VALUE_NONE, 'Disable catching of system signals')

            ->addOption('connection', 'c', InputOption::VALUE_OPTIONAL, 'Rabbitmq connection name', 'default')
            ->addOption('timeout', 't', InputOption::VALUE_OPTIONAL, 'Timeout', 500)
            ->addOption('qos-prefetch-size', null, InputOption::VALUE_OPTIONAL, 'Qos prefetch size', 0)
            ->addOption('qos-prefetch-count', null,InputOption::VALUE_OPTIONAL, 'Qos prefetch count', 10)
            ->addOption('qos-global', null,InputOption::VALUE_OPTIONAL, 'Qos global', true)
        ;
    }

    /**
     * @todo remove duplicate code from BaseConsumerCommand
     * extands BaseConsumerCommand, split code to separate GroupConsumer class
     *
     * @param InputInterface $input
     * @param OutputInterface $output
     * @return int
     * @throws \ErrorException
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $group = $input->getArgument('name');

        /** @var BaseConsumer[] $consumers */
        $consumers = [];

        foreach ($this->consumers as $consumersByType) {
            foreach ($consumersByType as $consumer) {
                if (in_array($group, $consumer->getGroups(), true)) {
                    $consumers[] = $consumer;
                }
            }
        }

        if ([] === $consumers) {
            throw new InvalidArgumentException(sprintf('No consumers with %s group', $group));

It seems like $group can also be of type string[]; however, parameter $args of sprintf() does only seem to accept string, maybe add an additional type check?

        }

        $connectionName = $input->getOption('connection');
        $connectionServiceId = sprintf('old_sound_rabbit_mq.connection.%s', $connectionName);
        if (false === $this->container->has($connectionServiceId)) {
            throw new InvalidOptionException(sprintf('Rabbitmq connection %s is not defined', $connectionName));
        };


        if (defined('AMQP_WITHOUT_SIGNALS') === false) {
            define('AMQP_WITHOUT_SIGNALS', $input->getOption('without-signals'));
        }

        if (!AMQP_WITHOUT_SIGNALS && extension_loaded('pcntl')) {
            if (!function_exists('pcntl_signal')) {
                throw new \BadFunctionCallException("Function 'pcntl_signal' is referenced in the php.ini 'disable_functions' and can't be called.");
            }

            $stopConsumer = function () use ($consumers) {
                foreach ($consumers as $consumer) {
                    $consumer->forceStopConsumer();

                    // Halt consumer if waiting for a new message from the queue
                    try {
                        $consumer->stopConsuming();
                    } catch (AMQPTimeoutException $e) {}

Consider adding a comment why this CATCH block is empty.

                }
            };

            pcntl_signal(SIGTERM, $stopConsumer);
            pcntl_signal(SIGINT, $stopConsumer);
            pcntl_signal(SIGHUP, $stopConsumer);
        }

        if (defined('AMQP_DEBUG') === false) {
            define('AMQP_DEBUG', (bool) $input->getOption('debug'));
        }



        /** @var AbstractConnection $connection */
        $connection = $this->container->get($connectionServiceId);
        $channel = $connection->channel();

        $channel->basic_qos(
            $input->getOption('qos-prefetch-size'),
            $input->getOption('qos-prefetch-count'),
            $input->getOption('qos-global')
        );

        if (!is_null($input->getOption('memory-limit')) && ctype_digit((string) $input->getOption('memory-limit')) && $input->getOption('memory-limit') > 0) {
            foreach ($consumers as $i => $consumer) {
                $consumer->setMemoryLimit($input->getOption('memory-limit'));
            }
        }

        foreach ($consumers as $i => $consumer) {
            $consumer->setChannel($channel);
            $consumer->setConsumerTag(sprintf("PHPPROCESS_%s_%s_%s", gethostname(), getmypid(), $i));
            $consumer->setupConsumer();
        }
        
        while (count($channel->callbacks)) {
            $channel->wait(null, false, $input->getOption('timeout'));
        }

        return 0;
    }
}