Skip to content

Commit 057aeda

Browse files
makasimmakasim
authored
Merge pull request #572 from php-enqueue/bundle-multi-transport-configuration
Bundle multi transport configuration
2 parents 7c26ed1 + 181740d commit 057aeda

15 files changed

+395
-156
lines changed

‎pkg/enqueue-bundle/DependencyInjection/Configuration.php

+49-4
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
use Enqueue\Symfony\Client\DependencyInjection\ClientFactory;
66
use Enqueue\Symfony\DependencyInjection\TransportFactory;
7+
use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
78
use Symfony\Component\Config\Definition\Builder\TreeBuilder;
89
use Symfony\Component\Config\Definition\ConfigurationInterface;
910

@@ -22,14 +23,58 @@ public function getConfigTreeBuilder(): TreeBuilder
2223
$rootNode = $tb->root('enqueue');
2324
$rootNode
2425
->beforeNormalization()
25-
->ifEmpty()->then(function () {
26-
return ['transport' => ['dsn' => 'null:']];
27-
});
26+
->always(function ($value) {
27+
if (empty($value)) {
28+
return [
29+
'transport' => [
30+
'default' => [
31+
'dsn' => 'null:',
32+
],
33+
],
34+
];
35+
}
36+
37+
if (is_string($value)) {
38+
return [
39+
'transport' => [
40+
'default' => [
41+
'dsn' => $value,
42+
],
43+
],
44+
];
45+
}
46+
47+
return $value;
48+
})
49+
;
2850

2951
$transportFactory = new TransportFactory('default');
3052

53+
/** @var ArrayNodeDefinition $transportNode */
3154
$transportNode = $rootNode->children()->arrayNode('transport');
32-
$transportFactory->addTransportConfiguration($transportNode);
55+
$transportNode
56+
->beforeNormalization()
57+
->always(function ($value) {
58+
if (empty($value)) {
59+
return ['default' => ['dsn' => 'null:']];
60+
}
61+
if (is_string($value)) {
62+
return ['default' => ['dsn' => $value]];
63+
}
64+
65+
if (is_array($value) && array_key_exists('dsn', $value)) {
66+
return ['default' => $value];
67+
}
68+
69+
return $value;
70+
});
71+
$transportPrototypeNode = $transportNode
72+
->requiresAtLeastOneElement()
73+
->useAttributeAsKey('key')
74+
->prototype('array')
75+
;
76+
77+
$transportFactory->addTransportConfiguration($transportPrototypeNode);
3378

3479
$consumptionNode = $rootNode->children()->arrayNode('consumption');
3580
$transportFactory->addQueueConsumerConfiguration($consumptionNode);

‎pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php

+11-7
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,15 @@ public function load(array $configs, ContainerBuilder $container): void
2525
$loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
2626
$loader->load('services.yml');
2727

28-
$transportFactory = (new TransportFactory('default'));
29-
$transportFactory->buildConnectionFactory($container, $config['transport']);
30-
$transportFactory->buildContext($container, []);
31-
$transportFactory->buildQueueConsumer($container, $config['consumption']);
32-
$transportFactory->buildRpcClient($container, []);
28+
foreach ($config['transport'] as $name => $transportConfig) {
29+
$transportFactory = (new TransportFactory($name));
30+
$transportFactory->buildConnectionFactory($container, $transportConfig);
31+
$transportFactory->buildContext($container, []);
32+
$transportFactory->buildQueueConsumer($container, $config['consumption']);
33+
$transportFactory->buildRpcClient($container, []);
34+
}
35+
36+
$container->setParameter('enqueue.transports', array_keys($config['transport']));
3337

3438
if (isset($config['client'])) {
3539
$this->setupAutowiringForProcessors($container);
@@ -38,12 +42,12 @@ public function load(array $configs, ContainerBuilder $container): void
3842

3943
$clientConfig = $config['client'];
4044
// todo
41-
$clientConfig['transport'] = $config['transport'];
45+
$clientConfig['transport'] = $config['transport']['default'];
4246
$clientConfig['consumption'] = $config['consumption'];
4347

4448
$clientFactory = new ClientFactory('default');
4549
$clientFactory->build($container, $clientConfig);
46-
$clientFactory->createDriver($container, $config['transport']);
50+
$clientFactory->createDriver($container, $config['transport']['default']);
4751
}
4852

