WorkerCompilerPass::setupBeanstalkd() - Code Metrics - Inspection of "Update beanstalk, rabbitmq, many minor fixes, incl..." - mmucklo/DtcQueueBundle - Measure and Improve Code Quality continuously with Scrutinizer

Test Setup Failed

Pull Request — master (#7)

by Matthew

created 2017-08-17 16:45 UTC

WorkerCompilerPass::setupBeanstalkd() A

↳ Parent: WorkerCompilerPass

Complexity

Conditions	3
Paths	3

Size

Total Lines	12
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	12
rs	9.4285
c	0
b	0
f	0
cc	3
eloc	8
nc	3
nop	1

<?php

namespace Dtc\QueueBundle\DependencyInjection\Compiler;

use Dtc\QueueBundle\Model\Job;
use Pheanstalk\Pheanstalk;
use Symfony\Component\DependencyInjection\Alias;
use Symfony\Component\DependencyInjection\ContainerBuilder;
use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
use Symfony\Component\DependencyInjection\Definition;
use Symfony\Component\DependencyInjection\Reference;

class WorkerCompilerPass implements CompilerPassInterface
{
    public function process(ContainerBuilder $container)
    {
        if (false === $container->hasDefinition('dtc_queue.worker_manager')) {
            return;
        }

        $defaultManagerType = $container->getParameter('dtc_queue.default_manager');
        if (!$container->hasDefinition('dtc_queue.job_manager.'.$defaultManagerType)) {
            throw new \Exception("No job manager found for dtc_queue.job_manager.$defaultManagerType");
        }

        $alias = new Alias('dtc_queue.job_manager.'.$defaultManagerType);
        $container->setAlias('dtc_queue.job_manager', $alias);

        // Setup beanstalkd if configuration is present
        $this->setupBeanstalkd($container);
        $this->setupRabbitMQ($container);

        $definition = $container->getDefinition('dtc_queue.worker_manager');
        $jobManagerRef = array(new Reference('dtc_queue.job_manager'));

        $jobClass = $this->getJobClass($container);

        // Add each worker to workerManager, make sure each worker has instance to work
        foreach ($container->findTaggedServiceIds('dtc_queue.worker') as $id => $attributes) {
            $worker = $container->getDefinition($id);
            $class = $container->getDefinition($id)->getClass();

            $refClass = new \ReflectionClass($class);
            $workerClass = 'Dtc\QueueBundle\Model\Worker';
            if (!$refClass->isSubclassOf($workerClass)) {
                throw new \InvalidArgumentException(sprintf('Service "%s" must extend class "%s".', $id, $workerClass));
            }

            // Give each worker access to job manager
            $worker->addMethodCall('setJobManager', $jobManagerRef);
            $worker->addMethodCall('setJobClass', array($jobClass));

            $definition->addMethodCall('addWorker', array(new Reference($id)));
        }

        $eventDispatcher = $container->getDefinition('dtc_queue.event_dispatcher');
        foreach ($container->findTaggedServiceIds('dtc_queue.event_subscriber') as $id => $attributes) {
            $eventSubscriber = $container->getDefinition($id);
            $eventDispatcher->addMethodCall('addSubscriber', [$eventSubscriber]);
        }
    }

    /**
     * Sets up beanstalkd instance if appropriate.
     *
     * @param ContainerBuilder $container
     */
    public function setupBeanstalkd(ContainerBuilder $container)
    {
        if ($container->hasParameter('dtc_queue.beanstalkd.host')) {
            $definition = new Definition('Pheanstalk\\Pheanstalk', [$container->getParameter('dtc_queue.beanstalkd.host')]);
            $container->setDefinition('dtc_queue.beanstalkd', $definition);
            $definition = $container->getDefinition('dtc_queue.job_manager.beanstalkd');
            $definition->addMethodCall('setBeanstalkd', [new Reference('dtc_queue.beanstalkd')]);
            if ($container->hasParameter('dtc_queue.beanstalkd.tube')) {
                $definition->addMethodCall('setTube', [$container->getParameter('dtc_queue.beanstalkd.tube')]);
            }
        }
    }

    /**
     * Sets up RabbitMQ instance if appropriate.
     *
     * @param ContainerBuilder $container
     */
    public function setupRabbitMQ(ContainerBuilder $container)
    {
        if ($container->hasParameter('dtc_queue.rabbit_mq')) {
            $class = 'PhpAmqpLib\\Connection\\AMQPStreamConnection';
            $rabbitMqConfig = $container->getParameter('dtc_queue.rabbit_mq');
            $arguments = [
                $rabbitMqConfig['host'],
                $rabbitMqConfig['port'],
                $rabbitMqConfig['user'],
                $rabbitMqConfig['password'],
                $rabbitMqConfig['vhost'],
            ];

            if ($container->hasParameter('dtc_queue.rabbit_mq.ssl') && $container->getParameter('dtc_queue.rabbit_mq.ssl')) {
                $class = 'PhpAmqpLib\\Connection\\AMQPSSLConnection';
                if ($container->hasParameter('dtc_queue.rabbit_mq.ssl_options')) {
                    $arguments[] = $container->getParameter('dtc_queue.rabbit_mq.ssl_options');
                } else {
                    $arguments[] = [];
                }
                if ($container->hasParameter('dtc_queue.rabbit_mq.options')) {
                    $arguments[] = $container->getParameter('dtc_queue.rabbit_mq.options');
                }
            } else {
                if ($container->hasParameter('dtc_queue.rabbit_mq.options')) {
                    $options = $container->getParameter('dtc_queue.rabbit_mq.options');
                    if (isset($options['insist'])) {
                        $arguments[] = $options['insist'];
                    } else {
                        $arguments[] = false;
                    }
                    if (isset($options['login_method'])) {
                        $arguments[] = $options['login_method'];
                    } else {
                        $arguments[] = 'AMQPLAIN';
                    }
                    if (isset($options['login_response'])) {
                        $arguments[] = $options['login_response'];
                    } else {
                        $arguments[] = null;
                    }
                    if (isset($options['locale'])) {
                        $arguments[] = $options['locale'];
                    } else {
                        $arguments[] = 'en_US';
                    }
                    if (isset($options['connection_timeout'])) {
                        $arguments[] = $options['connection_timeout'];
                    } else {
                        $arguments[] = 3.0;
                    }
                    if (isset($options['read_write_timeout'])) {
                        $arguments[] = $options['read_write_timeout'];
                    } else {
                        $arguments[] = 3.0;
                    }
                    if (isset($options['context'])) {
                        $arguments[] = $options['context'];
                    } else {
                        $arguments[] = null;
                    }
                    if (isset($options['keepalive'])) {
                        $arguments[] = $options['keepalive'];
                    } else {
                        $arguments[] = false;
                    }
                    if (isset($options['heartbeat'])) {
                        $arguments[] = $options['heartbeat'];
                    } else {
                        $arguments[] = 0;
                    }
                }
            }
            $definition = new Definition($class, $arguments);
            $container->setDefinition('dtc_queue.rabbit_mq', $definition);
            $definition = $container->getDefinition('dtc_queue.job_manager.rabbit_mq');
            $definition->addMethodCall('setAMQPConnection', [new Reference('dtc_queue.rabbit_mq')]);
            $definition->addMethodCall('setQueueArgs', array_values($rabbitMqConfig['queue_args']));
            $definition->addMethodCall('setExchangeArgs', array_values($rabbitMqConfig['exchange_args']));
        }
    }

    /**
     * Determines the job class based on teh queue manager type.
     *
     * @param ContainerBuilder $container
     *
     * @return mixed|string
     *
     * @throws \Exception
     */
    public function getJobClass(ContainerBuilder $container)
    {
        $jobClass = $container->getParameter('dtc_queue.job_class');
        if (!$jobClass) {
            switch ($defaultType = $container->getParameter('dtc_queue.default_manager')) {
                case 'mongodb':
                    $jobClass = 'Dtc\\QueueBundle\\Documents\\Job';
                    break;
                case 'beanstalkd':
                    $jobClass = 'Dtc\\QueueBundle\\Beanstalkd\\Job';
                    break;
                case 'rabbit_mq':
                    $jobClass = 'Dtc\\QueueBundle\\Model\\Job';
                    break;
                default:
                    throw new \Exception("Unknown type $defaultType - please specify a Job class in the 'class' configuration parameter");
            }
        }

        if (!class_exists($jobClass)) {
            throw new \Exception("Can't find Job class $jobClass");
        }

        return $jobClass;
    }
}


1			<?php
2
3			namespace Dtc\QueueBundle\DependencyInjection\Compiler;
4
5			use Dtc\QueueBundle\Model\Job;
6			use Pheanstalk\Pheanstalk;
7			use Symfony\Component\DependencyInjection\Alias;
8			use Symfony\Component\DependencyInjection\ContainerBuilder;
9			use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
10			use Symfony\Component\DependencyInjection\Definition;
11			use Symfony\Component\DependencyInjection\Reference;
12
13			class WorkerCompilerPass implements CompilerPassInterface
14			{
15			public function process(ContainerBuilder $container)
16			{
17			if (false === $container->hasDefinition('dtc_queue.worker_manager')) {
18			return;
19			}
20
21			$defaultManagerType = $container->getParameter('dtc_queue.default_manager');
22			if (!$container->hasDefinition('dtc_queue.job_manager.'.$defaultManagerType)) {
23			throw new \Exception("No job manager found for dtc_queue.job_manager.$defaultManagerType");
24			}
25
26			$alias = new Alias('dtc_queue.job_manager.'.$defaultManagerType);
27			$container->setAlias('dtc_queue.job_manager', $alias);
28
29			// Setup beanstalkd if configuration is present
30			$this->setupBeanstalkd($container);
31			$this->setupRabbitMQ($container);
32
33			$definition = $container->getDefinition('dtc_queue.worker_manager');
34			$jobManagerRef = array(new Reference('dtc_queue.job_manager'));
35
36			$jobClass = $this->getJobClass($container);
37
38			// Add each worker to workerManager, make sure each worker has instance to work
39			foreach ($container->findTaggedServiceIds('dtc_queue.worker') as $id => $attributes) {
40			$worker = $container->getDefinition($id);
41			$class = $container->getDefinition($id)->getClass();
42
43			$refClass = new \ReflectionClass($class);
44			$workerClass = 'Dtc\QueueBundle\Model\Worker';
45			if (!$refClass->isSubclassOf($workerClass)) {
46			throw new \InvalidArgumentException(sprintf('Service "%s" must extend class "%s".', $id, $workerClass));
47			}
48
49			// Give each worker access to job manager
50			$worker->addMethodCall('setJobManager', $jobManagerRef);
51			$worker->addMethodCall('setJobClass', array($jobClass));
52
53			$definition->addMethodCall('addWorker', array(new Reference($id)));
54			}
55
56			$eventDispatcher = $container->getDefinition('dtc_queue.event_dispatcher');
57			foreach ($container->findTaggedServiceIds('dtc_queue.event_subscriber') as $id => $attributes) {
58			$eventSubscriber = $container->getDefinition($id);
59			$eventDispatcher->addMethodCall('addSubscriber', [$eventSubscriber]);
60			}
61			}
62
63			/**
64			* Sets up beanstalkd instance if appropriate.
65			*
66			* @param ContainerBuilder $container
67			*/
68			public function setupBeanstalkd(ContainerBuilder $container)
69			{
70			if ($container->hasParameter('dtc_queue.beanstalkd.host')) {
71			$definition = new Definition('Pheanstalk\\Pheanstalk', [$container->getParameter('dtc_queue.beanstalkd.host')]);
72			$container->setDefinition('dtc_queue.beanstalkd', $definition);
73			$definition = $container->getDefinition('dtc_queue.job_manager.beanstalkd');
74			$definition->addMethodCall('setBeanstalkd', [new Reference('dtc_queue.beanstalkd')]);
75			if ($container->hasParameter('dtc_queue.beanstalkd.tube')) {
76			$definition->addMethodCall('setTube', [$container->getParameter('dtc_queue.beanstalkd.tube')]);
77			}
78			}
79			}
80
81			/**
82			* Sets up RabbitMQ instance if appropriate.
83			*
84			* @param ContainerBuilder $container
85			*/
86			public function setupRabbitMQ(ContainerBuilder $container)
87			{
88			if ($container->hasParameter('dtc_queue.rabbit_mq')) {
89			$class = 'PhpAmqpLib\\Connection\\AMQPStreamConnection';
90			$rabbitMqConfig = $container->getParameter('dtc_queue.rabbit_mq');
91			$arguments = [
92			$rabbitMqConfig['host'],
93			$rabbitMqConfig['port'],
94			$rabbitMqConfig['user'],
95			$rabbitMqConfig['password'],
96			$rabbitMqConfig['vhost'],
97			];
98
99			if ($container->hasParameter('dtc_queue.rabbit_mq.ssl') && $container->getParameter('dtc_queue.rabbit_mq.ssl')) {
100			$class = 'PhpAmqpLib\\Connection\\AMQPSSLConnection';
101			if ($container->hasParameter('dtc_queue.rabbit_mq.ssl_options')) {
102			$arguments[] = $container->getParameter('dtc_queue.rabbit_mq.ssl_options');
103			} else {
104			$arguments[] = [];
105			}
106			if ($container->hasParameter('dtc_queue.rabbit_mq.options')) {
107			$arguments[] = $container->getParameter('dtc_queue.rabbit_mq.options');
108			}
109			} else {
110			if ($container->hasParameter('dtc_queue.rabbit_mq.options')) {
111			$options = $container->getParameter('dtc_queue.rabbit_mq.options');
112			if (isset($options['insist'])) {
113			$arguments[] = $options['insist'];
114			} else {
115			$arguments[] = false;
116			}
117			if (isset($options['login_method'])) {
118			$arguments[] = $options['login_method'];
119			} else {
120			$arguments[] = 'AMQPLAIN';
121			}
122			if (isset($options['login_response'])) {
123			$arguments[] = $options['login_response'];
124			} else {
125			$arguments[] = null;
126			}
127			if (isset($options['locale'])) {
128			$arguments[] = $options['locale'];
129			} else {
130			$arguments[] = 'en_US';
131			}
132			if (isset($options['connection_timeout'])) {
133			$arguments[] = $options['connection_timeout'];
134			} else {
135			$arguments[] = 3.0;
136			}
137			if (isset($options['read_write_timeout'])) {
138			$arguments[] = $options['read_write_timeout'];
139			} else {
140			$arguments[] = 3.0;
141			}
142			if (isset($options['context'])) {
143			$arguments[] = $options['context'];
144			} else {
145			$arguments[] = null;
146			}
147			if (isset($options['keepalive'])) {
148			$arguments[] = $options['keepalive'];
149			} else {
150			$arguments[] = false;
151			}
152			if (isset($options['heartbeat'])) {
153			$arguments[] = $options['heartbeat'];
154			} else {
155			$arguments[] = 0;
156			}
157			}
158			}
159			$definition = new Definition($class, $arguments);
160			$container->setDefinition('dtc_queue.rabbit_mq', $definition);
161			$definition = $container->getDefinition('dtc_queue.job_manager.rabbit_mq');
162			$definition->addMethodCall('setAMQPConnection', [new Reference('dtc_queue.rabbit_mq')]);
163			$definition->addMethodCall('setQueueArgs', array_values($rabbitMqConfig['queue_args']));
164			$definition->addMethodCall('setExchangeArgs', array_values($rabbitMqConfig['exchange_args']));
165			}
166			}
167
168			/**
169			* Determines the job class based on teh queue manager type.
170			*
171			* @param ContainerBuilder $container
172			*
173			* @return mixed\|string
174			*
175			* @throws \Exception
176			*/
177			public function getJobClass(ContainerBuilder $container)
178			{
179			$jobClass = $container->getParameter('dtc_queue.job_class');
180			if (!$jobClass) {
181			switch ($defaultType = $container->getParameter('dtc_queue.default_manager')) {
182			case 'mongodb':
183			$jobClass = 'Dtc\\QueueBundle\\Documents\\Job';
184			break;
185			case 'beanstalkd':
186			$jobClass = 'Dtc\\QueueBundle\\Beanstalkd\\Job';
187			break;
188			case 'rabbit_mq':
189			$jobClass = 'Dtc\\QueueBundle\\Model\\Job';
190			break;
191			default:
192			throw new \Exception("Unknown type $defaultType - please specify a Job class in the 'class' configuration parameter");
193			}
194			}
195
196			if (!class_exists($jobClass)) {
197			throw new \Exception("Can't find Job class $jobClass");
198			}
199
200			return $jobClass;
201			}
202			}
203

mmucklo / DtcQueueBundle

Pull Request — master (#7)

WorkerCompilerPass::setupBeanstalkd() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like