Merge pull request #151 from paulkamer/master

Added optional AMQP-Messages Properties to publish method of the Producer (resolves merge conflicts of #55)

Author: videlalvaro

CHANGELOG

@@ -1,3 +1,6 @@
+- 2013-01-18
+    *  adds an an optional parameter for the AMQP Message Properties for the publish method of the Producer, so they can be set as well. For example, seeting the application_headers is now possible.
+
 - 2012-06-04
     * Revert PR #46. It is still possible to override parameter classes but in a proper way.
     * Some default options for exchanges declared in the "producers" config section

DependencyInjection/OldSoundRabbitMqExtension.php

@@ -35,7 +35,7 @@ public function load(array $configs, ContainerBuilder $container)
     {
         $this->container = $container;
 
-        $loader = new XmlFileLoader($this->container, new FileLocator(array(__DIR__.'/../Resources/config')));
+        $loader = new XmlFileLoader($this->container, new FileLocator(array(__DIR__ . '/../Resources/config')));
         $loader->load('rabbitmq.xml');
 
         $configuration = new Configuration();
@@ -90,7 +90,7 @@ protected function loadProducers()
                 $definition = new Definition($producer['class']);
                 $definition->addTag('old_sound_rabbit_mq.base_amqp');
                 $definition->addTag('old_sound_rabbit_mq.producer');
-                $definition->addMethodCall('setExchangeOptions', array($producer['exchange_options']));
+                $definition->addMethodCall('setExchangeOptions', array($this->normalizeArgumentKeys($producer['exchange_options'])));
                 //this producer doesn't define a queue
                 if (!isset($producer['queue_options'])) {
                     $producer['queue_options']['name'] = null;
@@ -121,8 +121,8 @@ protected function loadConsumers()
             $definition
                 ->addTag('old_sound_rabbit_mq.base_amqp')
                 ->addTag('old_sound_rabbit_mq.consumer')
-                ->addMethodCall('setExchangeOptions', array($consumer['exchange_options']))
-                ->addMethodCall('setQueueOptions', array($consumer['queue_options']))
+                ->addMethodCall('setExchangeOptions', array($this->normalizeArgumentKeys($consumer['exchange_options'])))
+                ->addMethodCall('setQueueOptions', array($this->normalizeArgumentKeys($consumer['queue_options'])))
                 ->addMethodCall('setCallback', array(array(new Reference($consumer['callback']), 'execute')));
 
             if (array_key_exists('qos_options', $consumer)) {
@@ -156,9 +156,8 @@ protected function loadAnonConsumers()
             $definition
                 ->addTag('old_sound_rabbit_mq.base_amqp')
                 ->addTag('old_sound_rabbit_mq.anon_consumer')
-                ->addMethodCall('setExchangeOptions', array($anon['exchange_options']))
-                ->addMethodCall('setCallback', array(array(new Reference($anon['callback']), 'execute')))
-            ;
+                ->addMethodCall('setExchangeOptions', array($this->normalizeArgumentKeys($anon['exchange_options'])))
+                ->addMethodCall('setCallback', array(array(new Reference($anon['callback']), 'execute')));
             $this->injectConnection($definition, $anon['connection']);
             if ($this->collectorEnabled) {
                 $this->injectLoggedChannel($definition, $key, $anon['connection']);
@@ -168,6 +167,59 @@ protected function loadAnonConsumers()
         }
     }
 
+    /**
+     * Symfony 2 converts '-' to '_' when defined in the configuration. This leads to problems when using x-ha-policy
+     * parameter. So we revert the change for right configurations.
+     *
+     * @param array $config
+     *
+     * @return array
+     */
+    private function normalizeArgumentKeys(array $config)
+    {
+        if (isset($config['arguments'])) {
+            $arguments = $config['arguments'];
+            // support for old configuration
+            if (is_string($arguments)) {
+                $arguments = $this->argumentsStringAsArray($arguments);
+            }
+
+            $newArguments = array();
+            foreach ($arguments as $key => $value) {
+                if (strstr($key, '_')) {
+                    $key = str_replace('_', '-', $key);
+                }
+                $newArguments[$key] = $value;
+            }
+            $config['arguments'] = $newArguments;
+        }
+        return $config;
+    }
+
+    /**
+     * Support for arguments provided as string. Support for old configuration files.
+     *
+     * @deprecated
+     * @param string $arguments
+     * @return array
+     */
+    private function argumentsStringAsArray($arguments)
+    {
+        $argumentsArray = array();
+
+        $argumentPairs = explode(',', $arguments);
+        foreach ($argumentPairs as $argument) {
+            $argumentPair = explode(':', $argument);
+            $type = 'S';
+            if (isset($argumentPair[2])) {
+                $type = $argumentPair[2];
+            }
+            $argumentsArray[$argumentPair[0]] = array($type, $argumentPair[1]);
+        }
+
+        return $argumentsArray;
+    }
+
     protected function loadRpcClients()
     {
         foreach ($this->config['rpc_clients'] as $key => $client) {
@@ -193,8 +245,7 @@ protected function loadRpcServers()
                 ->addTag('old_sound_rabbit_mq.base_amqp')
                 ->addTag('old_sound_rabbit_mq.rpc_server')
                 ->addMethodCall('initServer', array($key))
-                ->addMethodCall('setCallback', array(array(new Reference($server['callback']), 'execute')))
-            ;
+                ->addMethodCall('setCallback', array(array(new Reference($server['callback']), 'execute')));
             $this->injectConnection($definition, $server['connection']);
             if ($this->collectorEnabled) {
                 $this->injectLoggedChannel($definition, $key, $server['connection']);
@@ -206,12 +257,11 @@ protected function loadRpcServers()
 
     protected function injectLoggedChannel(Definition $definition, $name, $connectionName)
     {
-        $id = sprintf('old_sound_rabbit_mq.channel.%s', $name);
+        $id      = sprintf('old_sound_rabbit_mq.channel.%s', $name);
         $channel = new Definition('%old_sound_rabbit_mq.logged.channel.class%');
         $channel
             ->setPublic(false)
-            ->addTag('old_sound_rabbit_mq.logged_channel')
-        ;
+            ->addTag('old_sound_rabbit_mq.logged_channel');
         $this->injectConnection($channel, $connectionName);
 
         $this->container->setDefinition($id, $channel);

README.md

@@ -198,6 +198,8 @@ public function indexAction($name)
 
 As you can see, if in your configuration you have a producer called __upload\_picture__, then in the service container you will have a service called __old_sound_rabbit_mq.upload\_picture\_producer__.
 
+Besides the message itself, the `OldSound\RabbitMqBundle\RabbitMq\Producer#publish()` method also accepts an optional routing key parameter and an optional array of additional properties. The array of additional properties allows you to alter the properties with which an `PhpAmqpLib\Message\AMQPMessage` object gets constructed by default. This way, for example, you can change the application headers.
+
 You can use __setContentType__ and __setDeliveryMode__ methods in order to set the message content type and the message
 delivery mode respectively. Default values are __text/plain__ for content type and __2__ for delivery mode.
 

RabbitMq/BaseAmqp.php

@@ -14,7 +14,8 @@ abstract class BaseAmqp
     protected $queueDeclared = false;
     protected $routingKey = '';
     protected $autoSetupFabric = true;
-
+    protected $basicProperties = array('content_type' => 'text/plain', 'delivery_mode' => 2);
+	
     protected $exchangeOptions = array(
         'passive' => false,
         'durable' => true,

RabbitMq/Producer.php

@@ -5,6 +5,9 @@
 use OldSound\RabbitMqBundle\RabbitMq\BaseAmqp;
 use PhpAmqpLib\Message\AMQPMessage;
 
+/**
+ * Prodcuer, that publishes AMQP Messages
+ */
 class Producer extends BaseAmqp
 {
     protected $contentType = 'text/plain';
@@ -24,13 +27,25 @@ public function setDeliveryMode($deliveryMode)
         return $this;
     }
 
-    public function publish($msgBody, $routingKey = '')
+    protected function getBasicProperties()
+    {
+        return array('content_type' => $this->contentType, 'delivery_mode' => $this->deliveryMode);
+    }
+
+    /**
+     * Publishes the message and merges additional properties with basic properties
+     *
+     * @param string $msgBody
+     * @param string $routingKey
+     * @param array $additionalProperties
+     */
+    public function publish($msgBody, $routingKey = '', $additionalProperties = array())
     {
         if ($this->autoSetupFabric) {
             $this->setupFabric();
         }
 
-        $msg = new AMQPMessage($msgBody, array('content_type' => $this->contentType, 'delivery_mode' => $this->deliveryMode));
-        $this->getChannel()->basic_publish($msg, $this->exchangeOptions['name'], $routingKey);
+        $msg = new AMQPMessage((string) $msgBody, array_merge($this->getBasicProperties(), $additionalProperties));
+        $this->ch->basic_publish($msg, $this->exchangeOptions['name'], (string) $routingKey);
     }
 }