feat(GPS): allow send attributes in Google PubSub message.

Author: p-pichet

docs/transport/gps.md

@@ -62,6 +62,21 @@ $context->declareTopic($fooTopic);
 $context->createProducer()->send($fooTopic, $message);
 ```
 
+You can send attributes using headers :
+
+```php
+<?php
+/** @var \Enqueue\Gps\GpsContext $context */
+
+$fooTopic = $context->createTopic('foo');
+$attributes = ['key1' => 'value1'];
+$message = $context->createMessage('Hello world!', [], ['attributes' => $attributes]);
+
+$context->declareTopic($fooTopic);
+
+$context->createProducer()->send($fooTopic, $message);
+```
+
 ## Consume message:
 
 Before you can consume message you have to subscribe a queue to the topic.

pkg/gps/GpsMessage.php

@@ -34,10 +34,17 @@ class GpsMessage implements Message, \JsonSerializable
      */
     private $nativeMessage;
 
+    /**
+     * @var array
+     */
+    private $attributes;
+
     public function __construct(string $body = '', array $properties = [], array $headers = [])
     {
         $this->body = $body;
         $this->properties = $properties;
+        $this->attributes = $headers['attributes'] ?? [];
+        unset($headers['attributes']);
         $this->headers = $headers;
 
         $this->redelivered = false;
@@ -103,7 +110,7 @@ public function isRedelivered(): bool
         return $this->redelivered;
     }
 
-    public function setCorrelationId(string $correlationId = null): void
+    public function setCorrelationId(?string $correlationId = null): void
     {
         $this->setHeader('correlation_id', $correlationId);
     }
@@ -113,7 +120,7 @@ public function getCorrelationId(): ?string
         return $this->getHeader('correlation_id');
     }
 
-    public function setMessageId(string $messageId = null): void
+    public function setMessageId(?string $messageId = null): void
     {
         $this->setHeader('message_id', $messageId);
     }
@@ -130,12 +137,12 @@ public function getTimestamp(): ?int
         return null === $value ? null : (int) $value;
     }
 
-    public function setTimestamp(int $timestamp = null): void
+    public function setTimestamp(?int $timestamp = null): void
     {
         $this->setHeader('timestamp', $timestamp);
     }
 
-    public function setReplyTo(string $replyTo = null): void
+    public function setReplyTo(?string $replyTo = null): void
     {
         $this->setHeader('reply_to', $replyTo);
     }
@@ -169,8 +176,13 @@ public function getNativeMessage(): ?GoogleMessage
         return $this->nativeMessage;
     }
 
-    public function setNativeMessage(GoogleMessage $message = null): void
+    public function setNativeMessage(?GoogleMessage $message = null): void
     {
         $this->nativeMessage = $message;
     }
+
+    public function getAttributes(): array
+    {
+        return $this->attributes;
+    }
 }

pkg/gps/GpsProducer.php

@@ -39,10 +39,11 @@ public function send(Destination $destination, Message $message): void
         $topic = $this->context->getClient()->topic($destination->getTopicName());
         $topic->publish([
             'data' => json_encode($message),
+            'attributes' => $message->getAttributes(),
         ]);
     }
 
-    public function setDeliveryDelay(int $deliveryDelay = null): Producer
+    public function setDeliveryDelay(?int $deliveryDelay = null): Producer
     {
         if (null === $deliveryDelay) {
             return $this;
@@ -56,7 +57,7 @@ public function getDeliveryDelay(): ?int
         return null;
     }
 
-    public function setPriority(int $priority = null): Producer
+    public function setPriority(?int $priority = null): Producer
     {
         if (null === $priority) {
             return $this;
@@ -70,7 +71,7 @@ public function getPriority(): ?int
         return null;
     }
 
-    public function setTimeToLive(int $timeToLive = null): Producer
+    public function setTimeToLive(?int $timeToLive = null): Producer
     {
         if (null === $timeToLive) {
             return $this;

pkg/gps/Tests/GpsMessageTest.php

@@ -29,7 +29,7 @@ public function testCouldBeUnserializedFromJson()
 
         $json = json_encode($message);
 
-        //guard
+        // guard
         $this->assertNotEmpty($json);
 
         $unserializedMessage = GpsMessage::jsonUnserialize($json);
@@ -70,4 +70,13 @@ public function testThrowIfMalformedJsonGivenOnUnsterilizedFromJson()
 
         GpsMessage::jsonUnserialize('{]');
     }
+
+    public function testGetAttributes()
+    {
+        $message = new GpsMessage('the body', [], ['attributes' => ['key1' => 'value1']]);
+
+        $attributes = $message->getAttributes();
+
+        $this->assertSame(['key1' => 'value1'], $attributes);
+    }
 }

pkg/gps/Tests/GpsProducerTest.php

@@ -55,6 +55,37 @@ public function testShouldSendMessage()
         $producer->send($topic, $message);
     }
 
+    public function testShouldSendMessageWithAttributes()
+    {
+        $topic = new GpsTopic('topic-name');
+        $message = new GpsMessage('', [], ['attributes' => ['key1' => 'value1']]);
+
+        $gtopic = $this->createGTopicMock();
+        $gtopic
+            ->expects($this->once())
+            ->method('publish')
+            ->with($this->identicalTo(['data' => '{"body":"","properties":[],"headers":[]}', 'attributes' => ['key1' => 'value1']]))
+        ;
+
+        $client = $this->createPubSubClientMock();
+        $client
+            ->expects($this->once())
+            ->method('topic')
+            ->with('topic-name')
+            ->willReturn($gtopic)
+        ;
+
+        $context = $this->createContextMock();
+        $context
+            ->expects($this->once())
+            ->method('getClient')
+            ->willReturn($client)
+        ;
+
+        $producer = new GpsProducer($context);
+        $producer->send($topic, $message);
+    }
+
     /**
      * @return GpsContext|\PHPUnit\Framework\MockObject\MockObject|GpsContext
      */