Merge pull request #7 from bitheater/feature/allow-lazy-connections

Allow lazy connections
BedrockStreaming · Dec 13, 2014 · 4a1c2cd · 4a1c2cd
2 parents 5acbb72 + 425b476
commit 4a1c2cd
Show file tree

Hide file tree

Showing 6 changed files with 55 additions and 5 deletions.
diff --git a/README.md b/README.md
@@ -72,6 +72,7 @@ m6_web_amqp:
             login:     'guest'     # optional - default 'guest'
             password: 'guest'      # optional - default 'guest'
             vhost:    '/'          # optional - default '/'
+            lazy:     false        # optional - default false
     producers:
         myproducer:
             class: "My\\Provider\\Class"                           # optional - to overload the default provider class
@@ -147,9 +148,15 @@ The "flags" argument of getMessage accepts MQP_AUTOACK (auto acknowledge by defa
 To manually acknowledge a message, use the consumer's ackMessage/nackMessage methods with a delivery_tag argument's value from the AMQPEnvelope object. 
 If you choose to not acknowledge the message, the second parameter of nackMessage accepts AMQP_REQUEUE to requeue the message or AMQP_NOPARAM to forget it. 
 
+### Lazy connections
+
+It's highly recommended to set all connections to ```lazy: true``` in the configuration file. It'll prevent the bundle from connecting to RabbitMQ on each request.
+
+If you want lazy connections, you have to add ```"ocramius/proxy-manager": "~0.5"``` to your composer.json file, and (as said before) add ```lazy: true``` to your connections.
+
 ### DataCollector
 
-DataCollector is enabled by defaut if kernel.debug is set. Typically in the dev environment.
+DataCollector is enabled by default if kernel.debug is set. Typically in the dev environment.
 
 # Unit tests :
 

diff --git a/composer.json b/composer.json
@@ -17,5 +17,8 @@
         "atoum/atoum-bundle" : "dev-master",
         "m6web/coke" : "~1.2",
         "m6web/symfony2-coding-standard" : "~1.1"
+    },
+    "suggest": {
+        "ocramius/proxy-manager": "Required for lazy connections"
     }
 }
diff --git a/src/AmqpBundle/DependencyInjection/Configuration.php b/src/AmqpBundle/DependencyInjection/Configuration.php
@@ -48,6 +48,7 @@ protected function addConnections(ArrayNodeDefinition $node)
                             ->scalarNode('login')->defaultValue('guest')->end()
                             ->scalarNode('password')->defaultValue('guest')->end()
                             ->scalarNode('vhost')->defaultValue('/')->end()
+                            ->booleanNode('lazy')->defaultFalse()->end()
                         ->end()
                     ->end()
                 ->end()

diff --git a/src/AmqpBundle/DependencyInjection/M6WebAmqpExtension.php b/src/AmqpBundle/DependencyInjection/M6WebAmqpExtension.php
@@ -50,8 +50,11 @@ protected function loadConnections(ContainerBuilder $container, array $config)
                                 ->addMethodCall('setReadTimeout', [$connection['timeout']])
                                 ->addMethodCall('setLogin', [$connection['login']])
                                 ->addMethodCall('setPassword', [$connection['password']])
