Skip to content

Commit

Permalink
Add queues arguments support
Browse files Browse the repository at this point in the history
  • Loading branch information
@olaurendeau
olaurendeau committed Feb 11, 2016
1 parent b609e70 commit fd044cd
Show file tree
Hide file tree
Showing 4 changed files with 243 additions and 2 deletions.
33 changes: 33 additions & 0 deletions DependencyInjection/Configuration.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@ private function getQueuesConfiguration()
->scalarNode('name')->end()
->scalarNode('durable')->defaultTrue()->end()
->scalarNode('modulus')->defaultNull()->end()
->append($this->getQueueArgumentsConfiguration())
->arrayNode('bindings')
->performNoDeepMerging()
->prototype('array')
Expand Down Expand Up @@ -185,4 +186,36 @@ private function appendNameNormalization(NodeDefinition $node)
})
->end();
}

/**
* @return NodeDefinition
*/
private function getQueueArgumentsConfiguration()
{
$node = new ArrayNodeDefinition('arguments');

return $node
// We have to un-normalize arguments keys, as this is not configurable in SF 2.1
->beforeNormalization()
->ifTrue(function($arguments) {
foreach ($arguments as $k => $v) {
//Un-normalize keys
if (false !== strpos($k, '_')) {
return true;
}
}
})
->then(function($arguments) {
foreach ($arguments as $k => $v) {
if (false !== strpos($k, '_')) {
$arguments[str_replace('_', '-', $k)] = $v;
unset($arguments[$k]);
}
}

return $arguments;
})
->end()
->prototype('scalar')->end();
}
}
31 changes: 30 additions & 1 deletion Manager/AbstractManager.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,37 @@ protected function handleNotFoundException(ClientErrorResponseException $e)
*/
protected function isUpToDate($configuration, $remoteConfiguration)
{
$diff = array_diff_assoc($configuration, $remoteConfiguration);
$diff = $this->array_diff_assoc_recursive($configuration, $remoteConfiguration);

foreach ($remoteConfiguration as $k => $v) {
if (!isset($configuration[$k])) {
unset($remoteConfiguration[$k]);
}
}

$diff = array_merge_recursive(
$diff,
$this->array_diff_assoc_recursive($remoteConfiguration, $configuration)
);

return empty($diff);
}

private function array_diff_assoc_recursive($array1, $array2) {
$difference=array();
foreach($array1 as $key => $value) {
if( is_array($value) ) {
if( !isset($array2[$key]) || !is_array($array2[$key]) ) {
$difference[$key] = $value;
} else {
$new_diff = $this->array_diff_assoc_recursive($value, $array2[$key]);
if( !empty($new_diff) )
$difference[$key] = $new_diff;
}
} else if( !array_key_exists($key,$array2) || $array2[$key] !== $value ) {
$difference[$key] = $value;
}
}
return $difference;
}
}
175 changes: 175 additions & 0 deletions Tests/DependencyInjection/ConfigurationTest.php
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,175 @@
<?php

namespace Ola\RabbitMqAdminToolkitBundle\Tests\DependencyInjection;

use Ola\RabbitMqAdminToolkitBundle\DependencyInjection\Configuration;
use Symfony\Component\Config\Definition\Processor;

