feat: Use React/EventLoop for ticks during sub process execution

The previous setting only reacted to data sent from the child
process to the parent process via STDOUT and STDERR but had no
timeout based callback, so during long-running child processees
with no data interaction, the parent could not e.g. keep its
database alive.

Author: stephanschuler

Classes/Worker.php

@@ -8,11 +8,16 @@
 use Flowpack\JobQueue\Common\Queue\Message;
 use Neos\Cache\Frontend\FrontendInterface;
 use Neos\Flow\Cli\ConsoleOutput;
+use React\EventLoop;
 use Symfony\Component\Process\InputStream;
 use Symfony\Component\Process\Process;
 use t3n\JobQueue\RabbitMQ\Queue\RabbitQueue;
 
 use function array_shift;
+use function fputs;
+
+use const STDERR;
+use const STDOUT;
 
 final class Worker
 {
@@ -60,7 +65,6 @@ public function executeMessage(Message $message): void
                 '<success>Successfully executed job "%s"</success>',
                 [$message->getIdentifier()]
             );
-            $this->output->outputLine('Output: %s', [$process->getOutput()]);
         } else {
             $maximumNumberOfReleases = isset($this->queueSettings['maximumNumberOfReleases'])
                 ? (int) $this->queueSettings['maximumNumberOfReleases']
@@ -70,7 +74,6 @@ public function executeMessage(Message $message): void
                 $releaseOptions = isset($this->queueSettings['releaseOptions']) ? $this->queueSettings['releaseOptions'] : [];
                 $this->queue->release($message->getIdentifier(), $releaseOptions);
                 $this->queue->reQueueMessage($message, $releaseOptions);
-                $this->output->outputLine('Output: %s', [$process->getOutput()]);
                 $this->output->outputLine(
                     '<error>Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - RELEASE</error>',
                     [
@@ -82,7 +85,6 @@ public function executeMessage(Message $message): void
                 );
             } else {
                 $this->queue->abort($message->getIdentifier());
-                $this->output->outputLine('Output: %s', [$process->getOutput()]);
                 $this->output->outputLine(
                     '<error>Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - ABORTING</error>',
                     [
@@ -126,7 +128,22 @@ private function runFromPool(string $messageCacheIdentifier): Process
 
         $input->write($messageCacheIdentifier . PHP_EOL);
 
-        $process->wait();
+        $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) {
+            }
+
+            if (!$process->isRunning()) {
+                $loop->cancelTimer($timer);
+                $loop->stop();
+            }
+        });
+
+        $loop->run();
+
         return $process;
     }
 

composer.json

@@ -13,6 +13,7 @@
         "flowpack/jobqueue-common": "^3.0.0",
         "netlogix/supervisor": "^1.0",
         "php-amqplib/php-amqplib": "^2.11",
+        "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"