8000 Merge pull request #12 from chalasr/amqp-worker-commands · symfony/symfony@dbc6c11 · GitHub
[go: up one dir, main page]
Skip to content

Commit dbc6c11

Browse files
lyrixxlyrixx
authored
Merge pull request #12 from chalasr/amqp-worker-commands
Register amqp/worker commands as services
2 parents 1b7f379 + d36c1bf commit dbc6c11

File tree

11 files changed

+269
-113
lines changed

11 files changed

+269
-113
lines changed

‎src/Symfony/Bundle/FrameworkBundle/Command/WorkerAmqpMoveCommand.php

Lines changed: 0 additions & 50 deletions
This file was deleted.

‎src/Symfony/Bundle/FrameworkBundle/Command/WorkerListCommand.php

Lines changed: 0 additions & 26 deletions
This file was deleted.

‎src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
use Symfony\Component\DependencyInjection\Loader\XmlFileLoader;
3535
use Symfony\Component\DependencyInjection\Reference;
3636
use Symfony\Component\DependencyInjection\ServiceSubscriberInterface;
37+
use Symfony\Component\DependencyInjection\TypedReference;
3738
use Symfony\Component\EventDispatcher\ContainerAwareEventDispatcher;
3839
use Symfony\Component\EventDispatcher\EventDispatcher;
3940
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
@@ -241,9 +242,9 @@ public function load(array $configs, ContainerBuilder $container)
241242
$this->registerAnnotationsConfiguration($config['annotations'], $container, $loader);
242243
$this->registerPropertyAccessConfiguration($config['property_access'], $container);
243244
if (isset($config['amqp'])) {
244-
$this->registerAmqpConfiguration($config['amqp'], $container);
245+
$this->registerAmqpConfiguration($config['amqp'], $container, $loader);
245246
}
246-
$this->registerWorkerConfiguration($config['worker'], $container);
247+
$this->registerWorkerConfiguration($config['worker'], $container, $loader);
247248

