symfony
diff --git a/‎src/Symfony/Bundle/FrameworkBundle/Command/WorkerAmqpMoveCommand.php
Lines changed: 0 additions & 50 deletions b/‎src/Symfony/Bundle/FrameworkBundle/Command/WorkerAmqpMoveCommand.php
Lines changed: 0 additions & 50 deletions
diff --git a/‎src/Symfony/Bundle/FrameworkBundle/Command/WorkerListCommand.php
Lines changed: 0 additions & 26 deletions b/‎src/Symfony/Bundle/FrameworkBundle/Command/WorkerListCommand.php
Lines changed: 0 additions & 26 deletions
diff --git a/‎src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
Lines changed: 16 additions & 8 deletions b/‎src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
Lines changed: 16 additions & 8 deletions
diff --git a/‎src/Symfony/Bundle/FrameworkBundle/Resources/config/amqp.xml
Lines changed: 24 additions & 0 deletions b/‎src/Symfony/Bundle/FrameworkBundle/Resources/config/amqp.xml
Lines changed: 24 additions & 0 deletions
diff --git a/‎src/Symfony/Bundle/FrameworkBundle/Resources/config/worker.xml
Lines changed: 28 additions & 0 deletions b/‎src/Symfony/Bundle/FrameworkBundle/Resources/config/worker.xml
Lines changed: 28 additions & 0 deletions
diff --git a/‎src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php
Lines changed: 0 additions & 11 deletions b/‎src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php
Lines changed: 0 additions & 11 deletions
diff --git a/‎src/Symfony/Component/Amqp/Command/AmqpMoveCommand.php
Lines changed: 80 additions & 0 deletions b/‎src/Symfony/Component/Amqp/Command/AmqpMoveCommand.php
Lines changed: 80 additions & 0 deletions
diff --git a/‎src/Symfony/Component/Worker/Command/WorkerListCommand.php
Lines changed: 46 additions & 0 deletions b/‎src/Symfony/Component/Worker/Command/WorkerListCommand.php
Lines changed: 46 additions & 0 deletions
@@ -34,6 +34,7 @@
 use Symfony\Component\DependencyInjection\Loader\XmlFileLoader;
 use Symfony\Component\DependencyInjection\Reference;
 use Symfony\Component\DependencyInjection\ServiceSubscriberInterface;
+use Symfony\Component\DependencyInjection\TypedReference;
 use Symfony\Component\EventDispatcher\ContainerAwareEventDispatcher;
 use Symfony\Component\EventDispatcher\EventDispatcher;
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
@@ -241,9 +242,9 @@ public function load(array $configs, ContainerBuilder $container)
         $this->registerAnnotationsConfiguration($config['annotations'], $container, $loader);
         $this->registerPropertyAccessConfiguration($config['property_access'], $container);
         if (isset($config['amqp'])) {
-            $this->registerAmqpConfiguration($config['amqp'], $container);
+            $this->registerAmqpConfiguration($config['amqp'], $container, $loader);
         }
