Add --memory-limit, signal handling and graceful_max_execution to rabbitmq:rpc-server

Command/RpcServerCommand.php

@@ -2,12 +2,9 @@
 
 namespace OldSound\RabbitMqBundle\Command;
 
-use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Input\InputArgument;
-use Symfony\Component\Console\Input\InputInterface;
-use Symfony\Component\Console\Output\OutputInterface;
 
-class RpcServerCommand extends BaseRabbitMqCommand
+class RpcServerCommand extends BaseConsumerCommand
 {
     protected function configure(): void
     {
@@ -16,35 +13,17 @@ protected function configure(): void
         $this
             ->setName('rabbitmq:rpc-server')
             ->setDescription('Start an RPC server')
-            ->addArgument('name', InputArgument::REQUIRED, 'Server Name')
-            ->addOption('messages', 'm', InputOption::VALUE_OPTIONAL, 'Messages to consume', '0')
-            ->addOption('debug', 'd', InputOption::VALUE_OPTIONAL, 'Debug mode', false)
         ;
+
+        // Restore an RPC-specific description for the "name" argument, inherited
+        // from BaseConsumerCommand as "Consumer Name".
+        $this->getDefinition()->setArguments([
+            new InputArgument('name', InputArgument::REQUIRED, 'Server Name'),
+        ]);
     }
 
-    /**
-     * 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
-     */
-    protected function execute(InputInterface $input, OutputInterface $output): int
+    protected function getConsumerService()
     {
-        define('AMQP_DEBUG', (bool) $input->getOption('debug'));
-        $amount = (int)$input->getOption('messages');
-
-        if (0 > $amount) {
-            throw new \InvalidArgumentException("The -m option should be null or greater than 0");
-        }
-
-        $this->getContainer()
-               ->get(sprintf('old_sound_rabbit_mq.%s_server', $input->getArgument('name')))
-               ->start($amount);
-
-        return 0;
+        return 'old_sound_rabbit_mq.%s_server';
     }
 }

DependencyInjection/Configuration.php

@@ -420,6 +420,16 @@ protected function addRpcServers(ArrayNodeDefinition $node)
                         ->children()
                             ->scalarNode('connection')->defaultValue('default')->end()
                             ->scalarNode('callback')->isRequired()->end()
+                            ->scalarNode('idle_timeout')->end()
+                            ->scalarNode('idle_timeout_exit_code')->end()
+                            ->scalarNode('timeout_wait')->end()
+                            ->arrayNode('graceful_max_execution')
+                                ->canBeUnset()
+                                ->children()
+                                    ->integerNode('timeout')->end()
+                                    ->integerNode('exit_code')->defaultValue(0)->end()
+                                ->end()
+                            ->end()
                             ->arrayNode('qos_options')
                                 ->canBeUnset()
                                 ->children()

DependencyInjection/OldSoundRabbitMqExtension.php

@@ -666,6 +666,25 @@ protected function loadRpcServers()
             if (array_key_exists('serializer', $server)) {
                 $definition->addMethodCall('setSerializer', [$server['serializer']]);
             }
+            if (isset($server['idle_timeout'])) {
+                $definition->addMethodCall('setIdleTimeout', [$server['idle_timeout']]);
+            }
+            if (isset($server['idle_timeout_exit_code'])) {
+                $definition->addMethodCall('setIdleTimeoutExitCode', [$server['idle_timeout_exit_code']]);
+            }
+            if (isset($server['timeout_wait'])) {
+                $definition->addMethodCall('setTimeoutWait', [$server['timeout_wait']]);
+            }
+            if (isset($server['graceful_max_execution'])) {
+                $definition->addMethodCall(
+                    'setGracefulMaxExecutionDateTimeFromSecondsInTheFuture',
+                    [$server['graceful_max_execution']['timeout']]
+                );
+                $definition->addMethodCall(
+                    'setGracefulMaxExecutionTimeoutExitCode',
+                    [$server['graceful_max_execution']['exit_code']]
+                );
+            }
             $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_server', $key), $definition);
         }
     }