248249
if ($this->isConfigEnabled($container, $config['serializer'])) {
249250
$this->registerSerializerConfiguration($config['serializer'], $container, $loader);
@@ -1299,8 +1300,10 @@ private function registerPropertyAccessConfiguration(array $config, ContainerBui
12991300
;
13001301
}
13011302

1302-
private function registerAmqpConfiguration(array $config, ContainerBuilder $container)
1303+
private function registerAmqpConfiguration(array $config, ContainerBuilder $container, XmlFileLoader $loader)
13031304
{
1305+
$loader->load('amqp.xml');
1306+
13041307
$defaultConnectionName = null;
13051308
if (isset($config['default_connection'])) {
13061309
$defaultConnectionName = $config['default_connection'];
@@ -1330,10 +1333,11 @@ private function registerAmqpConfiguration(array $config, ContainerBuilder $cont
13301333
$container->setAlias('amqp.broker', sprintf('amqp.broker.%s', $defaultConnectionName));
13311334
}
13321335

1333-
private function registerWorkerConfiguration(array $config, ContainerBuilder $container)
1336+
private function registerWorkerConfiguration(array $config, ContainerBuilder $container, XmlFileLoader $loader)
13341337
{
1335-
$fetchers = array();
1338+
$loader->load('worker.xml');
13361339

1340+
$fetchers = array();
13371341
foreach ($config['fetchers']['amqps'] as $name => $fetcher) {
13381342
if (isset($fetchers[$name])) {
13391343
throw new \InvalidArgumentException(sprintf('A fetcher named "%s" already exist.', $name));
@@ -1447,11 +1451,15 @@ private function registerWorkerConfiguration(array $config, ContainerBuilder $co
14471451
->addArgument($name)
14481452
;
14491453

1450-
$workers[$name] = $id;
1454+
$workers[$name] = new TypedReference($id, Worker\Loop\Loop::class);
14511455
}
14521456

1453-
$container->setParameter('worker.cli_title_prefix', trim($config['cli_title_prefix'], '_'));
1454-
$container->setParameter('worker.workers', $workers);
1457+
$container->getDefinition('worker.command.list')->replaceArgument(0, $workerNames = array_keys($workers));
1458+
$container->getDefinition('worker.worker_locator')->replaceArgument(0, $workers);
1459+
$container
1460+
->getDefinition('worker.command.run')
1461+
->replaceArgument(1, trim($config['cli_title_prefix'], '_'))
1462+
->replaceArgument(2, $workerNames);
14551463
}
14561464

14571465
/**

‎src/Symfony/Bundle/FrameworkBundle/Resources/config/amqp.xml

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
<?xml version="1.0" ?>
2+
3+
<container xmlns="http://symfony.com/schema/dic/services"
4+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
5+
xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
6+
7+
<services>
8+
<defaults public="false" />
9+
10+
<service id="amqp.broker_locator" class="Symfony\Component\DependencyInjection\ServiceLocator">
11+
<tag name="container.service_locator" />
12+
<argument type="collection">
13+
<argument key="Symfony\Component\Amqp\Broker" type="service" id="amqp.broker" />
14+
</argument>
15+
</service>
16+
17+
<service id="amqp.command.move" class="Symfony\Component\Amqp\Command\AmqpMoveCommand">
18+
<tag name="console.command" />
19+
<argument type="service" id="amqp.broker_locator" />
20+
<argument type="service" id="logger" on-invalid="null" />
21+
</service>
22+
23+
</services>
24+
</container>

‎src/Symfony/Bundle/FrameworkBundle/Resources/config/worker.xml

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
<?xml version="1.0" ?>
2+
3+
<container xmlns="http://symfony.com/schema/dic/services"
4+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
5+
xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
6+
7+
<services>
8+
<defaults public="false" />
9+
10+
<service id="worker.worker_locator" class="Symfony\Component\DependencyInjection\ServiceLocator">
11+
<tag name="container.service_locator" />
12+
<argument type="collection" /> <!-- worker services keyed by names -->
13+
</service>
14+
15+
<service id="worker.command.list" class="Symfony\Component\Worker\Command\WorkerListCommand">
16+
<tag name="console.command" />
17+
<argument /> <!-- worker names -->
18+
</service>
19+
20+
<service id="worker.command.run" class="Symfony\Component\Worker\Command\WorkerRunCommand">
21+
<tag name="console.command" />
22+
<argument type="service" id="worker.worker_locator" />
23+
<argument /> <!-- process title prefix -->
24+
<argument /> <!-- worker names -->
25+
</service>
26+
27+
</services>
28+
</container>

‎src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1014,9 +1014,6 @@ public function testAmqpFull()
10141014
public function testWorkerEmpty()
10151015
{
10161016
$container = $this->createContainerFromFile('worker_empty');
1017-
1018-
$this->assertSame('app', $container->getParameter('worker.cli_title_prefix'));
1019-
$this->assertSame(array(), $container->getParameter('worker.workers'));
10201017
}
10211018

10221019
public function testWorkerFull()
@@ -1112,14 +1109,6 @@ public function testWorkerFull()
11121109
$this->assertInstanceOf(Reference::class, $worker->getArgument(2));
11131110
$this->assertSame('logger', (string) $worker->getArgument(2));
11141111
$this->assertSame('worker_service_a', $worker->getArgument(3));
1115-
1116-
/* Global configuration */
1117-
$this->assertSame('foobar', $container->getParameter('worker.cli_title_prefix'));
1118-
$workers = array(
1119-
'worker_d' => 'worker.worker.worker_d',
1120-
'worker_service_a' => 'worker.worker.worker_service_a',
1121-
);
1122-
$this->assertSame($workers, $container->getParameter('worker.workers'));
11231112
}
11241 1CF5 113

11251114
protected function createContainer(array $data = array())

‎src/Symfony/Component/Amqp/Command/AmqpMoveCommand.php

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
<?php
2+
3+
namespace Symfony\Component\Amqp\Command;
4+
5+
use Psr\Log\LoggerInterface;
6+
use Symfony\Component\Amqp\Broker;
7+
use Symfony\Component\Console\Command\Command;
8+
use Symfony\Component\Console\Input\InputArgument;
9+
use Symfony\Component\Console\Input\InputInterface;
10+
use Symfony\Component\Console\Output\OutputInterface;
11+
use Symfony\Component\Console\Style\SymfonyStyle;
12+
use Symfony\Component\DependencyInjection\ContainerInterface;
13+
14+
/**
15+
* @author Grégoire Pineau <lyrixx@lyrixx.info>
16+
* @author Robin Chalas <robin.chalas@gmail.com>
17+
*/
18+
class AmqpMoveCommand extends Command
19+
{
20+
private $container;
21+
private $logger;
22+
23+
/**
24+
* @param ContainerInterface $container A PSR11 container from which to load the Broker service
25+
* @param LoggerInterface|null $logger
26+
*/
27+
public function __construct(ContainerInterface $container, LoggerInterface $logger = null)
28+
{
29+
parent::__construct();
30+
31+
$this->container = $container;
32+
$this->logger = $logger;
33+
}
34+
35+
protected function configure()
36+
{
37+
$this
38+
->setName('amqp:move')
39+
->setDescription('Takes all messages from a queue and sends them to the default exchange with a new routing key.')
40+
->setDefinition(array(
41+
new InputArgument('from', InputArgument::REQUIRED, 'The queue.'),
42+
new InputArgument('to', InputArgument::REQUIRED, 'The new routing key.'),
43+
))
44+
;
45+
}
46+
47+
protected function execute(InputInterface $input, OutputInterface $output)
48+
{
49+
/** @var Broker $broker */
50+
$broker = $this->container->get(Broker::class);
51+
$io = new SymfonyStyle($input, $output);
52+
$from = $input->getArgument('from');
53+
$to = $input->getArgument('to');
54+
55+
while (false !== $message = $broker->get($from)) {
56+
$io->comment("Moving a message from $from to $to...");
57+
58+
if (null !== $this->logger) {
59+
$this->logger->info('Moving a message from {from} to {to}.', array(
60+
'from' => $from,
61+
'to' => $to,
62+
));
63+
}
64+
65+
$broker->move($message, $to);
66+
$broker->ack($message);
67+
68+
if ($output->isDebug()) {
69+
$io->comment("...message moved from $from to $to.");
70+
}
71+
72+
if (null !== $this->logger) {
73+
$this->logger->debug('...message moved {from} to {to}.', array(
74+
'from' => $from,
75+
'to' => $to,
76+
));
77+
}
78+
}
79+
}
80+
}

‎src/Symfony/Component/Worker/Command/WorkerListCommand.php

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
<?php
2+
3+
namespace Symfony\Component\Worker\Command;
4+
5+
use Symfony\Component\Console\Command\Command;
6+
use Symfony\Component\Console\Input\InputInterface;
7+
use Symfony\Component\Console\Output\OutputInterface;
8+
use Symfony\Component\Console\Style\SymfonyStyle;
9+
10+
/**
11+
* @author Grégoire Pineau <lyrixx@lyrixx.info>
12+
* @author Robin Chalas <robin.chalas@gmail.com>
13+
*/
14+
class WorkerListCommand extends Command
15+
{
16+
private $workers;
17+
18+
public function __construct(array $workers = array())
19+
{
20+
parent::__construct();
21+
22+
$this->workers = $workers;
23+
}
24+
25+
protected function configure()
26+
{
27+
$this
28+
->setName('worker:list')
29+
->setDescription('Lists available workers.')
30+
;
31+
}
32+
33+
protected function execute(InputInterface $input, OutputInterface $output)
34+
{
35+
$io = new SymfonyStyle($input, $output);
36+
37+
if (!$this->workers) {
38+
$io->getErrorStyle()->error('There are no available workers.');
39+
40+
return;
41+
}
42+
43+
$io->getErrorStyle()->title('Available workers');
44+
$io->listing($this->workers);
45+
}
46+
}

0 commit comments

Comments
 (0)
0