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

bin/changelog

@@ -8,6 +8,6 @@ then
     exit 1
 fi
 
-docker-compose run -e CHANGELOG_GITHUB_TOKEN=${CHANGELOG_GITHUB_TOKEN:-""} --workdir="/mqdev" --rm generate-changelog github_changelog_generator --future-release "$1" --no-issues --unreleased-only --output "CHANGELOG_FUTURE.md"
+docker compose run -e CHANGELOG_GITHUB_TOKEN=${CHANGELOG_GITHUB_TOKEN:-""} --workdir="/mqdev" --rm generate-changelog github_changelog_generator --future-release "$1" --no-issues --unreleased-only --output "CHANGELOG_FUTURE.md"
 
-#git add CHANGELOG.md && git commit -m "Release $1" -S && git push origin "$CURRENT_BRANCH"
+#git add CHANGELOG.md && git commit -m "Release $1" -S && git push origin "$CURRENT_BRANCH"

bin/dev

@@ -6,13 +6,13 @@ set -e
 while getopts "bustefdp" OPTION; do
   case $OPTION in
     b)
-      docker-compose pull -q && docker-compose build
+      docker compose pull -q && docker compose build
       ;;
     u)
-      docker-compose up
+      docker compose up
       ;;
     s)
-      docker-compose stop
+      docker compose stop
       ;;
     e)
       docker exec -it mqdev_dev_1 /bin/bash
@@ -21,7 +21,7 @@ while getopts "bustefdp" OPTION; do
       ./bin/php-cs-fixer fix
       ;;
 
-    d) docker-compose run --workdir="/mqdev" --rm dev php pkg/enqueue-bundle/Tests/Functional/app/console.php config:dump-reference enqueue -vvv
+    d) docker compose run --workdir="/mqdev" --rm dev php pkg/enqueue-bundle/Tests/Functional/app/console.php config:dump-reference enqueue -vvv
       ;;
     \?)
       echo "Invalid option: -$OPTARG" >&2

bin/test.sh

@@ -3,4 +3,4 @@
 set -x
 set -e
 
-docker-compose run --workdir="/mqdev" --rm dev ./docker/bin/test.sh $@
+docker compose run --workdir="/mqdev" --rm dev ./docker/bin/test.sh $@

composer.json

@@ -30,7 +30,7 @@
     "doctrine/persistence": "^2.0|^3.0",
     "mongodb/mongodb": "^1.2",
     "pda/pheanstalk": "^3.1",
-    "aws/aws-sdk-php": "^3.155",
+    "aws/aws-sdk-php": "^3.319",
     "stomp-php/stomp-php": "^4.5|^5",
     "php-http/guzzle7-adapter": "^0.1.1",
     "php-http/client-common": "^2.2.1",
@@ -137,4 +137,3 @@
     }
   }
 }
-

docker-compose.yml

@@ -38,16 +38,16 @@ services:
       - PREDIS_DSN=redis+predis://redis
       - PHPREDIS_DSN=redis+phpredis://redis
       - GPS_DSN=gps:?projectId=mqdev&emulatorHost=http://google-pubsub:8085
-      - SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4576&version=latest
-      - SNS_DSN=sns:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4575&version=latest
-      - SNSQS_DSN=snsqs:?key=key&secret=secret&region=us-east-1&sns_endpoint=http://localstack:4575&sqs_endpoint=http://localstack:4576&version=latest
+      - SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
+      - SNS_DSN=sns:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
+      - SNSQS_DSN=snsqs:?key=key&secret=secret&region=us-east-1&sns_endpoint=http://localstack:4566&sqs_endpoint=http://localstack:4566&version=latest
       - WAMP_DSN=wamp://thruway:9090
       - REDIS_HOST=redis
       - REDIS_PORT=6379
       - AWS_SQS_KEY=key
       - AWS_SQS_SECRET=secret
       - AWS_SQS_REGION=us-east-1
-      - AWS_SQS_ENDPOINT=http://localstack:4576
+      - AWS_SQS_ENDPOINT=http://localstack:4566
       - AWS_SQS_VERSION=latest
       - BEANSTALKD_DSN=beanstalk://beanstalkd:11300
       - GEARMAN_DSN=gearman://gearmand:4730
@@ -127,13 +127,13 @@ services:
       - '9090:9090'
 
   localstack:
-    image: 'localstack/localstack:0.8.10'
+    image: 'localstack/localstack'
     ports:
-      - '4576:4576'
-      - '4575:4575'
+      - "127.0.0.1:4566:4566"            # LocalStack Gateway
+      - "127.0.0.1:4510-4559:4510-4559"  # external services port range
     environment:
       HOSTNAME_EXTERNAL: 'localstack'
-      SERVICES: 'sqs,sns'
+      SERVICES: 's3,sqs,sns'
 
   influxdb:
     image: 'influxdb:latest'

docker/bin/test.sh

@@ -39,7 +39,7 @@ waitForService gearmand 4730 50
 waitForService kafka 9092 50
 waitForService mongo 27017 50
 waitForService thruway 9090 50
-waitForService localstack 4576 50
+waitForService localstack 4566 50
 
 php docker/bin/refresh-mysql-database.php || exit 1
 php docker/bin/refresh-postgres-database.php  || exit 1

docs/contribution.md

