Custom header option for consumers

phpunit.xml

@@ -27,4 +27,4 @@
             </exclude>
         </whitelist>
     </filter>
-</phpunit>
+</phpunit>

src/Messenger/KafkaReceiver.php

@@ -61,7 +61,7 @@ public function get(): iterable
 
                 $envelope = $this->serializer->decode([
                     'body' => $message->payload,
-                    'headers' => $message->headers,
+                    'headers' => array_merge($this->properties->getHeaders(), $message->headers ?? []),
                     'key' => $message->key,
                     'offset' => $message->offset,
                     'timestamp' => $message->timestamp,

src/Messenger/KafkaReceiverProperties.php

@@ -20,16 +20,21 @@ final class KafkaReceiverProperties
     /** @var bool */
     private $commitAsync;
 
+    /** @var array */
+    private $headers;
+
     public function __construct(
         KafkaConf $kafkaConf,
         string $topicName,
         int $receiveTimeoutMs,
-        bool $commitAsync
+        bool $commitAsync,
+        array $headers
     ) {
         $this->kafkaConf = $kafkaConf;
         $this->topicName = $topicName;
         $this->receiveTimeoutMs = $receiveTimeoutMs;
         $this->commitAsync = $commitAsync;
+        $this->headers = $headers;
     }
 
     public function getKafkaConf(): KafkaConf
@@ -51,4 +56,9 @@ public function isCommitAsync(): bool
     {
         return $this->commitAsync;
     }
+
+    public function getHeaders(): array
+    {
+        return $this->headers;
+    }
 }

src/Messenger/KafkaTransportFactory.php

@@ -77,7 +77,8 @@ public function createTransport(string $dsn, array $options, SerializerInterface
                 $conf,
                 $options['topic']['name'],
                 $options['receiveTimeout'] ?? 10000,
-                $options['commitAsync'] ?? false
+                $options['commitAsync'] ?? false,
+                $options['headers'] ?? []
             )
         );
     }

tests/Unit/Messenger/KafkaTransportTest.php

@@ -73,7 +73,8 @@ public function testConstruct()
                 new KafkaConf(),
                 'test',
                 10000,
-                false
+                false,
+                []
             )
         );
 
@@ -110,6 +111,9 @@ public function testGet()
                     'type' => TestMessage::class,
                     'Content-Type' => 'application/json',
                 ],
+                'key' => null,
+                'offset' => 0,
+                'timestamp' => 1586861356,
             ])
             ->willReturn(new Envelope(new TestMessage()));
 
@@ -127,7 +131,8 @@ public function testGet()
                 new KafkaConf(),
                 'test',
                 10000,
-                false
+                false,
+                []
             )
         );