class ConfigurationTest extends \PHPUnit_Framework_TestCase
{
public function test_basicConfiguration()
{
$this->assertEquals(
$this->getBasicExpectedConfiguration(),
$this->processConfiguration($this->getBasicConfiguration())
);
}

public function test_overridingConfiguration()
{
$expectedConfiguration = $this->getBasicExpectedConfiguration();
$expectedConfiguration['delete_allowed'] = true;
$expectedConfiguration['vhosts']['default']['exchanges']['exchange.c'] = array(
'name' => 'exchange.c',
'durable' => true,
'type' => 'topic'
);

$expectedConfiguration['vhosts']['default']['queues']['queue.a']['durable'] = false;
$expectedConfiguration['vhosts']['default']['queues']['queue.a']['bindings'] = array(
array(
'exchange' => 'exchange.c',
'routing_key' => 'b.#',
),
);

$configs = $this->getBasicConfiguration();
$configs[0]['vhosts']['default']['queues']['queue.b'] = array(
'bindings' => array(
array(
'exchange' => 'exchange.a',
'routing_key' => 'c.#',
),
)
);
$configs[] = array(
'delete_allowed' => true,
'vhosts' => array(
'default' => array(
'exchanges' => array(
'exchange.c' => NULL,
),
'queues' => array(
'queue.b' => false,
'queue.a' => array(
'durable' => false,
'bindings' => array(
array(
'exchange' => 'exchange.c',
'routing_key' => 'b.#',
),
),
),
),
),
),
);

$this->assertEquals(
$expectedConfiguration,
$this->processConfiguration($configs)
);
}

public function test_queueArgumentsNormalization()
{
$expectedConfiguration = $this->getBasicExpectedConfiguration();

$expectedConfiguration['vhosts']['default']['queues']['queue.a']['arguments'] = array(
'x-ha-policy' => 'all'
);

$configs = $this->getBasicConfiguration();
$configs[0]['vhosts']['default']['queues']['queue.a']['arguments'] = array(
'x-ha-policy' => 'all'
);

$this->assertEquals(
$expectedConfiguration,
$this->processConfiguration($configs)
);
}

private function processConfiguration($configs)
{
$processor = new Processor();

return $processor->processConfiguration(new Configuration(), $configs);
}

private function getBasicExpectedConfiguration()
{
return array(
'default_vhost' => 'default',
'delete_allowed' => false,
'silent_failure' => false,
'connections' => array(
'default' => 'http://user:password@localhost:15672',
),
'vhosts' => array(
'default' => array(
'connection' => 'default',
'permissions' => array(
'user' => array(
'configure' => '.*',
'read' => '.*',
'write' => '.*'
),
),
'exchanges' => array(
'exchange.a' => array(
'name' => 'exchange.a',
'durable' => true,
'type' => 'topic'
),
),
'queues' => array(
'queue.a' => array(
'name' => 'queue.a',
'durable' => true,
'modulus' => null,
'arguments' => array(),
'bindings' => array(
array(
'exchange' => 'exchange.a',
'routing_key' => 'a.#',
),
),
),
),
),
),
);
}

private function getBasicConfiguration()
{
return array(
array(
'connections' => array(
'default' => 'http://user:password@localhost:15672',
),
'vhosts' => array(
'default' => array(
'permissions' => array(
'user' => NULL,
),
'exchanges' => array(
'exchange.a' => NULL,
),
'queues' => array(
'queue.a' => array(
'bindings' => array(
array(
'exchange' => 'exchange.a',
'routing_key' => 'a.#',
),
),
),
),
),
),
)
);
}
}
6 changes: 5 additions & 1 deletion Tests/Manager/ExchangeManagerTest.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,17 +57,21 @@ public function test_define_update()
{
$this->configuration->getConfiguration('exchanges')->willReturn(array(
'foo' => array('name' => 'foo', 'durable' => true),
'bar' => array('name' => 'doe', 'durable' => true)
'bar' => array('name' => 'doe', 'durable' => true),
'moo' => array('name' => 'moo', 'durable' => true, 'arguments' => array('x-bar' => 12))
));

$this->configuration->isDeleteAllowed()->willReturn(true);

$this->exchanges->get('vhost', 'foo')->willReturn(array('durable' => true));
$this->exchanges->get('vhost', 'doe')->willReturn(array('durable' => false));
$this->exchanges->get('vhost', 'moo')->willReturn(array('durable' => true, 'arguments' => array('x-foo' => 19)));

$this->exchanges->create('vhost', 'foo', array('durable' => true))->shouldNotBeCalled();
$this->exchanges->delete('vhost', 'doe')->shouldBeCalled();
$this->exchanges->create('vhost', 'doe', array('durable' => true))->shouldBeCalled();
$this->exchanges->delete('vhost', 'moo')->shouldBeCalled();
$this->exchanges->create('vhost', 'moo', array('durable' => true, 'arguments' => array('x-bar' => 12)))->shouldBeCalled();

$this->exchangeManager->define($this->configuration->reveal());
}
Expand Down

0 comments on commit fd044cd

Please sign in to comment.