4953
if ($config['job']) {

‎pkg/enqueue-bundle/EnqueueBundle.php

+2-2
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,8 @@ class EnqueueBundle extends Bundle
2323
public function build(ContainerBuilder $container): void
2424
{
2525
//transport passes
26-
$container->addCompilerPass(new BuildConsumptionExtensionsPass('default'));
27-
$container->addCompilerPass(new BuildProcessorRegistryPass('default'));
26+
$container->addCompilerPass(new BuildConsumptionExtensionsPass());
27+
$container->addCompilerPass(new BuildProcessorRegistryPass());
2828

2929
//client passes
3030
$container->addCompilerPass(new BuildClientConsumptionExtensionsPass('default'));

‎pkg/enqueue-bundle/Tests/Unit/DependencyInjection/ConfigurationTest.php

+115-19
Original file line numberDiff line numberDiff line change
@@ -28,41 +28,133 @@ public function testCouldBeConstructedWithDebugAsArgument()
2828
new Configuration(true);
2929
}
3030

31-
public function testShouldUseDefaultConfigurationIfNothingIsConfiguredAtAll()
31+
public function testShouldProcessNullAsDefaultNullTransport()
3232
{
3333
$configuration = new Configuration(true);
3434

3535
$processor = new Processor();
36-
$config = $processor->processConfiguration($configuration, [[]]);
36+
$config = $processor->processConfiguration($configuration, [null]);
3737

38-
$this->assertEquals([
39-
'transport' => ['dsn' => 'null:'],
40-
'consumption' => [
41-
'receive_timeout' => 10000,
38+
$this->assertConfigEquals([
39+
'transport' => [
40+
'default' => ['dsn' => 'null:'],
4241
],
43-
'job' => false,
44-
'async_events' => ['enabled' => false],
45-
'async_commands' => ['enabled' => false],
46-
'extensions' => [
47-
'doctrine_ping_connection_extension' => false,
48-
'doctrine_clear_identity_map_extension' => false,
49-
'signal_extension' => function_exists('pcntl_signal_dispatch'),
50-
'reply_extension' => true,
42+
], $config);
43+
}
44+
45+
public function testShouldProcessStringAsDefaultDsnTransport()
46+
{
47+
$configuration = new Configuration(true);
48+
49+
$processor = new Processor();
50+
$config = $processor->processConfiguration($configuration, ['foo://bar?option=val']);
51+
52+
$this->assertConfigEquals([
53+
'transport' => [
54+
'default' => ['dsn' => 'foo://bar?option=val'],
55+
],
56+
], $config);
57+
}
58+
59+
public function testShouldProcessEmptyArrayAsDefaultNullTransport()
60+
{
61+
$configuration = new Configuration(true);
62+
63+
$processor = new Processor();
64+
$config = $processor->processConfiguration($configuration, ['foo://bar?option=val']);
65+
66+
$this->assertConfigEquals([
67+
'transport' => [
68+
'default' => ['dsn' => 'foo://bar?option=val'],
69+
],
70+
], $config);
71+
}
72+
73+
public function testShouldProcessSingleTransportAsDefault()
74+
{
75+
$configuration = new Configuration(true);
76+
77+
$processor = new Processor();
78+
$config = $processor->processConfiguration($configuration, [[
79+
'transport' => 'foo://bar?option=val',
80+
]]);
81+
82+
$this->assertConfigEquals([
83+
'transport' => [
84+
'default' => ['dsn' => 'foo://bar?option=val'],
85+
],
86+
], $config);
87+
}
88+
89+
public function testShouldProcessTransportWithDsnKeyAsDefault()
90+
{
91+
$configuration = new Configuration(true);
92+
93+
$processor = new Processor();
94+
$config = $processor->processConfiguration($configuration, [[
95+
'transport' => [
96+
'dsn' => 'foo://bar?option=val',
97+
],
98+
]]);
99+
100+
$this->assertConfigEquals([
101+
'transport' => [
102+
'default' => ['dsn' => 'foo://bar?option=val'],
51103
],
52104
], $config);
53105
}
54106

55-
public function testShouldUseDefaultTransportIfIfTransportIsConfiguredAtAll()
107+
public function testShouldProcessSeveralTransports()
56108
{
57109
$configuration = new Configuration(true);
58110

59111
$processor = new Processor();
60112
$config = $processor->processConfiguration($configuration, [[
61-
'transport' => null,
113+
'transport' => [
114+
'default' => ['dsn' => 'default:'],
115+
'foo' => ['dsn' => 'foo:'],
116+
'bar' => ['dsn' => 'bar:'],
117+
],
62118
]]);
63119

120+
$this->assertConfigEquals([
121+
'transport' => [
122+
'default' => ['dsn' => 'default:'],
123+
'foo' => ['dsn' => 'foo:'],
124+
'bar' => ['dsn' => 'bar:'],
125+
],
126+
], $config);
127+
}
128+
129+
public function testTransportFactoryShouldValidateEachTransportAccordingToItsRules()
130+
{
131+
$configuration = new Configuration(true);
132+
133+
$processor = new Processor();
134+
135+
$this->expectException(\LogicException::class);
136+
$this->expectExceptionMessage('Both options factory_class and factory_service are set. Please choose one.');
137+
$processor->processConfiguration($configuration, [
138+
[
139+
'transport' => [
140+
'default' => [
141+
'factory_class' => 'aClass',
142+
'factory_service' => 'aService',
143+
],
144+
],
145+
],
146+
]);
147+
}
148+
149+
public function testShouldUseDefaultConfigurationIfNothingIsConfiguredAtAll()
150+
{
151+
$configuration = new Configuration(true);
152+
153+
$processor = new Processor();
154+
$config = $processor->processConfiguration($configuration, [[]]);
155+
64156
$this->assertEquals([
65-
'transport' => ['dsn' => 'null:'],
157+
'transport' => ['default' => ['dsn' => 'null:']],
66158
'consumption' => [
67159
'receive_timeout' => 10000,
68160
],
@@ -88,8 +180,7 @@ public function testShouldSetDefaultConfigurationForClient()
88180
'client' => null,
89181
]]);
90182

91-
$this->assertArraySubset([
92-
'transport' => ['dsn' => 'null:'],
183+
$this->assertConfigEquals([
93184
'client' => [
94185
'prefix' => 'enqueue',
95186
'app_name' => 'app',
@@ -403,4 +494,9 @@ public function testShouldAllowConfigureConsumption()
403494
],
404495
], $config);
405496
}
497+
498+
private function assertConfigEquals(array $expected, array $actual): void
499+
{
500+
$this->assertArraySubset($expected, $actual, false, var_export($actual, true));
501+
}
406502
}

‎pkg/enqueue-bundle/Tests/Unit/DependencyInjection/EnqueueExtensionTest.php

+18
Original file line numberDiff line numberDiff line change
@@ -443,6 +443,24 @@ public function testShouldConfigureQueueConsumer()
443443
$this->assertSame(456, $def->getArgument(4));
444444
}
445445

446+
public function testShouldSetPropertyWithAllConfiguredTransports()
447+
{
448+
$container = $this->getContainerBuilder(true);
449+
450+
$extension = new EnqueueExtension();
451+
$extension->load([[
452+
'client' => [],
453+
'transport' => [
454+
'default' => ['dsn' => 'default:'],
455+
'foo' => ['dsn' => 'foo:'],
456+
'bar' => ['dsn' => 'foo:'],
457+
],
458+
]], $container);
459+
460+
$this->assertTrue($container->hasParameter('enqueue.transports'));
461+
$this->assertEquals(['default', 'foo', 'bar'], $container->getParameter('enqueue.transports'));
462+
}
463+
446464
public function testShouldLoadProcessAutoconfigureChildDefinition()
447465
{
448466
$container = $this->getContainerBuilder(true);

‎pkg/enqueue/Client/DriverFactory.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ public function create(ConnectionFactory $factory, string $dsn, array $config):
3333
$dsn = new Dsn($dsn);
3434

3535
if ($driverInfo = $this->findDriverInfo($dsn, Resources::getAvailableDrivers())) {
36-
$driverClass = $driverInfo['factoryClass'];
36+
$driverClass = $driverInfo['driverClass'];
3737

3838
if (RabbitMqDriver::class === $driverClass) {
3939
if (false == $factory instanceof AmqpConnectionFactory) {

0 commit comments

Comments
 (0)