Add idle callback to the Consumer

Author: grongor

README.md

@@ -67,6 +67,10 @@ messages. Simply pass it a message you wish to commit. For most cases it's recom
 and manually commit each message, so that you don't ever "lose" a message if your application dies
 in the middle of the processing.
 
+You can also (optionally) set an idle callback using `setIdleCallback` method. This callback will be called whenever
+there are no messages to return. The idle interval is then equal to the `timeout` parameter if provided, or to the
+`consumerRequestTimeoutMs` option if set, otherwise to the proxy configuration option `consumer.request.timeout.ms`.
+
 ```php
 $consumerFactory = new ConsumerFactory($restClient);
 $consumer = $consumerFactory->create('your-consumer-group', Subscription::topic('your-topic'));

src/Consumer.php

@@ -9,6 +9,7 @@
 use Grongor\KafkaRest\Api\Value\Response\Message;
 use Grongor\KafkaRest\Exception\ConsumerClosed;
 use Throwable;
+use function count;
 
 final class Consumer
 {
@@ -18,6 +19,9 @@ final class Consumer
  /** @var Api\Value\Response\Consumer */
  private $consumer;
 
+ /** @var callable|null */
+ private $idleCallback;
+
  public function __construct(RestClient $client, Api\Value\Response\Consumer $consumer)
  {
  $this->client = $client;
@@ -33,6 +37,11 @@ public function __destruct()
  $this->close();
  }
 
+ public function setIdleCallback(callable $idleCallback) : void
+ {
+ $this->idleCallback = $idleCallback;
+ }
+
  /**
  * @return Message[]
  */
@@ -44,7 +53,14 @@ public function consume(?int $timeout = null, ?int $maxBytes = null) : iterable
 
  try {
  while (true) {
- yield from $this->client->getConsumerMessages($this->consumer, $timeout, $maxBytes);
+ $messages = $this->client->getConsumerMessages($this->consumer, $timeout, $maxBytes);
+ if ($this->idleCallback !== null && count($messages) === 0) {
+ ($this->idleCallback)();
+
+ continue;
+ }
+
+ yield from $messages;
  }
  } catch (Throwable $throwable) {
  $this->close();

tests/ConsumerTest.php

@@ -71,6 +71,67 @@ public function testConsume(?int $timeout, ?int $maxBytes) : void
  }
  }
 
+ /**
+ * @dataProvider providerConsume
+ */
+ public function testConsumeWithIdleCallback(?int $timeout, ?int $maxBytes) : void
+ {
+ $consumerVO = new \Grongor\KafkaRest\Api\Value\Response\Consumer();
+
+ $firstMessages = [];
+ $secondMessages = [new Message(), new Message()];
+ $thirdMessages = [];
+
+ $client = Mockery::mock(RestClient::class);
+ $client->shouldReceive('getConsumerMessages')
+ ->once()
+ ->with($consumerVO, $timeout, $maxBytes)
+ ->andReturns($firstMessages);
+ $client->shouldReceive('getConsumerMessages')
+ ->once()
+ ->with($consumerVO, $timeout, $maxBytes)
+ ->andReturns($secondMessages);
+ $client->shouldReceive('getConsumerMessages')
+ ->once()
+ ->with($consumerVO, $timeout, $maxBytes)
+ ->andReturns($thirdMessages);
+ $client->shouldReceive('getConsumerMessages')
+ ->once()
+ ->andThrow(new Exception('some exception'));
+
+ $client->shouldReceive('deleteConsumer')
+ ->once()
+ ->with($consumerVO);
+
+ $callbackCalledTimes = 0;
+
+ $consumer = new Consumer($client, $consumerVO);
+ $consumer->setIdleCallback(static function () use (&$callbackCalledTimes) : void {
+ $callbackCalledTimes++;
+ });
+ $messages = $consumer->consume($timeout, $maxBytes);
+
+ self::assertInstanceOf(Generator::class, $messages);
+
+ self::assertTrue($messages->valid());
+ self::assertSame($secondMessages[0], $messages->current());
+
+ $messages->next();
+ self::assertTrue($messages->valid());
+ self::assertSame($secondMessages[1], $messages->current());
+
+ self::assertTrue($messages->valid());
+
+ try {
+ $messages->next();
+ self::fail('Expected an exception');
+ } catch (ConsumerClosed $exception) {
+ self::assertFalse($messages->valid());
+ }
+
+ self::assertSame(2, $callbackCalledTimes);
+ }
+
  /**
  * @return iterable<array<mixed>>
  */