TASK: Netlogix.JobQueue.FastRabbit depends on Netlogix.Supervisor

Original: dd0a17b668fa7b1ea93608f936a88d3feb00c164

Author: stephanschuler

Classes/Command/SupervisorCommandController.php

@@ -5,96 +5,132 @@
 
 use Flowpack\JobQueue\Common\Queue\QueueManager;
 use Neos\Flow\Annotations as Flow;
+use Neos\Flow\Configuration\ConfigurationManager;
+use Neos\Flow\Core\Booting\Scripts;
+use Neos\Utility\ObjectAccess;
+use function defined;
+use function preg_replace;
+use function rtrim;
+use function sprintf;
+use function strtolower;
+use function trim;
 
 class ConfigurationFactory
 {
-    const __CONFIG_FILE__ = '__CONFIG_FILE__';
-    const __QUEUE_NAME__ = '__QUEUE_NAME__';
-    const __JOB_NAME__ = '__JOB_NAME__';
-    const __CONTEXT__ = '__CONTEXT__';
-    const __NUMPROCS__ = '__NUMPROCS__';
-
-    /**
-     * @var string
-     * @Flow\InjectConfiguration(package="Netlogix.JobQueue.FastRabbit", path="supervisor.contextName")
-     */
-    protected $contextName;
-
     /**
      * @var string
-     * @Flow\InjectConfiguration(package="Netlogix.JobQueue.FastRabbit", path="supervisor.programTemplate")
+     * @Flow\InjectConfiguration(package="Netlogix.JobQueue.FastRabbit", path="supervisor.applicationName")
      */
-    protected $programTemplate;
+    protected $applicationName;
 
     /**
-     * @var string
-     * @Flow\InjectConfiguration(package="Netlogix.JobQueue.FastRabbit", path="supervisor.groupTemplate")
+     * @var QueueManager
      */
-    protected $groupTemplate;
+    protected $queueManager;
 
     /**
-     * @var QueueManager
-     * @Flow\Inject
+     * @var ConfigurationManager
      */
-    protected $queueManager;
+    private $configurationManager;
 
-    public function getShortNameForQueueName(string $queueName): string
+    public function injectQueueManager(QueueManager $queueManager): void
     {
-        return preg_replace('%[^a-z0-9]%iUm', '-', strtolower($queueName));
+        $this->queueManager = $queueManager;
     }
 
-    public function getJobNameForQueueName(string $queueName): string
+    public function injectConfigurationManager(ConfigurationManager $configurationManager): void
     {
-        return $this->getContextName() . '-' . $this->getShortNameForQueueName($queueName);
+        $this->configurationManager = $configurationManager;
     }
 
-    public function buildJobConfigurationForQueue(string $queueName): string
+    /**
+     * @param string $queueName
+     * @return array<string, mixed>
+     * @throws \Flowpack\JobQueue\Common\Exception
+     * @throws \Neos\Flow\Configuration\Exception\InvalidConfigurationTypeException
+     * @throws \Neos\Flow\Exception
+     */
+    public function buildJobConfiguration(string $queueName): array
     {
-        $jobName = $this->getJobNameForQueueName($queueName);
-        $queueSettings = $this->queueManager->getQueueSettings($queueName);
-        $fastRabbitSettings = $queueSettings['fastRabbit'] ?? [];
-        $numProcs = (int)($fastRabbitSettings['numProcs'] ?? 1);
-
-        $config = $this->programTemplate;
-        $config = str_replace(self::__CONFIG_FILE__, $this->getJobConfigurationFile($queueName), $config);
-        $config = str_replace(self::__QUEUE_NAME__, $queueName, $config);
-        $config = str_replace(self::__JOB_NAME__, $jobName, $config);
-        $config = str_replace(self::__CONTEXT__, $this->contextName, $config);
-        $config = str_replace(self::__NUMPROCS__, $numProcs, $config);
-
-        return $config;
+        assert(defined('FLOW_PATH_FLOW'));
+        assert(defined('FLOW_PATH_TEMPORARY_BASE'));
+
+        $flowSettings = (array)$this->configurationManager->getConfiguration(
+            ConfigurationManager::CONFIGURATION_TYPE_SETTINGS,
+            'Neos.Flow'
+        );
+
+        $command = Scripts::buildPhpCommand(
+            $flowSettings
+        );
+        $command .= sprintf(
+            ' %s %s --queue=%s',
+            escapeshellarg(\FLOW_PATH_FLOW . 'Scripts/flow.php'),
+            escapeshellarg('flowpack.jobqueue.common:job:execute'),
+            escapeshellarg($queueName)
+        );
+
+        $workerPool = (array)$this->configurationManager->getConfiguration(
+            ConfigurationManager::CONFIGURATION_TYPE_SETTINGS,
+            'Netlogix.JobQueue.FastRabbit.supervisor.workerPool'
+        );
+
+        $jobConfig = [
+            'command' => $command,
+            'queueSettings' => $this->queueManager->getQueueSettings($queueName),
+            'cacheConfiguration' => $this->configurationManager->getConfiguration(
+                ConfigurationManager::CONFIGURATION_TYPE_CACHES,
+                'FlowPackJobQueueCommon_MessageCache'
+            ),
+            'queueName' => $queueName,
+            'temporaryDirectoryBase' => \FLOW_PATH_TEMPORARY_BASE,
+            'applicationIdentifier' => ObjectAccess::getPropertyPath($flowSettings, 'cache.applicationIdentifier'),
+            'contextString' => ObjectAccess::getPropertyPath($flowSettings, 'core.context'),
+            'workerPool' => $workerPool,
+        ];
+
+        return $jobConfig;
     }
 
-    public function buildGroupConfigurationForQueues(string ...$queueNames): string
+    public function getJobName(string $queueName): string
     {
-        $jobNames = array_map([$this, 'getJobNameForQueueName'], $queueNames);
-
-        $programs = $this->groupTemplate;
-        $programs = str_replace('__PROGRAMS__', join(',', $jobNames), $programs);
-        $programs = str_replace('__CONTEXT__', $this->contextName, $programs);
-
-        return $programs;
+        return self::removeInvalidCharactersFromSupervisorIdentifier($this->applicationName . '-' . $queueName);
     }
 
-    public function getJobConfigurationFile(string $queueName): string
+    public function getJobConfigurationFilePath(string $queueName): string
     {
         return $this->getJobFilePath($queueName, 'json');
     }
 
-    public function getJobSupervisorFile(string $queueName): string
+    public function getJobSupervisorFilePath(string $queueName): string
     {
         return $this->getJobFilePath($queueName, 'conf');
     }
 
+    public function getNumberOfProcesses(string $queueName): int
+    {
+        $queueSettings = $this->queueManager->getQueueSettings($queueName);
+        return (int)(ObjectAccess::getPropertyPath($queueSettings, 'fastRabbit.numProcs') ?? 1);
+    }
+
     protected function getJobFilePath(string $queueName, string $suffix): string
     {
-        $jobName = $this->getJobNameForQueueName($queueName);
+        assert(defined('FLOW_PATH_CONFIGURATION'));
+        $jobName = $this->getJobName($queueName);
         $pathPrefix = rtrim(FLOW_PATH_CONFIGURATION, '/') . '/Supervisor/';
         return sprintf($pathPrefix . 'program-%s.%s', $jobName, $suffix);
     }
 
-    protected function getContextName(): string
+    private static function removeInvalidCharactersFromSupervisorIdentifier(string $subject): string
     {
-        return trim(preg_replace('%[^\\pL]+%iUum', '-', $this->contextName), '-');
+        /**
+         * \p{xx}   A character with the xx unicode property.   https://php.net/manual/en/regexp.reference.escape.php
+         * \pL      Every Letter                                https://php.net/manual/en/regexp.reference.unicode.php
+         */
+        $cleanupPattern = '%[^\\pL\\d]+%iUum';
+
+        $subject = (string)preg_replace($cleanupPattern, '-', $subject);
+        $subject = trim($subject, '-');
+        return strtolower($subject);
     }
 }