FEATURE: Exit fast-rabbit loop after 600 seconds

Original: b9c3a8c670e544c5926448bfe96ddbb983d00782

Author: paxuclus

Classes/Loop.php

@@ -4,6 +4,7 @@
 namespace Netlogix\JobQueue\FastRabbit;
 
 use Neos\Flow\Annotations as Flow;
+use PhpAmqpLib\Exception\AMQPTimeoutException;
 use t3n\JobQueue\RabbitMQ\Queue\RabbitQueue;
 
 /**
@@ -13,17 +14,44 @@ final class Loop
 {
     protected $queue;
 
-    public function __construct(RabbitQueue $queue)
+    /**
+     * Unix timestamp after which the Loop should exit
+     *
+     * @var int|null
+     */
+    protected $exitAfterTimestamp;
+
+    /**
+     * Timeout in seconds when waiting for new messages
+     *
+     * @var int|null
+     */
+    protected $timeout;
+
+    /**
+     * @param RabbitQueue $queue The Queue to watch
+     * @param int $exitAfter Time in seconds after which the loop should exit
+     */
+    public function __construct(RabbitQueue $queue, int $exitAfter = 0)
     {
         $this->queue = $queue;
+        $this->exitAfterTimestamp = $exitAfter > 0 ? time() + $exitAfter : null;
+        $this->timeout = $exitAfter > 0 ? $exitAfter : null;
     }
 
     public function runMessagesOnWorker(Worker $worker)
     {
         $worker->prepare();
         do {
-            $message = $this->queue->waitAndReserve();
-            $worker->executeMessage($message);
+            try {
+                $message = $this->queue->waitAndReserve($this->timeout);
+                $worker->executeMessage($message);
+            } catch (AMQPTimeoutException $e) {
+            }
+
+            if ($this->exitAfterTimestamp !== null && time() >= $this->exitAfterTimestamp) {
+                break;
+            }
         } while (true);
     }
 }

bin/fast-rabbit

@@ -25,6 +25,6 @@ $messageCache = CacheFactory::get($config);
 $lock = new Lock($config['workerPool']['numberOfWorkers'], $config['workerPool']['lockFileDirectory']);
 
 $worker = new Worker($command, $queue, $queueSettings, $messageCache, $lock);
-$loop = new Loop($queue);
+$loop = new Loop($queue, 600);
 
 $loop->runMessagesOnWorker($worker);