fix: Prevent MissingInputExceptions when loop restarts

Our FastRabbit loop will retart every 6 hours. This results in child
processes loosing their STDIN, so they will throw an exceptions.

Since this exception is expected, we don't need to log it.

Author: stephanschuler

Classes/Loop.php

@@ -45,11 +45,14 @@ public function runMessagesOnWorker(Worker $worker)
         do {
             try {
                 $message = $this->queue->waitAndReserve($this->timeout);
-                $worker->executeMessage($message);
+                if ($message) {
+                    $worker->executeMessage($message);
+                }
             } catch (AMQPTimeoutException $e) {
             }
 
             if ($this->exitAfterTimestamp !== null && time() >= $this->exitAfterTimestamp) {
+                $worker->shutdownObject();
                 break;
             }
         } while (true);

Classes/PreventLoggingOfMissingInput/JobCommandInitializationAspect.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Netlogix\JobQueue\FastRabbit\NoFoo;
+
+use Flowpack\JobQueue\Common\Command\JobCommandController;
+use Neos\Flow\Annotations as Flow;
+use Neos\Flow\Aop\JoinPointInterface;
+use Neos\Flow\Cli\Exception\StopCommandException;
+use Neos\Flow\Reflection\ClassReflection;
+use Symfony\Component\Console\Exception\MissingInputException;
+
+/**
+ * There might be orphaned worker processes when the parent process
+ * of FastRabbit loop restarts due to its max wait time.
+ *
+ * Those are still waiting for message identifiers, which can never
+ * be fulfilled once their STDIN closes, so a MissingInputException
+ * gets thrown.
+ *
+ * This is expected and will happen every 6 hours due to the current
+ * configuration. We don't need those exceptions tracked. So redirect
+ * to StopCommandException instead.
+ */
+#[Flow\Aspect]
+#[Flow\Proxy(false)]
+class JobCommandInitializationAspect
+{
+    #[Flow\Around('within(' . JobCommandController::class . ') && method(.*->mapRequestArgumentsToControllerArguments())')]
+    public function preventLoggingOfMissingInputExceptions(JoinPointInterface $joinPoint): void
+    {
+        $jobCommandController = $joinPoint->getProxy();
+        assert($jobCommandController instanceof JobCommandController);
+
+        $reflection = new ClassReflection($jobCommandController);
+
+        $commandMethodName = $reflection
+            ->getProperty('commandMethodName')
+            ->getValue($jobCommandController);
+
+        if ($commandMethodName !== 'executeCommand') {
+            $joinPoint->getAdviceChain()->proceed($joinPoint);
+            return;
+        }
+
+        try {
+            $joinPoint->getAdviceChain()->proceed($joinPoint);
+        } catch (MissingInputException $e) {
+            throw new StopCommandException();
+        }
+    }
+}

Classes/Worker.php

@@ -49,6 +49,15 @@ public function __construct(
         $this->poolSize = max(0, (int) ($queueSettings['poolSize'] ?? 1));
     }
 
+    public function shutdownObject()
+    {
+        foreach ($this->pool as $process) {
+            $process->terminate();
+            $process->stdin->close();
+        }
+        $this->pool = [];
+    }
+
     public function prepare(): void
     {
         $this->fillPool($this->poolSize);