Use producev to support headers in kafka

TiMESPLiNTER · TiMESPLiNTER · commit 879e6fb46054 · 2019-09-25T08:53:05.000+02:00
diff --git a/pkg/rdkafka/RdKafkaProducer.php b/pkg/rdkafka/RdKafkaProducer.php
@@ -42,7 +42,7 @@ public function send(Destination $destination, Message $message): void
         $key = $message->getKey() ?: $destination->getKey() ?: null;
 
         $topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());
-        $topic->produce($partition, 0 /* must be 0 */, $payload, $key);
+        $topic->producev($partition, 0 /* must be 0 */, $payload, $key, $message->getHeaders());
     }
 
     /**
diff --git a/pkg/rdkafka/composer.json b/pkg/rdkafka/composer.json
@@ -7,7 +7,7 @@
     "license": "MIT",
     "require": {
         "php": "^7.1.3",
-        "ext-rdkafka": "^3.0.3",
+        "ext-rdkafka": "^3.1.0",
         "queue-interop/queue-interop": "^0.7|^0.8"
     },
     "require-dev": {

Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,7 @@ public function send(Destination $destination, Message $message): void`
`42`	`42`	`$key = $message->getKey() ?: $destination->getKey() ?: null;`
`43`	`43`
`44`	`44`	`$topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());`
`45`		`- $topic->produce($partition, 0 /* must be 0 */, $payload, $key);`
	`45`	`+ $topic->producev($partition, 0 /* must be 0 */, $payload, $key, $message->getHeaders());`
`46`	`46`	`}`
`47`	`47`
`48`	`48`	`/**`