feat(consumer): add a way to stop consumer

src/Clients/Consumer/KafkaConsumer.php

@@ -33,11 +33,11 @@
 
     public function __construct(ConsumerConfig $config, LoggerInterface|null $logger = null)
     {
         $this->logger = $logger ?? new NullLogger();
 
         $this->setupInternalTerminationSignal($config);
 
         $config->getConf()->setErrorCb(
             function (RdKafkaConsumer $kafka, int $err, string $reason): void {
                 $this->logger->error(
                     sprintf('Kafka error: "%s": "%s"', rd_kafka_err2str($err), $reason),
@@ -51,9 +51,9 @@
                 /** @phpstan-var array<string, TopicPartition>|null $partitions */
                 switch ($err) {
                     case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
                         $this->logger->debug(
                             'Assigning partitions',
                             $partitions === null ? [] : array_map(
                                 static fn (TopicPartition $partition): string => (string) $partition->getPartition(),
                                 $partitions,
                             ),
@@ -77,7 +77,7 @@
                         $kafka->assign();
                 }
             };
         $config->getConf()->setRebalanceCb($rebalanceCallback);
 
         parent::__construct($config->getConf());
     }
@@ -120,8 +120,8 @@
                 $consumerRecords,
             ): void {
                 $consumerRecords->add($message);
                 if ($processRecord !== null) {
                     $processRecord($message);
                 }
 
                 if ($consumerRecords->count() === $maxBatchSize) {
@@ -146,6 +146,11 @@
     {
         $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->shouldRun = false;
+        $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);