WorkflowBundle/Services/CommandConsumer.php

<?php

namespace WorkflowBundle\Services;

use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;
use Symfony\Bundle\FrameworkBundle\Console\Application;
use Symfony\Component\DependencyInjection\ContainerInterface;
use Symfony\Component\Console\Input\ArrayInput;
use Symfony\Component\Console\Output\BufferedOutput;

class CommandConsumer implements ConsumerInterface
{

    /**
     * @var ContainerInterface
     */
    protected $serviceContainer;


    /**
     * @param ContainerInterface $serviceContainer
     */
    public function __construct(ContainerInterface $serviceContainer)
    {
        $this->serviceContainer = $serviceContainer;
    }

    /**
     * $msg will be an instance of `PhpAmqpLib\Message\AMQPMessage` 
     * with the $msg->body being the data sent over RabbitMQ.
     * 
     * @param AMQPMessage $msg
     */
    public function execute(AMQPMessage $msg)
    {
        $msgBody = unserialize($msg->body);
        if (isset($msgBody['name']) && isset($msgBody['args'])) {
            // command name and args
            $name = $msgBody['name'];
            $args = $msgBody['args'];

            $kernel = $this->serviceContainer->get('kernel');
            $application = new Application($kernel);
            $application->setAutoExit(false);

            $input = array('command' => $name,);
            foreach ($args as $arg) {
                $pieces = explode(':', $arg);
                if (isset($pieces[0]) && isset($pieces[1])) {
                    $input[$pieces[0]] = $pieces[1];
                }
            }
            
            $input = new ArrayInput($input);

            $output = new BufferedOutput();

            $application->run($input, $output);

            // return the output
            $content = $output->fetch();
            
            echo $content;

            return true;
        }

        return false;
    }

}