task: Switch from symfony/process to react/child-process

The react/child-process Process implementation can work on an event
loop, which also supports periodic timers.

Author: stephanschuler

Classes/Worker.php

@@ -8,9 +8,8 @@
 use Flowpack\JobQueue\Common\Queue\Message;
 use Neos\Cache\Frontend\FrontendInterface;
 use Neos\Flow\Cli\ConsoleOutput;
+use React\ChildProcess\Process;
 use React\EventLoop;
-use Symfony\Component\Process\InputStream;
-use Symfony\Component\Process\Process;
 use t3n\JobQueue\RabbitMQ\Queue\RabbitQueue;
 
 use function array_shift;
@@ -23,8 +22,10 @@ final class Worker
 {
     protected readonly ConsoleOutput $output;
 
+    private readonly EventLoop\LoopInterface $loop;
+
     /**
-     * @var array{input: InputStream, process: Process}[]
+     * @var Process[]
      */
     private array $pool = [];
 
@@ -40,6 +41,7 @@ public function __construct(
         protected readonly FrontendInterface $messageCache,
         protected readonly Lock $lock
     ) {
+        $this->loop = EventLoop\Loop::get();
     }
 
     public function prepare(): void
@@ -102,47 +104,32 @@ public function executeMessage(Message $message): void
         }
     }
 
-    /**
-     * @return array{input: InputStream, process: Process}
-     */
-    private function createProcess(): array
+    private function createProcess(): Process
     {
-        $input = new InputStream();
-        $process = Process::fromShellCommandline(
-            command: $this->command,
-            input: $input,
-            timeout: 0
-        );
-        $process->start();
-        return ['input' => $input, 'process' => $process];
+        $process = new Process($this->command);
+        $timer = $this->loop->addPeriodicTimer(0.01, function () {
+            // TODO: Add keepalive for database if necessary
+        });
+        $process->on('exit', function () use ($timer) {
+            $this->loop->cancelTimer($timer);
+            $this->loop->stop();
+        });
+        $process->start(loop: $this->loop, interval: 0.01);
+        return $process;
     }
 
     private function runFromPool(string $messageCacheIdentifier): Process
     {
         $this->cleanPool();
-        ['input' => $input, 'process' => $process] = array_shift($this->pool);
-        $this->pool[] = $this->createProcess();
-
-        assert($input instanceof InputStream);
+        $process = array_shift($this->pool);
         assert($process instanceof Process);
 
-        $input->write($messageCacheIdentifier . PHP_EOL);
-
-        $loop = EventLoop\Loop::get();
-        $loop->addPeriodicTimer(0.01, function (EventLoop\TimerInterface $timer) use ($process, $loop) {
-            try {
-                fputs(STDOUT, $process->getIncrementalOutput());
-                fputs(STDERR, $process->getIncrementalErrorOutput());
-            } catch (\Throwable $e) {
-            }
+        $process->stdout->on('data', fn ($chunk) => fputs(STDOUT, $chunk));
+        $process->stderr->on('data', fn ($chunk) => fputs(STDERR, $chunk));
 
-            if (!$process->isRunning()) {
-                $loop->cancelTimer($timer);
-                $loop->stop();
-            }
-        });
+        $process->stdin->write($messageCacheIdentifier . PHP_EOL);
 
-        $loop->run();
+        $this->loop->run();
 
         return $process;
     }
@@ -151,7 +138,7 @@ private function cleanPool(): void
     {
         $this->pool = array_filter(
             $this->pool,
-            fn (array $item) => $item['process']->isRunning()
+            fn (Process $process) => $process->isRunning()
         );
         while (count($this->pool) < $this->poolSize) {
             $this->pool[] = $this->createProcess();

composer.json

@@ -13,8 +13,8 @@
         "flowpack/jobqueue-common": "^3.0.0",
         "netlogix/supervisor": "^1.0",
         "php-amqplib/php-amqplib": "^2.11",
+        "react/child-process": "^0.6",
         "react/event-loop": "^1.5",
-        "symfony/process": "^5.4 || ^6.0 || ^7.0",
         "t3n/jobqueue-rabbitmq": "^2.3.0",
         "php": "~8.2 || ~8.3 || ~8.4"
     },