feat: introduce KafkaProducerWrapper (#228)

Author: simPod

src/Clients/Producer/KafkaProducer.php

@@ -15,6 +15,7 @@
 use const RD_KAFKA_PARTITION_UA;
 use const RD_KAFKA_RESP_ERR_NO_ERROR;
 
+/** @deprecated Use {@see \SimPod\Kafka\Clients\Producer\KafkaProducerWrapper} instead */
 class KafkaProducer extends Producer
 {
     // phpcs:disable Cdn77.NamingConventions.ValidConstantName.ClassConstantNotUpperCase

src/Clients/Producer/KafkaProducerWrapper.php

@@ -0,0 +1,86 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SimPod\Kafka\Clients\Producer;
+
+use Closure;
+use InvalidArgumentException;
+use RdKafka\Producer;
+use RuntimeException;
+
+use function assert;
+use function sprintf;
+
+use const RD_KAFKA_PARTITION_UA;
+use const RD_KAFKA_RESP_ERR_NO_ERROR;
+
+final readonly class KafkaProducerWrapper
+{
+    private const int RdKafkaMsgFCopy = 0;
+
+    public Producer $producer;
+
+    /** @var (Closure(self):void)|null */
+    private Closure|null $exitCallback;
+
+    /** @param (Closure(self):void)|null $exitCallback */
+    public function __construct(ProducerConfig $config, callable|null $exitCallback = null)
+    {
+        $this->exitCallback = $exitCallback;
+        $this->producer = new Producer($config->getConf());
+    }
+
+    public function __destruct()
+    {
+        if ($this->exitCallback === null) {
+            return;
+        }
+
+        ($this->exitCallback)($this);
+    }
+
+    /** @param array<string, string>|null $headers */
+    public function produce(
+        string $topicName,
+        int|null $partition,
+        string $value,
+        string|null $key = null,
+        array|null $headers = null,
+        int|null $timestampMs = null,
+    ): void {
+        if ($partition < 0) {
+            throw new InvalidArgumentException(
+                sprintf('Invalid partition: %d. Partition number should always be non-negative or null.', $partition),
+            );
+        }
+
+        $topic = $this->producer->newTopic($topicName);
+        $topic->producev(
+            $partition ?? RD_KAFKA_PARTITION_UA,
+            self::RdKafkaMsgFCopy,
+            $value,
+            $key,
+            $headers,
+            $timestampMs ?? 0,
+        );
+        $this->producer->poll(0);
+    }
+
+    public function flushMessages(int $timeoutMs = 10000): void
+    {
+        $result = null;
+        for ($flushRetries = 0; $flushRetries < 10; $flushRetries++) {
+            $result = $this->producer->flush($timeoutMs);
+            if ($result === RD_KAFKA_RESP_ERR_NO_ERROR) {
+                break;
+            }
+        }
+
+        assert($result !== null);
+
+        if ($result !== RD_KAFKA_RESP_ERR_NO_ERROR) {
+            throw new RuntimeException('Was unable to flush, messages might be lost!', $result);
+        }
+    }
+}

tests/Clients/Consumer/TestProducer.php renamed to tests/Clients/Consumer/Fixture/TestProducer.php

@@ -2,22 +2,22 @@
 
 declare(strict_types=1);
 
-namespace SimPod\Kafka\Tests\Clients\Consumer;
+namespace SimPod\Kafka\Tests\Clients\Consumer\Fixture;
 
-use SimPod\Kafka\Clients\Producer\KafkaProducer;
+use SimPod\Kafka\Clients\Producer\KafkaProducerWrapper;
 use SimPod\Kafka\Clients\Producer\ProducerConfig;
 
 use function Safe\gethostname;
 
 final class TestProducer
 {
-    private KafkaProducer $producer;
+    private KafkaProducerWrapper $producer;
 
     public function __construct()
     {
-        $this->producer = new KafkaProducer(
+        $this->producer = new KafkaProducerWrapper(
             $this->getConfig(),
-            static function (KafkaProducer $producer): void {
+            static function (KafkaProducerWrapper $producer): void {
                 $producer->flushMessages(5000);
             },
         );

tests/Clients/Consumer/KafkaBatchConsumerTest.php

@@ -4,35 +4,40 @@
 
 namespace SimPod\Kafka\Tests\Clients\Consumer;
 
+use PHPUnit\Framework\Attributes\CoversClass;
 use PHPUnit\Framework\TestCase;
 use RdKafka\Message;
 use SimPod\Kafka\Clients\Consumer\ConsumerConfig;
 use SimPod\Kafka\Clients\Consumer\ConsumerRecords;
 use SimPod\Kafka\Clients\Consumer\KafkaConsumer;
+use SimPod\Kafka\Clients\Producer\KafkaProducerWrapper;
+use SimPod\Kafka\Tests\Clients\Consumer\Fixture\TestProducer;
 
 use function mt_rand;
 use function Safe\gethostname;
 
+#[CoversClass(KafkaConsumer::class)]
+#[CoversClass(KafkaProducerWrapper::class)]
 final class KafkaBatchConsumerTest extends TestCase
 {
-    public const string PAYLOAD = 'Tasty, chilled pudding is best flavored with juicy lime.';
-    public const string TOPIC = 'kafka-batch-consumer';
+    public const string Payload = 'Tasty, chilled pudding is best flavored with juicy lime.';
+    public const string Topic = 'kafka-batch-consumer';
 
     public function testMaxBatchSize(): void
     {
         $testProducer = new TestProducer();
         for ($i = 0; $i < 100; $i++) {
-            $testProducer->run(self::TOPIC, self::PAYLOAD);
+            $testProducer->run(self::Topic, self::Payload);
         }
 
         $consumer = new KafkaConsumer($this->getConfig());
-        $consumer->subscribe([self::TOPIC]);
+        $consumer->subscribe([self::Topic]);
 
         $consumer->startBatch(
             90,
             10000,
             static function (Message $message): void {
-                self::assertSame(self::PAYLOAD, $message->payload);
+                self::assertSame(self::Payload, $message->payload);
             },
             static function (ConsumerRecords $consumerRecords) use ($consumer): void {
                 self::assertCount(90, $consumerRecords);

tests/Clients/Consumer/KafkaTest.php

@@ -8,7 +8,7 @@
 use PHPUnit\Framework\TestCase;
 use SimPod\Kafka\Clients\Consumer\ConsumerConfig;
 use SimPod\Kafka\Clients\Consumer\KafkaConsumer;
-use SimPod\Kafka\Tests\Clients\Consumer\TestProducer;
+use SimPod\Kafka\Tests\Clients\Consumer\Fixture\TestProducer;
 
 use function mt_rand;
 use function Safe\gethostname;