Skip to content

Commit b617dea

Browse files
makasimmakasim
committed
[bundle] Add BC for topic\command subscribers.
1 parent 815083c commit b617dea

5 files changed

Lines changed: 169 additions & 3 deletions

File tree

‎pkg/enqueue/Client/TopicSubscriberInterface.php‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,22 @@ interface TopicSubscriberInterface
1515
*
1616
* or
1717
*
18-
* ['aTopicName' => [
19-
* 'processor' => 'processor',
18+
* [
19+
* [
20+
* 'topic' => 'aTopicName',
21+
* 'processor' => 'fooProcessor',
2022
* 'queue' => 'a_client_queue_name',
21-
* ]]
23+
*
24+
* 'aCustomOption' => 'aVal',
25+
* ],
26+
* [
27+
* 'topic' => 'anotherTopicName',
28+
* 'processor' => 'barProcessor',
29+
* 'queue' => 'a_client_queue_name',
30+
*
31+
* 'aCustomOption' => 'aVal',
32+
* ],
33+
* ]
2234
*
2335
* Note: If you set prefix_queue to true then the queue is used as is and therefor the driver is not used to prepare a transport queue name.
2436
* It is possible to pass other options, they could be accessible on a route instance through options.

‎pkg/enqueue/Symfony/Client/DependencyInjection/BuildCommandSubscriberRoutesPass.php‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,41 @@ public function process(ContainerBuilder $container): void
6666
throw new \LogicException('Command subscriber configuration is invalid. Should be an array or string.');
6767
}
6868

69+
// 0.8 command subscriber
70+
if (isset($commands['processorName'])) {
71+
@trigger_error('The command subscriber 0.8 syntax is deprecated since Enqueue 0.9.', E_USER_DEPRECATED);
72+
73+
$source = $commands['processorName'];
74+
$processor = $params['processorName'] ?? $serviceId;
75+
76+
$options = $commands;
77+
unset(
78+
$options['processorName'],
79+
$options['queueName'],
80+
$options['queueNameHardcoded'],
81+
$options['exclusive'],
82+
$options['topic'],
83+
$options['source'],
84+
$options['source_type'],
85+
$options['processor'],
86+
$options['options']
87+
);
88+
89+
$options['processor_service_id'] = $serviceId;
90+
91+
if (isset($commands['queueName'])) {
92+
$options['queue'] = $commands['queueName'];
93+
}
94+
95+
if (isset($commands['queueNameHardcoded']) && $commands['queueNameHardcoded']) {
96+
$options['prefix_queue'] = false;
97+
}
98+
99+
$routeCollection->add(new Route($source, Route::COMMAND, $processor, $options));
100+
101+
continue;
102+
}
103+
69104
if (isset($commands['command'])) {
70105
$commands = [$commands];
71106
}

‎pkg/enqueue/Symfony/Client/DependencyInjection/BuildTopicSubscriberRoutesPass.php‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,37 @@ public function process(ContainerBuilder $container): void
6969
foreach ($topics as $key => $params) {
7070
if (is_string($params)) {
7171
$routeCollection->add(new Route($params, Route::TOPIC, $serviceId, ['processor_service_id' => $serviceId]));
72+
73+
// 0.8 topic subscriber
74+
} elseif (is_array($params) && is_string($key)) {
75+
@trigger_error('The topic subscriber 0.8 syntax is deprecated since Enqueue 0.9.', E_USER_DEPRECATED);
76+
77+
$source = $key;
78+
$processor = $params['processorName'] ?? $serviceId;
79+
80+
$options = $params;
81+
unset(
82+
$options['processorName'],
83+
$options['queueName'],
84+
$options['queueNameHardcoded'],
85+
$options['topic'],
86+
$options['source'],
87+
$options['source_type'],
88+
$options['processor'],
89+
$options['options']
90+
);
91+
92+
$options['processor_service_id'] = $serviceId;
93+
94+
if (isset($params['queueName'])) {
95+
$options['queue'] = $params['queueName'];
96+
}
97+
98+
if (isset($params['queueNameHardcoded']) && $params['queueNameHardcoded']) {
99+
$options['prefix_queue'] = false;
100+
}
101+
102+
$routeCollection->add(new Route($source, Route::TOPIC, $processor, $options));
72103
} elseif (is_array($params)) {
73104
$source = $params['topic'] ?? null;
74105
$processor = $params['processor'] ?? $serviceId;

‎pkg/enqueue/Tests/Symfony/Client/DependencyInjection/BuildCommandSubscriberRoutesPassTest.php‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -398,6 +398,48 @@ public function testShouldMergeExtractedRoutesWithAlreadySetInCollection()
398398
);
399399
}
400400

