OldSound\RabbitMqBundle\Command\GroupConsumerCommand

Complexity

Total Complexity

10

Size/Duplication

Total Lines	85
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	1

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	14	1
A	configure()	0	15	1
B	execute()	0	46	8

<?php

namespace OldSound\RabbitMqBundle\Command;

use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer;
use OldSound\RabbitMqBundle\RabbitMq\Consumer;
use PhpAmqpLib\Connection\AbstractConnection;
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')
            ->addArgument('name', InputArgument::OPTIONAL, 'Consumer group name', 'default')
            ->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 ?! ->addOption('messages', 'm', InputOption::VALUE_OPTIONAL, 'Messages to consume', 0)
            // ->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')
            ->setDescription('Synchronous execute combined multiple consumers');
    }

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

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

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