Async commands extended

Author: uro

docs/bundle/async_commands.md

@@ -21,7 +21,10 @@ $ composer require enqueue/async-command:0.9.x-dev
 
 enqueue:
     default:
-        async_commands: true
+        async_commands:
+            enabled: true
+            timeout: 60
+            prefix: ~
 ```
 
 ## Usage

docs/bundle/config_reference.md

@@ -62,6 +62,8 @@ enqueue:
             storage_factory_class: ~
         async_commands:
             enabled:              false
+            timeout:              60
+            prefix: ~
         job:
             enabled:              false
         async_events:

pkg/async-command/DependencyInjection/AsyncCommandExtension.php

@@ -12,13 +12,22 @@ class AsyncCommandExtension extends Extension
     public function load(array $configs, ContainerBuilder $container)
     {
         foreach ($configs['clients'] as $client) {
-            $id = sprintf('enqueue.async_command.%s.run_command_processor', $client);
+            // BC compatibility
+            if (!is_array($client)) {
+                $client = [
+                    'name' => $client,
+                    'prefix' => '',
+                    'timeout' => 60,
+                ];
+            }
+
+            $id = sprintf('enqueue.async_command.%s.run_command_processor', $client['name']);
             $container->register($id, RunCommandProcessor::class)
-                ->addArgument('%kernel.project_dir%')
+                ->addArgument('%kernel.project_dir%', $client['timeout'])
                 ->addTag('enqueue.processor', [
-                    'client' => $client,
-                    'command' => Commands::RUN_COMMAND,
-                    'queue' => Commands::RUN_COMMAND,
+                    'client' => $client['name'],
+                    'command' => $client['prefix'].Commands::RUN_COMMAND,
+                    'queue' => $client['prefix'].Commands::RUN_COMMAND,
                     'prefix_queue' => false,
                     'exclusive' => true,
                 ])

pkg/async-command/RunCommandProcessor.php

@@ -11,14 +11,20 @@
 
 final class RunCommandProcessor implements Processor
 {
+    /**
+     * @var int
+     */
+    private $timeout;
+
     /**
      * @var string
      */
     private $projectDir;
 
-    public function __construct(string $projectDir)
+    public function __construct(string $projectDir, int $timeout = 60)
     {
         $this->projectDir = $projectDir;
+        $this->timeout = $timeout;
     }
 
     public function process(Message $message, Context $context): Result
@@ -29,7 +35,7 @@ public function process(Message $message, Context $context): Result
         $consoleBin = file_exists($this->projectDir.'/bin/console') ? './bin/console' : './app/console';
 
         $process = new Process($phpBin.' '.$consoleBin.' '.$this->getCommandLine($command), $this->projectDir);
-
+        $process->setTimeout($this->timeout);
         $process->run();
 
         if ($message->getReplyTo()) {

pkg/async-command/Tests/RunCommandProcessorTest.php

@@ -28,4 +28,11 @@ public function testCouldBeConstructedWithProjectDirAsFirstArgument()
 
         $this->assertAttributeSame('aProjectDir', 'projectDir', $processor);
     }
+
+    public function testCouldBeConstructedWithTimeoutAsSecondArgument()
+    {
+        $processor = new RunCommandProcessor('aProjectDir', 60);
+
+        $this->assertAttributeSame(60, 'timeout', $processor);
+    }
 }

pkg/enqueue-bundle/DependencyInjection/Configuration.php

@@ -73,6 +73,11 @@ private function getAsyncCommandsConfiguration(): ArrayNodeDefinition
         }
 
         return (new ArrayNodeDefinition('async_commands'))
+            ->children()
+                ->booleanNode('enabled')->defaultFalse()->end()
+                ->integerNode('timeout')->min(0)->defaultValue(60)->end()
+                ->scalarNode('prefix')->defaultValue('')->end()
+            ->end()
             ->addDefaultsIfNotSet()
             ->canBeEnabled()
         ;

pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php

@@ -291,14 +291,18 @@ private function loadReplyExtension(array $config, ContainerBuilder $container):
 
     private function loadAsyncCommands(array $config, ContainerBuilder $container): void
     {
-        $configNames = [];
+        $configs = [];
         foreach ($config as $name => $modules) {
             if (false === empty($modules['async_commands']['enabled'])) {
-                $configNames[] = $name;
+                $configs[] = [
+                    'name' => $name,
+                    'timeout' => $modules['async_commands']['timeout'],
+                    'prefix' => $modules['async_commands']['prefix'],
+                ];
             }
         }
 
-        if (false == $configNames) {
+        if (false == $configs) {
             return;
         }
 
@@ -307,7 +311,7 @@ private function loadAsyncCommands(array $config, ContainerBuilder $container):
         }
 
         $extension = new AsyncCommandExtension();
-        $extension->load(['clients' => $configNames], $container);
+        $extension->load(['clients' => $configs], $container);
     }
 
     private function loadMessageQueueCollector(array $config, ContainerBuilder $container)

pkg/enqueue-bundle/Tests/Functional/App/config/config.yml

@@ -28,7 +28,10 @@ enqueue:
             traceable_producer: true
         job: true
         async_events: true
-        async_commands: true
+        async_commands:
+            enabled: true
+            timeout: 60
+            prefix: ''
 
 services:
     test_enqueue.client.default.traceable_producer:
@@ -122,4 +125,4 @@ services:
     enqueue.events.async_listener:
         class: 'Enqueue\Bundle\Tests\Functional\App\AsyncListener'
         public: true
-        arguments: ['@enqueue.client.default.producer', '@enqueue.events.registry']
+        arguments: ['@enqueue.client.default.producer', '@enqueue.events.registry']