Producer send with header support key-value and RecordHeader (#55)

* Add product and consume with header

* producer send with header support key-value and RecordHeader

* Fix

* Fix

* Fix

Author: Yurunsoft

.github/workflows/ci.yml

@@ -40,6 +40,7 @@ jobs:
         run: |
           docker exec swoole composer update
           docker exec kafka1 /opt/kafka/bin/kafka-topics.sh --zookeeper zookeeper:2181 --create --partitions 3 --replication-factor 1 --topic test
+          docker exec kafka1 /opt/kafka/bin/kafka-topics.sh --zookeeper zookeeper:2181 --create --partitions 3 --replication-factor 1 --topic test-header
 
       - name: plaintext-test
         run: |

doc/producer.en.md

@@ -47,6 +47,7 @@ If partition === null && key === null, then use Round Robin to select partition
 ```php
 use longlang\phpkafka\Producer\Producer;
 use longlang\phpkafka\Producer\ProducerConfig;
+use longlang\phpkafka\Protocol\RecordBatch\RecordHeader;
 
 $config = new ProducerConfig();
 $config->setBootstrapServer('127.0.0.1:9092');
@@ -57,6 +58,14 @@ $topic = 'test';
 $value = (string) microtime(true);
 $key = uniqid('', true);
 $producer->send('test', $value, $key);
+
+// set headers
+// key-value or use RecordHeader
+$headers = [
+    'key1' => 'value1',
+    (new RecordHeader())->setHeaderKey('key2')->setValue('value2'),
+];
+$producer->send('test', $value, $key, $headers);
 ```
 
 ## Send batch messages
@@ -83,14 +92,17 @@ $producer->sendBatch([
 ```
 
 ## SASL Support
+
 ### Configuration
+
 | Key | Description | Default |
 | - | - | - |
 | type | SASL Authentication Type. PLAIN is ``\longlang\phpkafka\Sasl\PlainSasl::class``| ''|
 | username | username  | '' |
 | password | password  | '' |
 
 **Example**
+
 ```php
 use longlang\phpkafka\Producer\ProducerConfig;
 use longlang\phpkafka\Producer\Producer;
@@ -106,13 +118,14 @@ $producer = new Producer($config);
 // ....  Your Business Code
 ```
 
-
 ## SSL Support
+
 Class `longlang\phpkafka\Config\SslConfig`
 
 > You can pass an array to a constructor.
 
 ### Configuration keys
+
 | Key | Description | Default |
 | - | - | - |
 | open  | Enable SSL  | `false` |
@@ -145,4 +158,4 @@ $sslConfig->setCafile("/kafka-client/.github/kafka/cert/ca-cert");
 $config->setSsl($sslConfig);
 $producer = new Producer($config);
 // ....  Your Business Code
-```
+```

doc/producer.md

@@ -47,6 +47,7 @@
 ```php
 use longlang\phpkafka\Producer\Producer;
 use longlang\phpkafka\Producer\ProducerConfig;
+use longlang\phpkafka\Protocol\RecordBatch\RecordHeader;
 
 $config = new ProducerConfig();
 $config->setBootstrapServer('127.0.0.1:9092');
@@ -57,6 +58,14 @@ $topic = 'test';
 $value = (string) microtime(true);
 $key = uniqid('', true);
 $producer->send('test', $value, $key);
+
+// 指定 headers
+// key-value或使用 RecordHeader 对象，都可以
+$headers = [
+    'key1' => 'value1',
+    (new RecordHeader())->setHeaderKey('key2')->setValue('value2'),
+];
+$producer->send('test', $value, $key, $headers);
 ```
 
 ## 批量发送消息
@@ -82,17 +91,18 @@ $producer->sendBatch([
 ]);
 ```
 
-
-
 ## SASL支持
+
 ### 相关配置
+
 |参数名|说明|默认值|
 | - | - | - |
 | type | SASL授权对应的类。PLAIN为``\longlang\phpkafka\Sasl\PlainSasl::class``| ''|
 | username | 账号 | '' |
 | password | 密码 | '' |
 
 **代码示例：**
+
 ```php
 use longlang\phpkafka\Producer\ProducerConfig;
 
@@ -107,11 +117,12 @@ $producer = new Producer($config);
 // ....  你的业务代码
 ```
 
-
 ## SSL支持
+
 类名：`longlang\phpkafka\Config\SslConfig`
 
 > 支持构造方法传入数组赋值
+
 ### 配置参数
 
 |参数名|说明|默认值|
@@ -146,4 +157,4 @@ $sslConfig->setCafile("/kafka-client/.github/kafka/cert/ca-cert");
 $config->setSsl($sslConfig);
 $producer = new Producer($config);
 // ....  你的业务代码
-```
+```

src/Consumer/ConsumeMessage.php

@@ -4,6 +4,8 @@
 
 namespace longlang\phpkafka\Consumer;
 
+use longlang\phpkafka\Protocol\RecordBatch\RecordHeader;
+
 class ConsumeMessage
 {
     /**
@@ -32,10 +34,13 @@ class ConsumeMessage
     protected $value;
 
     /**
-     * @var array
+     * @var RecordHeader[]
      */
     protected $headers;
 
+    /**
+     * @param RecordHeader[] $headers
+     */
     public function __construct(Consumer $consumer, string $topic, int $partition, ?string $key, ?string $value, array $headers)
     {
         $this->consumer = $consumer;
@@ -106,11 +111,17 @@ public function setValue(?string $value): self
         return $this;
     }
 
+    /**
+     * @return RecordHeader[]
+     */
     public function getHeaders(): array
     {
         return $this->headers;
     }
 
+    /**
+     * @param RecordHeader[] $headers
+     */
     public function setHeaders(array $headers): self
     {
         $this->headers = $headers;

src/Producer/ProduceMessage.php

@@ -4,6 +4,8 @@
 
 namespace longlang\phpkafka\Producer;
 
+use longlang\phpkafka\Protocol\RecordBatch\RecordHeader;
+
 class ProduceMessage
 {
     /**
@@ -22,7 +24,7 @@ class ProduceMessage
     protected $key;
 
     /**
-     * @var array
+     * @var RecordHeader[]|array
      */
     protected $headers;
 
@@ -31,6 +33,9 @@ class ProduceMessage
      */
     protected $partitionIndex;
 
+    /**
+     * @param RecordHeader[]|array $headers
+     */
     public function __construct(string $topic, ?string $value, ?string $key = null, array $headers = [], ?int $partitionIndex = null)
     {
         $this->topic = $topic;
@@ -55,6 +60,9 @@ public function getKey(): ?string
         return $this->key;
     }
 
+    /**
+     * @return RecordHeader[]|array
+     */
     public function getHeaders(): array
     {
         return $this->headers;

src/Producer/Producer.php

@@ -13,6 +13,7 @@
 use longlang\phpkafka\Protocol\Produce\TopicProduceData;
 use longlang\phpkafka\Protocol\RecordBatch\Record;
 use longlang\phpkafka\Protocol\RecordBatch\RecordBatch;
+use longlang\phpkafka\Protocol\RecordBatch\RecordHeader;
 
 class Producer
 {
@@ -44,6 +45,9 @@ public function __construct(ProducerConfig $config)
         $this->partitioner = new $class();
     }
 
+    /**
+     * @param RecordHeader[]|array $headers
+     */
     public function send(string $topic, ?string $value, ?string $key = null, array $headers = [], ?int $partitionIndex = null): void
     {
         $message = new ProduceMessage($topic, $value, $key, $headers, $partitionIndex);
@@ -113,7 +117,16 @@ public function sendBatch(array $messages): void
             $record = $records[] = new Record();
             $record->setKey($key);
             $record->setValue($value);
-            $record->setHeaders($message->getHeaders());
+            $headers = [];
+            foreach ($message->getHeaders() as $key => $value) {
+                if ($value instanceof RecordHeader) {
+                    $headers[] = $value;
+                // @phpstan-ignore-next-line
+                } else {
+                    $headers[] = (new RecordHeader())->setHeaderKey($key)->setValue($value);
+                }
+            }
+            $record->setHeaders($headers);
             $record->setOffsetDelta($offsetDelta);
             $record->setTimestampDelta(((int) (microtime(true) * 1000)) - $timestamp);
             $recordBatch->setRecords($records);

tests/ConsumerTest.php

@@ -70,4 +70,29 @@ public function testConsumeWithStickyAssignor(): void
         $consumer->start();
         $consumer->close();
     }
+
+    public function testConsumeWithHeader(): void
+    {
+        $config = new ConsumerConfig();
+        $config->setBroker(TestUtil::getHost() . ':' . TestUtil::getPort());
+        TestUtil::addConfigInfo($config);
+        $config->setTopic('test-header');
+        $config->setGroupId('testGroup');
+        $config->setClientId('testConsumeWithHeader');
+        $config->setGroupInstanceId('testConsumeWithHeader');
+        $config->setInterval(0.1);
+        $consumer = new Consumer($config, function (ConsumeMessage $message) {
+            $consumer = $message->getConsumer();
+            $this->assertNotEmpty($message->getValue());
+            $headers = $message->getHeaders();
+            $this->assertCount(2, $headers);
+            $this->assertEquals('key1', $headers[0]->getHeaderKey());
+            $this->assertEquals('value1', $headers[0]->getValue());
+            $this->assertEquals('key2', $headers[1]->getHeaderKey());
+            $this->assertEquals('value2', $headers[1]->getValue());
+            $consumer->stop();
+        });
+        $consumer->start();
+        $consumer->close();
+    }
 }

tests/ProducerTest.php

@@ -7,6 +7,7 @@
 use longlang\phpkafka\Producer\ProduceMessage;
 use longlang\phpkafka\Producer\Producer;
 use longlang\phpkafka\Producer\ProducerConfig;
+use longlang\phpkafka\Protocol\RecordBatch\RecordHeader;
 use PHPUnit\Framework\TestCase;
 
 class ProducerTest extends TestCase
@@ -48,4 +49,21 @@ public function testSendBatch(): void
         $producer->close();
         $this->assertTrue(true);
     }
+
+    public function testSendWithHeader(): void
+    {
+        $config = new ProducerConfig();
+        $config->setBootstrapServer(TestUtil::getHost() . ':' . TestUtil::getPort());
+        TestUtil::addConfigInfo($config);
+        $config->setAcks(-1);
+        $producer = new Producer($config);
+        $headers = [
+            'key1' => 'value1',
+            (new RecordHeader())->setHeaderKey('key2')->setValue('value2'),
+        ];
+        // @phpstan-ignore-next-line
+        $producer->send('test-header', (string) microtime(true), uniqid('', true), $headers);
+        $producer->close();
+        $this->assertTrue(true);
+    }
 }