Graceful max execution time for batch consumer

DependencyInjection/Configuration.php

@@ -273,6 +273,12 @@ protected function addBatchConsumers(ArrayNodeDefinition $node)
                             ->scalarNode('timeout_wait')->defaultValue(3)->end()
                             ->scalarNode('idle_timeout_exit_code')->end()
                             ->scalarNode('keep_alive')->defaultFalse()->end()
+                            ->arrayNode('graceful_max_execution')
+                                ->canBeUnset()
+                                ->children()
+                                    ->integerNode('timeout')->end()
+                                ->end()
+                            ->end()
                             ->scalarNode('auto_setup_fabric')->defaultTrue()->end()
                             ->arrayNode('qos_options')
                                 ->children()

DependencyInjection/OldSoundRabbitMqExtension.php

@@ -414,6 +414,13 @@ protected function loadBatchConsumers()
                 $definition->addMethodCall('setIdleTimeout', array($consumer['idle_timeout']));
             }
 
+            if (isset($consumer['graceful_max_execution'])) {
+                $definition->addMethodCall(
+                    'setGracefulMaxExecutionDateTimeFromSecondsInTheFuture',
+                    array($consumer['graceful_max_execution']['timeout'])
+                );
+            }
+
             if (!$consumer['auto_setup_fabric']) {
                 $definition->addMethodCall('disableAutoSetupFabric');
             }

README.md

@@ -811,6 +811,8 @@ batch_consumers:
         auto_setup_fabric: false
         idle_timeout_exit_code: -2
         keep_alive: false
+        graceful_max_execution:
+            timeout: 60
 ```
 
 *Note*: If the `keep_alive` option is set to `true`, `idle_timeout_exit_code` will be ignored and the consumer process continues.

RabbitMq/BatchConsumer.php

@@ -7,22 +7,22 @@
 use PhpAmqpLib\Exception\AMQPTimeoutException;
 use PhpAmqpLib\Message\AMQPMessage;
 
-final class BatchConsumer extends BaseAmqp implements DequeuerInterface
+class BatchConsumer extends BaseAmqp implements DequeuerInterface
 {
     /**
      * @var \Closure|callable
      */
-    private $callback;
+    protected $callback;
 
     /**
      * @var bool
      */
-    private $forceStop = false;
+    protected $forceStop = false;
 
     /**
      * @var int
      */
-    private $idleTimeout = 0;
+    protected $idleTimeout = 0;
 
     /**
      * @var bool
@@ -32,32 +32,54 @@ final class BatchConsumer extends BaseAmqp implements DequeuerInterface
     /**
      * @var int
      */
-    private $idleTimeoutExitCode;
+    protected $idleTimeoutExitCode;
 
     /**
      * @var int
      */
-    private $memoryLimit = null;
+    protected $memoryLimit = null;
 
     /**
      * @var int
      */
-    private $prefetchCount;
+    protected $prefetchCount;
 
     /**
      * @var int
      */
-    private $timeoutWait = 3;
+    protected $timeoutWait = 3;
 
     /**
      * @var array
      */
-    private $messages = array();
+    protected $messages = array();
 
     /**
      * @var int
      */
-    private $batchCounter = 0;
+    protected $batchCounter = 0;
+
+    /**
+     * @var \DateTime|null DateTime after which the consumer will gracefully exit. "Gracefully" means, that
+     *      any currently running consumption will not be interrupted.
+     */
+    protected $gracefulMaxExecutionDateTime;
+
+    /**
+     * @param \DateTime|null $dateTime
+     */
+    public function setGracefulMaxExecutionDateTime(\DateTime $dateTime = null)
+    {
+        $this->gracefulMaxExecutionDateTime = $dateTime;
+    }
+
+    /**
+     * @param int $secondsInTheFuture
+     */
+    public function setGracefulMaxExecutionDateTimeFromSecondsInTheFuture($secondsInTheFuture)
+    {
+        $this->setGracefulMaxExecutionDateTime(new \DateTime("+{$secondsInTheFuture} seconds"));
+    }
 
     /**
      * @param   \Closure|callable    $callback
@@ -80,6 +102,7 @@ public function consume()
                 $this->batchConsume();
             }
 
+            $this->checkGracefulMaxExecutionDateTime();
             $this->maybeStopConsumer();
 
             $timeout = $this->isEmptyBatch() ? $this->getIdleTimeout() : $this->getTimeoutWait();
@@ -530,4 +553,24 @@ public function getMemoryLimit()
     {
         return $this->memoryLimit;
     }
+
+    /**
+     * Check graceful max execution date time and stop if limit is reached
+     *
+     * @return void
+     */
+    private function checkGracefulMaxExecutionDateTime()
+    {
+        if (!$this->gracefulMaxExecutionDateTime) {
+            return;
+        }
+
+        $now = new \DateTime();
+
+        if ($this->gracefulMaxExecutionDateTime > $now) {
+            return;
+        }
+
+        $this->forceStopConsumer();
+    }
 }