feat(consumer): add a way to stop consumer

src/Clients/Consumer/KafkaConsumer.php

@@ -120,22 +120,22 @@
                 $consumerRecords,
             ): void {
                 $consumerRecords->add($message);
                 if ($processRecord !== null) {
                     $processRecord($message);
                 }
 
                 if ($consumerRecords->count() === $maxBatchSize) {
                     if ($onBatchProcessed !== null && ! $consumerRecords->isEmpty()) {
                         $onBatchProcessed($consumerRecords);
                     }
 
                     $consumerRecords->clear();
                     $batchTime->reset($timeoutMs, new DateTimeImmutable());
 
                     return;
                 }
 
                 $this->checkBatchTimedOut($timeoutMs, $batchTime, $onBatchProcessed, $consumerRecords)();
             },
             $this->checkBatchTimedOut($timeoutMs, $batchTime, $onBatchProcessed, $consumerRecords),
             $this->checkBatchTimedOut($timeoutMs, $batchTime, $onBatchProcessed, $consumerRecords),
@@ -144,8 +144,13 @@
 
     public function shutdown(): void
     {
         $this->logger->debug('Shutting down');
 
+        $this->stop();
+    }
+
+    public function stop(): void
+    {
         $this->shouldRun = false;
     }
 
@@ -160,7 +165,9 @@
         callable|null $onPartitionEof = null,
         callable|null $onTimedOut = null,
     ): void {
-        $this->registerSignals($this->shouldRun);
+        $this->shouldRun = true;
+        $terminationCallback = fn () => $this->stop();
+        $this->registerSignals($terminationCallback);
 
         while ($this->shouldRun) {
             $message = $this->consume($timeoutMs);

src/Clients/Consumer/WithSignalControl.php

@@ -4,6 +4,8 @@
 
 namespace SimPod\Kafka\Clients\Consumer;
 
+use Safe\Exceptions\PcntlException;
+
 use function Safe\pcntl_signal;
 use function Safe\pcntl_sigprocmask;
 
@@ -22,12 +24,13 @@ private function setupInternalTerminationSignal(ConsumerConfig $config): void
         $config->set('internal.termination.signal', SIGIO);
     }
 
-    private function registerSignals(bool &$shouldRun): void
+    /**
+     * @param callable():mixed $terminationCallback
+     *
+     * @throws PcntlException
+     */
+    private function registerSignals(callable $terminationCallback): void
     {
-        $terminationCallback = static function () use (&$shouldRun): void {
-            $shouldRun = false;
-        };
-
         pcntl_signal(SIGTERM, $terminationCallback);
         pcntl_signal(SIGINT, $terminationCallback);
         pcntl_signal(SIGHUP, $terminationCallback);