Event/AMQPEvent.php

@@ -2,7 +2,7 @@
 
 namespace OldSound\RabbitMqBundle\Event;
 
-use OldSound\RabbitMqBundle\RabbitMq\Consumer;
+use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer;
 use OldSound\RabbitMqBundle\RabbitMq\Producer;
 use PhpAmqpLib\Message\AMQPMessage;
 
@@ -27,7 +27,7 @@ class AMQPEvent extends AbstractAMQPEvent
     protected $AMQPMessage;
 
     /**
-     * @var Consumer
+     * @var BaseConsumer
      */
     protected $consumer;
 
@@ -57,19 +57,19 @@ public function setAMQPMessage(AMQPMessage $AMQPMessage)
     }
 
     /**
-     * @return Consumer
+     * @return BaseConsumer
      */
     public function getConsumer()
     {
         return $this->consumer;
     }
 
     /**
-     * @param Consumer $consumer
+     * @param BaseConsumer $consumer
      *
      * @return AMQPEvent
      */
-    public function setConsumer(Consumer $consumer)
+    public function setConsumer(BaseConsumer $consumer)
     {
         $this->consumer = $consumer;
 

Event/OnConsumeEvent.php

@@ -2,7 +2,7 @@
 
 namespace OldSound\RabbitMqBundle\Event;
 
-use OldSound\RabbitMqBundle\RabbitMq\Consumer;
+use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer;
 
 /**
  * Class OnConsumeEvent
@@ -16,9 +16,9 @@ class OnConsumeEvent extends AMQPEvent
     /**
      * OnConsumeEvent constructor.
      *
-     * @param Consumer $consumer
+     * @param BaseConsumer $consumer
      */
-    public function __construct(Consumer $consumer)
+    public function __construct(BaseConsumer $consumer)
     {
         $this->setConsumer($consumer);
     }

Event/OnIdleEvent.php

@@ -2,7 +2,7 @@
 
 namespace OldSound\RabbitMqBundle\Event;
 
-use OldSound\RabbitMqBundle\RabbitMq\Consumer;
+use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer;
 
 /**
  * Class OnIdleEvent
@@ -21,9 +21,9 @@ class OnIdleEvent extends AMQPEvent
     /**
      * OnConsumeEvent constructor.
      *
-     * @param Consumer $consumer
+     * @param BaseConsumer $consumer
      */
-    public function __construct(Consumer $consumer)
+    public function __construct(BaseConsumer $consumer)
     {
         $this->setConsumer($consumer);
 

README.md

@@ -707,6 +707,12 @@ rpc_servers:
         exchange_options: {name: random_int, type: topic}
         queue_options: {name: random_int_queue, durable: false, auto_delete: true}
         serializer: json_encode
+        idle_timeout: 60 #default: ~
+        idle_timeout_exit_code: 0 #default: ~
+        timeout_wait: 10 #default: ~
+        graceful_max_execution:
+            timeout: 1800 #default: ~
+            exit_code: 10 #default: 0
 ```
 
 *For a full configuration reference please use the `php app/console config:dump-reference old_sound_rabbit_mq` command.*
@@ -719,6 +725,18 @@ First we have to start the server from the command line:
 $ ./app/console_dev rabbitmq:rpc-server random_int
 ```
 
+The RPC server is a long-running consumer process, so `rabbitmq:rpc-server` accepts the same options as `rabbitmq:consumer`:
+
+- `--messages|-m` stops the server after the given number of requests;
+- `--memory-limit|-l` sets a memory limit in MB — the server stops gracefully five MB before reaching it, so a supervisor can restart it (useful against slow memory leaks);
+- `--without-signals|-w` disables the graceful handling of SIGTERM/SIGINT/SIGQUIT.
+
+The `idle_timeout`, `timeout_wait` and `graceful_max_execution` options shown above behave exactly as they do for consumers.
+
+```bash
+$ ./app/console_dev rabbitmq:rpc-server random_int -l 256
+```
+
 And then add the following code to our controller:
 
 ```php