-                                ->addMethodCall('setVhost', [$connection['vhost']])
-                                ->addMethodCall('connect');
+                                ->addMethodCall('setVhost', [$connection['vhost']]);
+
+            if (!$connection['lazy']) {
+                $connexionDefinition->addMethodCall('connect');
+            }
 
             $container->setDefinition(
                 sprintf('m6_web_amqp.connection.%s', $key),
@@ -67,13 +70,16 @@ protected function loadConnections(ContainerBuilder $container, array $config)
     protected function loadProducers(ContainerBuilder $container, array $config)
     {
         foreach ($config['producers'] as $key => $producer) {
+            $lazy = $config['connections'][$producer['connection']]['lazy'];
+
             // Create the producer with the factory
             $producerDefinition = new Definition(
                 $producer['class'],
                 [
                     $producer['class'],
                     new Reference(sprintf('m6_web_amqp.connection.%s', $producer['connection'])),
                     $producer['exchange_options'],
+                    $lazy,
                 ]
             );
 
@@ -90,6 +96,14 @@ protected function loadProducers(ContainerBuilder $container, array $config)
             $producerDefinition->setFactoryService('m6_web_amqp.producer_factory')
                                ->setFactoryMethod('get');
 
+            if ($lazy) {
+                if (!method_exists($producerDefinition, 'setLazy')) {
+                    throw new \InvalidArgumentException('It\'s not possible to declare a service as lazy. Are you using Symfony 2.3?');
+                }
+
+                $producerDefinition->setLazy(true);
+            }
+
             $container->setDefinition(
                 sprintf('m6_web_amqp.producer.%s', $key),
                 $producerDefinition
@@ -104,6 +118,8 @@ protected function loadProducers(ContainerBuilder $container, array $config)
     protected function loadConsumers(ContainerBuilder $container, array $config)
     {
         foreach ($config['consumers'] as $key => $consumer) {
+            $lazy = $config['connections'][$consumer['connection']]['lazy'];
+
             // Create the consumer with the factory
             $consumerDefinition = new Definition(
                 $consumer['class'],
@@ -112,6 +128,7 @@ protected function loadConsumers(ContainerBuilder $container, array $config)
                     new Reference(sprintf('m6_web_amqp.connection.%s', $consumer['connection'])),
                     $consumer['exchange_options'],
                     $consumer['queue_options'],
+                    $lazy,
                 ]
             );
 
@@ -128,6 +145,14 @@ protected function loadConsumers(ContainerBuilder $container, array $config)
             $consumerDefinition->setFactoryService('m6_web_amqp.consumer_factory')
                                ->setFactoryMethod('get');
 
+            if ($lazy) {
+                if (!method_exists($consumerDefinition, 'setLazy')) {
+                    throw new \InvalidArgumentException('It\'s not possible to declare a service as lazy. Are you using Symfony 2.3?');
+                }
+
+                $consumerDefinition->setLazy(true);
+            }
+
             $container->setDefinition(
                 sprintf('m6_web_amqp.consumer.%s', $key),
                 $consumerDefinition

diff --git a/src/AmqpBundle/Factory/ConsumerFactory.php b/src/AmqpBundle/Factory/ConsumerFactory.php
@@ -55,17 +55,24 @@ public function __construct($channelClass, $queueClass)
      * @param string $connexion       AMQP connexion
      * @param array  $exchangeOptions Exchange Options
      * @param array  $queueOptions    Queue Options
+     * @param bool   $lazy            Specifies if it should connect
      *
      * @return Consumer
      */
-    public function get($class, $connexion, array $exchangeOptions, array $queueOptions)
+    public function get($class, $connexion, array $exchangeOptions, array $queueOptions, $lazy = false)
     {
         if (!class_exists($class)) {
             throw new \InvalidArgumentException(
                 sprintf("Consumer class '%s' doesn't exist", $class)
             );
         }
 
+        if ($lazy) {
+            if (!$connexion->isConnected()) {
+                $connexion->connect();
+            }
+        }
+
         // Open a new channel
         $channel = new $this->channelClass($connexion);
 

diff --git a/src/AmqpBundle/Factory/ProducerFactory.php b/src/AmqpBundle/Factory/ProducerFactory.php
@@ -54,17 +54,24 @@ public function __construct($channelClass, $exchangeClass)
      * @param string $class           Provider class name
      * @param string $connexion       AMQP connexion
      * @param array  $exchangeOptions Exchange Options
+     * @param bool   $lazy            Specifies if it should connect
      *
      * @return Producer
      */
-    public function get($class, $connexion, array $exchangeOptions)
+    public function get($class, $connexion, array $exchangeOptions, $lazy = false)
     {
         if (!class_exists($class)) {
             throw new \InvalidArgumentException(
                 sprintf("Producer class '%s' doesn't exist", $class)
             );
         }
 
+        if ($lazy) {
+            if (!$connexion->isConnected()) {
+                $connexion->connect();
+            }
+        }
+
         // Open a new channel
         $channel = new $this->channelClass($connexion);