Defining before callbacks

docs/advanced-usage/8-before-callbacks.md

@@ -0,0 +1,23 @@
+---
+title: Before callbacks
+weight: 8
+---
+
+Before consuming any message, you can call any callbacks. For example to wait for
+maintenance mode. Callbacks will receive consumer instance as an argument.
+The callbacks get executed in the order they are defined:
+
+```php
+$consumer = \Junges\Kafka\Facades\Kafka::createConsumer()
+    ->withBeforeConsuming(function($consumer) {
+        while (app()->isDownForMaintenance()) {
+            $sleepTimeInSeconds = random_int(1, 5);
+            sleep($sleepTimeInSeconds);
+        }       
+    });
+```
+
+You can add as many callback as you need, so you can divide different tasks into 
+different callbacks.
+
+If callbacks return `false` then consumer will go to stop phase.

src/Config/Config.php

@@ -2,6 +2,7 @@
 
 namespace Junges\Kafka\Config;
 
+use Closure;
 use JetBrains\PhpStorm\Pure;
 use Junges\Kafka\Contracts\Consumer;
 use Junges\Kafka\Contracts\HandlesBatchConfiguration;
@@ -77,7 +78,8 @@ public function __construct(
         private readonly HandlesBatchConfiguration $batchConfig = new NullBatchConfig(),
         private readonly bool $stopAfterLastMessage = false,
         private readonly int $restartInterval = 1000,
-        private readonly array $callbacks = []
+        private readonly array $callbacks = [],
+        private array              $beforeConsumings = [],
     ) {}
 
     public function getCommit(): int
@@ -188,4 +190,9 @@ private function usingSasl(): bool
             && (strtoupper($this->securityProtocol) === static::SASL_PLAINTEXT
                 || strtoupper($this->securityProtocol) === static::SASL_SSL);
     }
+
+    public function getBeforeConsumings(): array
+    {
+        return $this->beforeConsumings;
+    }
 }

src/Consumers/Consumer.php

@@ -101,6 +101,12 @@ public function consume(): void
         }
 
         do {
+            foreach ($this->config->getBeforeConsumings() as $beforeConsuming) {
+                $result = $beforeConsuming(...)();
+                if ($result === false) {
+                    break;
+                }
+            }
             $this->retryable->retry(fn () => $this->doConsume());
             $this->checkForRestart();
         } while (! $this->maxMessagesLimitReached() && ! $this->stopRequested);

src/Consumers/ConsumerBuilder.php

@@ -41,6 +41,7 @@ class ConsumerBuilder implements ConsumerBuilderContract
     protected int $batchSizeLimit = 0;
     protected int $batchReleaseInterval = 0;
     protected bool $stopAfterLastMessage = false;
+    protected array $beforeConsumings = [];
 
     protected function __construct(protected string $brokers, array $topics = [], protected ?string $groupId = null)
     {
@@ -257,6 +258,13 @@ public function stopAfterLastMessage(bool $stopAfterLastMessage = true): self
         return $this;
     }
 
+    public function withBeforeConsuming(callable $callable): self
+    {
+        $this->beforeConsumings[] = $callable;
+
+        return $this;
+    }
+
     /** @inheritDoc */
     public function build(): MessageConsumer
     {
@@ -276,6 +284,7 @@ public function build(): MessageConsumer
             batchConfig: $this->getBatchConfig(),
             stopAfterLastMessage: $this->stopAfterLastMessage,
             callbacks: $this->callbacks,
+            beforeConsumings: $this->beforeConsumings,
         );
 
         return new Consumer($config, $this->deserializer, $this->committerFactory);

tests/Consumers/ConsumerTest.php

@@ -288,4 +288,41 @@ function (ConsumerMessage $message) {
         //finaly only one message should be consumed
         $this->assertEquals(1, $consumer->consumedMessagesCount());
     }
+
+    public function testItRunCallbacksBeforeConsume()
+    {
+        $fakeHandler = new FakeHandler();
+
+        $message = new Message();
+        $message->err = 0;
+        $message->key = 'key';
+        $message->topic_name = 'test-topic';
+        $message->payload = '{"body": "message payload"}';
+        $message->offset = 0;
+        $message->partition = 1;
+        $message->headers = [];
+
+        $this->mockConsumerWithMessage($message);
+
+        $this->mockProducer();
+
+        $config = new Config(
+            broker: 'broker',
+            topics: ['test-topic'],
+            securityProtocol: 'security',
+            commit: 1,
+            groupId: 'group',
+            consumer: $fakeHandler,
+            sasl: null,
+            dlq: null,
+            maxMessages: 1,
+            maxCommitRetries: 1,
+            beforeConsumings: [function(){var_dump('HERE');},function(){return false;},function(){var_dump('THERE');}]
+        );
+
+        $consumer = new Consumer($config, new JsonDeserializer());
+        $consumer->consume();
+
+        $this->assertInstanceOf(ConsumedMessage::class, $fakeHandler->lastMessage());
+    }
 }