Merge pull request #5 from ensi-platform/task-82065

#82065 signals

Author: arrilot

README.md

@@ -124,6 +124,12 @@ class ConsumeMessageJob implements ShouldQueue
 
 ```
 
+### Handling signals
+
+`php artisan kafka:consume ...` command can be configured to gracefully stop after receiving some OS signals.  
+Such signals can be set in the `stop_signals` key of the package config, e.g `'stop_signals' => [SIGINT, SIGQUIT]`.  
+You can use any of the constants defined by the pcntl extension https://www.php.net/manual/en/pcntl.constants.php  
+
 ## Testing
 
 ```bash

composer.json

@@ -50,7 +50,10 @@
         "test-coverage": "vendor/bin/phpunit --coverage-html coverage"
     },
     "config": {
-        "sort-packages": true
+        "sort-packages": true,
+        "allow-plugins": {
+            "pestphp/pest-plugin": true
+        }
     },
     "extra": {
         "laravel": {

src/Commands/KafkaConsumeCommand.php

@@ -6,9 +6,10 @@
 use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
 use Ensi\LaravelPhpRdKafkaConsumer\ProcessorData;
 use Illuminate\Console\Command;
+use Symfony\Component\Console\Command\SignalableCommandInterface;
 use Throwable;
 
-class KafkaConsumeCommand extends Command
+class KafkaConsumeCommand extends Command implements SignalableCommandInterface
 {
     /**
      * The name and signature of the console command.
@@ -26,11 +27,32 @@ class KafkaConsumeCommand extends Command
      */
     protected $description = 'Consume concrete topic';
 
+    protected ?HighLevelConsumer $consumer = null;
+
+    public function getStopSignalsFromConfig(): array
+    {
+        return config('kafka-consumer.stop_signals', []);
+    }
+
+    public function getSubscribedSignals(): array
+    {
+        return $this->getStopSignalsFromConfig();
+    }
+
+    public function handleSignal(int $signal): void
+    {
+        if ($this->consumer && in_array($signal, $this->getStopSignalsFromConfig())) {
+            $this->line("Stopping the consumer...");
+            $this->consumer->forceStop();
+        }
+    }
+
     /**
      * Execute the console command.
      */
     public function handle(HighLevelConsumer $highLevelConsumer): int
     {
+        $this->consumer = $highLevelConsumer;
         $topic = $this->argument('topic');
         $consumer = $this->argument('consumer');
         $availableConsumers = array_keys(config('kafka.consumers', []));

src/HighLevelConsumer.php

@@ -14,6 +14,8 @@ class HighLevelConsumer
 {
     protected ?KafkaConsumer $consumer;
 
+    protected $forceStop = false;
+
     public function __construct(
         protected KafkaManager $kafkaManager,
         protected Pipeline $pipeline
@@ -29,6 +31,13 @@ public function for(?string $consumerName): static
         return $this;
     }
 
+    public function forceStop(): static
+    {
+        $this->forceStop = true;
+
+        return $this;
+    }
+
     /**
      * @throws KafkaException
      * @throws RdKafkaException
@@ -115,6 +124,6 @@ protected function shouldBeStopped(int|float $startTime, int $eventsProcessed, C
             return true;
         }
 
-        return false;
+        return $this->forceStop;
     }
 }