401+
public function testShouldRegister08CommandProcessor()
402+
{
403+
$routeCollection = new Definition(RouteCollection::class);
404+
$routeCollection->addArgument([]);
405+
406+
$processor = $this->createCommandSubscriberProcessor([
407+
'processorName' => 'fooCommand',
408+
'queueName' => 'a_client_queue_name',
409+
'queueNameHardcoded' => true,
410+
'exclusive' => true,
411+
'anOption' => 'aFooVal',
412+
]);
413+
414+
$container = new ContainerBuilder();
415+
$container->setParameter('enqueue.clients', ['default']);
416+
$container->setDefinition('enqueue.client.default.route_collection', $routeCollection);
417+
$container->register('aFooProcessor', get_class($processor))
418+
->addTag('enqueue.command_subscriber')
419+
;
420+
421+
$pass = new BuildCommandSubscriberRoutesPass();
422+
$pass->process($container);
423+
424+
$this->assertInternalType('array', $routeCollection->getArgument(0));
425+
$this->assertCount(1, $routeCollection->getArgument(0));
426+
427+
$this->assertEquals(
428+
[
429+
[
430+
'source' => 'fooCommand',
431+
'source_type' => 'enqueue.client.command_route',
432+
'processor' => 'aFooProcessor',
433+
'processor_service_id' => 'aFooProcessor',
434+
'anOption' => 'aFooVal',
435+
'queue' => 'a_client_queue_name',
436+
'prefix_queue' => false,
437+
],
438+
],
439+
$routeCollection->getArgument(0)
440+
);
441+
}
442+
401443
private function createCommandSubscriberProcessor($commandSubscriberReturns = ['aCommand'])
402444
{
403445
$processor = new class() implements Processor, CommandSubscriberInterface {

‎pkg/enqueue/Tests/Symfony/Client/DependencyInjection/BuildTopicSubscriberRoutesPassTest.php‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -358,6 +358,52 @@ public function testShouldMergeExtractedRoutesWithAlreadySetInCollection()
358358
);
359359
}
360360

361+
public function testShouldRegister08TopicSubscriber()
362+
{
363+
$routeCollection = new Definition(RouteCollection::class);
364+
$routeCollection->addArgument([]);
365+
366+
$processor = $this->createTopicSubscriberProcessor([
367+
'fooTopic' => ['processorName' => 'aCustomFooProcessorName', 'queueName' => 'fooQueue', 'queueNameHardcoded' => true, 'anOption' => 'aFooVal'],
368+
'barTopic' => ['processorName' => 'aCustomBarProcessorName', 'anOption' => 'aBarVal'],
369+
]);
370+
371+
$container = new ContainerBuilder();
372+
$container->setParameter('enqueue.clients', ['default']);
373+
$container->setDefinition('enqueue.client.default.route_collection', $routeCollection);
374+
$container->register('aFooProcessor', get_class($processor))
375+
->addTag('enqueue.topic_subscriber')
376+
;
377+
378+
$pass = new BuildTopicSubscriberRoutesPass();
379+
$pass->process($container);
380+
381+
$this->assertInternalType('array', $routeCollection->getArgument(0));
382+
$this->assertCount(2, $routeCollection->getArgument(0));
383+
384+
$this->assertEquals(
385+
[
386+
[
387+
'source' => 'fooTopic',
388+
'source_type' => 'enqueue.client.topic_route',
389+
'processor' => 'aCustomFooProcessorName',
390+
'processor_service_id' => 'aFooProcessor',
391+
'anOption' => 'aFooVal',
392+
'queue' => 'fooQueue',
393+
'prefix_queue' => false,
394+
],
395+
[
396+
'source' => 'barTopic',
397+
'source_type' => 'enqueue.client.topic_route',
398+
'processor' => 'aCustomBarProcessorName',
399+
'processor_service_id' => 'aFooProcessor',
400+
'anOption' => 'aBarVal',
401+
],
402+
],
403+
$routeCollection->getArgument(0)
404+
);
405+
}
406+
361407
private function createTopicSubscriberProcessor($topicSubscriberReturns = ['aTopic'])
362408
{
363409
$processor = new class() implements Processor, TopicSubscriberInterface {

0 commit comments

Comments
 (0)