@@ -49,6 +49,6 @@ Once everything is done open a pull request on official repository.
 
 ## WTF?!
 
-* If you get `rabbitmqssl: forward host lookup failed: Unknown host, wait for service rabbitmqssl:5671` do `docker-compose down`.
+* If you get `rabbitmqssl: forward host lookup failed: Unknown host, wait for service rabbitmqssl:5671` do `docker compose down`.
 
 [back to index](index.md)

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.

phpunit.xml.dist

@@ -68,10 +68,6 @@
             <directory>pkg/sns/Tests</directory>
         </testsuite>
 
-        <testsuite name="snsqs transport">
-            <directory>pkg/snsqs/Tests</directory>
-        </testsuite>
-
         <testsuite name="pheanstalk transport">
             <directory>pkg/pheanstalk/Tests</directory>
         </testsuite>
@@ -123,6 +119,10 @@
         <testsuite name="monitoring">
             <directory>pkg/monitoring/Tests</directory>
         </testsuite>
+
+        <testsuite name="snsqs transport">
+            <directory>pkg/snsqs/Tests</directory>
+        </testsuite>
     </testsuites>
 
     <php>

pkg/amqp-ext/Tests/Functional/AmqpCommonUseCasesTest.php

@@ -112,6 +112,7 @@ public function testProduceAndReceiveOneMessageSentDirectlyToTemporaryQueue()
         $queue = $this->amqpContext->createTemporaryQueue();
 
         $message = $this->amqpContext->createMessage(__METHOD__);
+        $message->setDeliveryTag(145);
 
         $producer = $this->amqpContext->createProducer();
         $producer->send($queue, $message);
@@ -137,6 +138,7 @@ public function testProduceAndReceiveOneMessageSentDirectlyToTopic()
         $this->amqpContext->bind(new AmqpBind($topic, $queue));
 
         $message = $this->amqpContext->createMessage(__METHOD__);
+        $message->setDeliveryTag(145);
 
         $producer = $this->amqpContext->createProducer();
         $producer->send($topic, $message);
@@ -158,10 +160,11 @@ public function testConsumerReceiveMessageFromTopicDirectly()
         $this->amqpContext->declareTopic($topic);
 
         $consumer = $this->amqpContext->createConsumer($topic);
-        //guard
+        // guard
         $this->assertNull($consumer->receive(1000));
 
         $message = $this->amqpContext->createMessage(__METHOD__);
+        $message->setDeliveryTag(145);
 
         $producer = $this->amqpContext->createProducer();
         $producer->send($topic, $message);
@@ -181,10 +184,11 @@ public function testConsumerReceiveMessageWithZeroTimeout()
         $this->amqpContext->declareTopic($topic);
 
         $consumer = $this->amqpContext->createConsumer($topic);
-        //guard
+        // guard
         $this->assertNull($consumer->receive(1000));
 
         $message = $this->amqpContext->createMessage(__METHOD__);
+        $message->setDeliveryTag(145);
 
         $producer = $this->amqpContext->createProducer();
         $producer->send($topic, $message);
@@ -205,6 +209,7 @@ public function testPurgeMessagesFromQueue()
         $consumer = $this->amqpContext->createConsumer($queue);
 
         $message = $this->amqpContext->createMessage(__METHOD__);
+        $message->setDeliveryTag(145);
 
         $producer = $this->amqpContext->createProducer();
         $producer->send($queue, $message);

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

@@ -37,12 +37,17 @@ public function send(Destination $destination, Message $message): void
 
         /** @var Topic $topic */
         $topic = $this->context->getClient()->topic($destination->getTopicName());
-        $topic->publish([
-            'data' => json_encode($message),
-        ]);
+
+        $params = ['data' => json_encode($message)];
+
+        if (count($message->getAttributes()) > 0) {
+            $params['attributes'] = $message->getAttributes();
+        }
+
+        $topic->publish($params);
     }
 
-    public function setDeliveryDelay(int $deliveryDelay = null): Producer
+    public function setDeliveryDelay(?int $deliveryDelay = null): Producer
     {
         if (null === $deliveryDelay) {
             return $this;
@@ -56,7 +61,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 +75,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

@@ -33,7 +33,40 @@ public function testShouldSendMessage()
         $gtopic
             ->expects($this->once())
             ->method('publish')
-            ->with($this->identicalTo(['data' => '{"body":"","properties":[],"headers":[]}']))
+            ->with($this->identicalTo([
+                'data' => '{"body":"","properties":[],"headers":[]}',
+            ])
+            );
+
+        $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);
+    }
+
+    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();

pkg/sns/composer.json

@@ -9,7 +9,7 @@
         "php": "^7.4|^8.0",
         "queue-interop/queue-interop": "^0.8",
         "enqueue/dsn": "^0.10",
-        "aws/aws-sdk-php": "~3.155"
+        "aws/aws-sdk-php": "~3.319"
     },
     "require-dev": {
         "phpunit/phpunit": "^9.5",

pkg/snsqs/Tests/Spec/SnsQsSendToTopicAndReceiveNoWaitFromQueueTest.php

@@ -8,6 +8,7 @@
 
 /**
  * @group functional
+ *
  * @retry 5
  */
 class SnsQsSendToTopicAndReceiveNoWaitFromQueueTest extends SendToTopicAndReceiveNoWaitFromQueueSpec