OldSound\RabbitMqBundle\Command\GroupConsumerCommand::execute()

Complexity

Conditions	19
Paths	272

Size

Total Lines	82
Code Lines	43

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

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

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