Permalink
Fetching contributors…
Cannot retrieve contributors at this time
105 lines (85 sloc) 3.8 KB
<?php
namespace OldSound\RabbitMqBundle\Command;
use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer as Consumer;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
abstract class BaseConsumerCommand extends BaseRabbitMqCommand
{
protected $consumer;
protected $amount;
abstract protected function getConsumerService();
public function stopConsumer()
{
if ($this->consumer instanceof Consumer) {
// Process current message, then halt consumer
$this->consumer->forceStopConsumer();
// Halt consumer if waiting for a new message from the queue
try {
$this->consumer->stopConsuming();
} catch (AMQPTimeoutException $e) {}
} else {
exit();
}
}
public function restartConsumer()
{
// TODO: Implement restarting of consumer
}
protected function configure()
{
parent::configure();
$this
->addArgument('name', InputArgument::REQUIRED, 'Consumer Name')
->addOption('messages', 'm', InputOption::VALUE_OPTIONAL, 'Messages to consume', 0)
->addOption('route', 'r', InputOption::VALUE_OPTIONAL, 'Routing Key', '')
->addOption('memory-limit', 'l', InputOption::VALUE_OPTIONAL, 'Allowed memory for this process', null)
->addOption('debug', 'd', InputOption::VALUE_NONE, 'Enable Debugging')
->addOption('without-signals', 'w', InputOption::VALUE_NONE, 'Disable catching of system signals')
;
}
/**
* Executes the current command.
*
* @param InputInterface $input An InputInterface instance
* @param OutputInterface $output An OutputInterface instance
*
* @return integer 0 if everything went fine, or an error code
*
* @throws \InvalidArgumentException When the number of messages to consume is less than 0
* @throws \BadFunctionCallException When the pcntl is not installed and option -s is true
*/
protected function execute(InputInterface $input, OutputInterface $output)
{
if (defined('AMQP_WITHOUT_SIGNALS') === false) {
define('AMQP_WITHOUT_SIGNALS', $input->getOption('without-signals'));
}
if (!AMQP_WITHOUT_SIGNALS && extension_loaded('pcntl')) {
if (!function_exists('pcntl_signal')) {
throw new \BadFunctionCallException("Function 'pcntl_signal' is referenced in the php.ini 'disable_functions' and can't be called.");
}
pcntl_signal(SIGTERM, array(&$this, 'stopConsumer'));
pcntl_signal(SIGINT, array(&$this, 'stopConsumer'));
pcntl_signal(SIGHUP, array(&$this, 'restartConsumer'));
}
if (defined('AMQP_DEBUG') === false) {
define('AMQP_DEBUG', (bool) $input->getOption('debug'));
}
$this->amount = $input->getOption('messages');
if (0 > $this->amount) {
throw new \InvalidArgumentException("The -m option should be null or greater than 0");
}
$this->initConsumer($input);
$this->consumer->consume($this->amount);
}
protected function initConsumer($input) {
$this->consumer = $this->getContainer()
->get(sprintf($this->getConsumerService(), $input->getArgument('name')));
if (!is_null($input->getOption('memory-limit')) && ctype_digit((string) $input->getOption('memory-limit')) && $input->getOption('memory-limit') > 0) {
$this->consumer->setMemoryLimit($input->getOption('memory-limit'));
}
$this->consumer->setRoutingKey($input->getOption('route'));
}
}