-        $this->registerWorkerConfiguration($config['worker'], $container);
+        $this->registerWorkerConfiguration($config['worker'], $container, $loader);
 
         if ($this->isConfigEnabled($container, $config['serializer'])) {
             $this->registerSerializerConfiguration($config['serializer'], $container, $loader);
@@ -1305,8 +1306,10 @@ private function registerPropertyAccessConfiguration(array $config, ContainerBui
         ;
     }
 
-    private function registerAmqpConfiguration(array $config, ContainerBuilder $container)
+    private function registerAmqpConfiguration(array $config, ContainerBuilder $container, XmlFileLoader $loader)
     {
+        $loader->load('amqp.xml');
+
         $defaultConnectionName = null;
         if (isset($config['default_connection'])) {
             $defaultConnectionName = $config['default_connection'];
@@ -1336,10 +1339,11 @@ private function registerAmqpConfiguration(array $config, ContainerBuilder $cont
         $container->setAlias('amqp.broker', sprintf('amqp.broker.%s', $defaultConnectionName));
     }
 
-    private function registerWorkerConfiguration(array $config, ContainerBuilder $container)
+    private function registerWorkerConfiguration(array $config, ContainerBuilder $container, XmlFileLoader $loader)
     {
-        $fetchers = array();
+        $loader->load('worker.xml');
 
+        $fetchers = array();
         foreach ($config['fetchers']['amqps'] as $name => $fetcher) {
             if (isset($fetchers[$name])) {
                 throw new \InvalidArgumentException(sprintf('A fetcher named "%s" already exist.', $name));
@@ -1453,11 +1457,15 @@ private function registerWorkerConfiguration(array $config, ContainerBuilder $co
                 ->addArgument($name)
             ;
 
-            $workers[$name] = $id;
+            $workers[$name] = new TypedReference($id, Worker\Loop\Loop::class);
         }
 
-        $container->setParameter('worker.cli_title_prefix', trim($config['cli_title_prefix'], '_'));
-        $container->setParameter('worker.workers', $workers);
+        $container->getDefinition('worker.command.list')->replaceArgument(0, $workerNames = array_keys($workers));
+        $container->getDefinition('worker.worker_locator')->replaceArgument(0, $workers);
+        $container
+            ->getDefinition('worker.command.run')
+            ->replaceArgument(1, trim($config['cli_title_prefix'], '_'))
+            ->replaceArgument(2, $workerNames);
     }
 
     /**
 
@@ -0,0 +1,24 @@
+<?xml version="1.0" ?>
+
+<container xmlns="http://symfony.com/schema/dic/services"
+           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+           xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+
+    <services>
+        <defaults public="false" />
+
+        <service id="amqp.broker_locator" class="Symfony\Component\DependencyInjection\ServiceLocator">
+            <tag name="container.service_locator" />
+            <argument type="collection">
+                <argument key="Symfony\Component\Amqp\Broker" type="service" id="amqp.broker" />
+            </argument>
+        </service>
+
+        <service id="amqp.command.move" class="Symfony\Component\Amqp\Command\AmqpMoveCommand">
+            <tag name="console.command" />
+            <argument type="service" id="amqp.broker_locator" />
+            <argument type="service" id="logger" on-invalid="null" />
+        </service>
+
+    </services>
+</container>
@@ -0,0 +1,28 @@
+<?xml version="1.0" ?>
+
+<container xmlns="http://symfony.com/schema/dic/services"
+           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+
+    <services>
+        <defaults public="false" />
+
+        <service id="worker.worker_locator" class="Symfony\Component\DependencyInjection\ServiceLocator">
+            <tag name="container.service_locator" />
+            <argument type="collection" /> <!-- worker services keyed by names -->
+        </service>
+
+        <service id="worker.command.list" class="Symfony\Component\Worker\Command\WorkerListCommand">
+            <tag name="console.command" />
+            <argument /> <!-- worker names -->
+        </service>
+
+        <service id="worker.command.run" class="Symfony\Component\Worker\Command\WorkerRunCommand">
+            <tag name="console.command" />
+            <argument type="service" id="worker.worker_locator" />
+            <argument /> <!-- process title prefix -->
+            <argument /> <!-- worker names -->
+        </service>
+
+    </services>
+</container>
@@ -1039,9 +1039,6 @@ public function testAmqpFull()
     public function testWorkerEmpty()
     {
         $container = $this->createContainerFromFile('worker_empty');
-
-        $this->assertSame('app', $container->getParameter('worker.cli_title_prefix'));
-        $this->assertSame(array(), $container->getParameter('worker.workers'));
     }
 
     public function testWorkerFull()
@@ -1137,14 +1134,6 @@ public function testWorkerFull()
         $this->assertInstanceOf(Reference::class, $worker->getArgument(2));
         $this->assertSame('logger', (string) $worker->getArgument(2));
         $this->assertSame('worker_service_a', $worker->getArgument(3));
-
-        /* Global configuration */
-        $this->assertSame('foobar', $container->getParameter('worker.cli_title_prefix'));
-        $workers = array(
-            'worker_d' => 'worker.worker.worker_d',
-            'worker_service_a' => 'worker.worker.worker_service_a',
-        );
-        $this->assertSame($workers, $container->getParameter('worker.workers'));
     }
 
     protected function createContainer(array $data = array())
 
@@ -0,0 +1,80 @@
+<?php
+
+namespace Symfony\Component\Amqp\Command;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Amqp\Broker;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\DependencyInjection\ContainerInterface;
+
+/**
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ * @author Robin Chalas <robin.chalas@gmail.com>
+ */
+class AmqpMoveCommand extends Command
+{
+    private $container;
+    private $logger;
+
+    /**
+     * @param ContainerInterface $container A PSR11 container from which to load the Broker service
+     * @param LoggerInterface|null $logger
+     */
+    public function __construct(ContainerInterface $container, LoggerInterface $logger = null)
+    {
+        parent::__construct();
+
+        $this->container = $container;
+        $this->logger = $logger;
+    }
+
+    protected function configure()
+    {
+        $this
+            ->setName('amqp:move')
+            ->setDescription('Takes all messages from a queue and sends them to the default exchange with a new routing key.')
+            ->setDefinition(array(
+                new InputArgument('from', InputArgument::REQUIRED, 'The queue.'),
+                new InputArgument('to', InputArgument::REQUIRED, 'The new routing key.'),
+            ))
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        /** @var Broker $broker */
+        $broker = $this->container->get(Broker::class);
+        $io = new SymfonyStyle($input, $output);
+        $from = $input->getArgument('from');
+        $to = $input->getArgument('to');
+
+        while (false !== $message = $broker->get($from)) {
+            $io->comment("Moving a message from $from to $to...");
+
+            if (null !== $this->logger) {
+                $this->logger->info('Moving a message from {from} to {to}.', array(
+                    'from' => $from,
+                    'to' => $to,
+                ));
+            }
+
+            $broker->move($message, $to);
+            $broker->ack($message);
+
+            if ($output->isDebug()) {
+                $io->comment("...message moved from $from to $to.");
+            }
+
+            if (null !== $this->logger) {
+                $this->logger->debug('...message moved {from} to {to}.', array(
+                    'from' => $from,
+                    'to' => $to,
+                ));
+            }
+        }
+    }
+}
@@ -0,0 +1,46 @@
+<?php
+
+namespace Symfony\Component\Worker\Command;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+
+/**
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ * @author Robin Chalas <robin.chalas@gmail.com>
+ */
+class WorkerListCommand extends Command
+{
+    private $workers;
+
+    public function __construct(array $workers = array())
+    {
+        parent::__construct();
+
+        $this->workers = $workers;
+    }
+
+    protected function configure()
+    {
+        $this
+            ->setName('worker:list')
+            ->setDescription('Lists available workers.')
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $io = new SymfonyStyle($input, $output);
+
+        if (!$this->workers) {
+            $io->getErrorStyle()->error('There are no available workers.');
+
+            return;
+        }
+
+        $io->getErrorStyle()->title('Available workers');
+        $io->listing($this->workers);
+    }
+}