From 654b8f0a5a712e14fdd5e4682d7897d46e7ef549 Mon Sep 17 00:00:00 2001
From: Ferdinand Kuhl <f.kuhl@digital-competence.de>
Date: Mon, 28 Sep 2020 00:35:32 +0200
Subject: [PATCH] FEATURE: Allow jobManager to be interrupted by system SIGINT

---
 Classes/Command/JobCommandController.php | 12 ++++++++++++
 Classes/InterruptException.php           |  8 ++++++++
 Classes/Job/JobManager.php               | 13 +++++++++++++
 3 files changed, 33 insertions(+)
 create mode 100644 Classes/InterruptException.php

diff --git a/Classes/Command/JobCommandController.php b/Classes/Command/JobCommandController.php
index faf1b8a..07a8b45 100644
--- a/Classes/Command/JobCommandController.php
+++ b/Classes/Command/JobCommandController.php
@@ -12,6 +12,7 @@
  */
 
 use Flowpack\JobQueue\Common\Exception as JobQueueException;
+use Flowpack\JobQueue\Common\InterruptException;
 use Flowpack\JobQueue\Common\Job\JobManager;
 use Flowpack\JobQueue\Common\Queue\Message;
 use Flowpack\JobQueue\Common\Queue\QueueManager;
@@ -74,6 +75,11 @@ public function workCommand($queue, $exitAfter = null, $limit = null, $verbose =
             }
             $this->outputLine('...');
         }
+        if (function_exists('pcntl_signal')) {
+            pcntl_signal(SIGINT, static function () {
+                throw new InterruptException('Interrupted by SIGINT');
+            });
+        }
         $startTime = time();
         $timeout = null;
         $numberOfJobExecutions = 0;
@@ -84,6 +90,12 @@ public function workCommand($queue, $exitAfter = null, $limit = null, $verbose =
             }
             try {
                 $message = $this->jobManager->waitAndExecute($queue, $timeout);
+                $this->jobManager->interruptMe();
+            } catch (InterruptException $exception) {
+                if ($verbose) {
+                    $this->outputLine('Quitting after %d seconds due to received interruption', [time() - $startTime]);
+                }
+                $this->quit();
             } catch (JobQueueException $exception) {
                 $numberOfJobExecutions ++;
                 $this->outputLine('<error>%s</error>', [$exception->getMessage()]);
diff --git a/Classes/InterruptException.php b/Classes/InterruptException.php
new file mode 100644
index 0000000..84a04ef
--- /dev/null
+++ b/Classes/InterruptException.php
@@ -0,0 +1,8 @@
+<?php
+
+namespace Flowpack\JobQueue\Common;
+
+class InterruptException extends Exception
+{
+
+}
diff --git a/Classes/Job/JobManager.php b/Classes/Job/JobManager.php
index 45714cd..0f21824 100644
--- a/Classes/Job/JobManager.php
+++ b/Classes/Job/JobManager.php
@@ -11,6 +11,7 @@
  * source code.
  */
 
+use Flowpack\JobQueue\Common\InterruptException;
 use Flowpack\JobQueue\Common\Queue\QueueInterface;
 use Neos\Cache\Frontend\VariableFrontend;
 use Neos\Flow\Annotations as Flow;
@@ -175,6 +176,18 @@ public function peek(string $queueName, int $limit = 1): array
         }, $messages);
     }
 
+    /**
+     * This method is here to be called by queues if they reached local polling timeouts, if this applies
+     *
+     * @throws InterruptException
+     */
+    public function interruptMe(): void
+    {
+        if (function_exists('pcntl_signal_dispatch')) {
+            pcntl_signal_dispatch();
+        }
+    }
+
     /**
      * Signal that is triggered when a message has been submitted to a queue
      *