Skip to content

Commit 6b48a41

Browse files
makasimmakasim
authored
Merge pull request #1349 from ubitransports/master-gps-attributes
feat(GPS): allow send attributes in Google PubSub message.
2 parents bb6e759 + cf6bbec commit 6b48a41

16 files changed

+117
-46
lines changed

‎bin/changelog

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,6 @@ then
88
exit 1
99
fi
1010

11-
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"
11+
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"
1212

13-
#git add CHANGELOG.md && git commit -m "Release $1" -S && git push origin "$CURRENT_BRANCH"
13+
git add CHANGELOG.md && git commit -m "Release $1" -S && git push origin "$CURRENT_BRANCH"

‎bin/dev

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,13 +6,13 @@ set -e
66
while getopts "bustefdp" OPTION; do
77
case $OPTION in
88
b)
9-
docker-compose pull -q && docker-compose build
9+
docker compose pull -q && docker compose build
1010
;;
1111
u)
12-
docker-compose up
12+
docker compose up
1313
;;
1414
s)
15-
docker-compose stop
15+
docker compose stop
1616
;;
1717
e)
1818
docker exec -it mqdev_dev_1 /bin/bash
@@ -21,7 +21,7 @@ while getopts "bustefdp" OPTION; do
2121
./bin/php-cs-fixer fix
2222
;;
2323

24-
d) docker-compose run --workdir="/mqdev" --rm dev php pkg/enqueue-bundle/Tests/Functional/app/console.php config:dump-reference enqueue -vvv
24+
d) docker compose run --workdir="/mqdev" --rm dev php pkg/enqueue-bundle/Tests/Functional/app/console.php config:dump-reference enqueue -vvv
2525
;;
2626
\?)
2727
echo "Invalid option: -$OPTARG" >&2

‎bin/test.sh

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,4 @@
33
set -x
44
set -e
55

6-
docker-compose run --workdir="/mqdev" --rm dev ./docker/bin/test.sh $@
6+
docker compose run --workdir="/mqdev" --rm dev ./docker/bin/test.sh $@

‎composer.json

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
"doctrine/persistence": "^2.0|^3.0",
3131
"mongodb/mongodb": "^1.2",
3232
"pda/pheanstalk": "^3.1",
33-
"aws/aws-sdk-php": "^3.155",
33+
"aws/aws-sdk-php": "^3.290",
3434
"stomp-php/stomp-php": "^4.5|^5",
3535
"php-http/guzzle7-adapter": "^0.1.1",
3636
"php-http/client-common": "^2.2.1",
@@ -137,4 +137,3 @@
137137
}
138138
}
139139
}
140-

‎docker-compose.yml

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -38,16 +38,16 @@ services:
3838
- PREDIS_DSN=redis+predis://redis
3939
- PHPREDIS_DSN=redis+phpredis://redis
4040
- GPS_DSN=gps:?projectId=mqdev&emulatorHost=http://google-pubsub:8085
41-
- SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4576&version=latest
42-
- SNS_DSN=sns:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4575&version=latest
43-
- SNSQS_DSN=snsqs:?key=key&secret=secret&region=us-east-1&sns_endpoint=http://localstack:4575&sqs_endpoint=http://localstack:4576&version=latest
41+
- SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
42+
- SNS_DSN=sns:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
43+
- SNSQS_DSN=snsqs:?key=key&secret=secret&region=us-east-1&sns_endpoint=http://localstack:4566&sqs_endpoint=http://localstack:4566&version=latest
4444
- WAMP_DSN=wamp://thruway:9090
4545
- REDIS_HOST=redis
4646
- REDIS_PORT=6379
4747
- AWS_SQS_KEY=key
4848
- AWS_SQS_SECRET=secret
4949
- AWS_SQS_REGION=us-east-1
50-
- AWS_SQS_ENDPOINT=http://localstack:4576
50+
- AWS_SQS_ENDPOINT=http://localstack:4566
5151
- AWS_SQS_VERSION=latest
5252
- BEANSTALKD_DSN=beanstalk://beanstalkd:11300
5353
- GEARMAN_DSN=gearman://gearmand:4730
@@ -83,6 +83,7 @@ services:
8383

8484
mysql:
8585
image: mysql:5.7
86+
platform: linux/amd64
8687
environment:
8788
MYSQL_ROOT_PASSWORD: rootpass
8889
MYSQL_DATABASE: mqdev
@@ -127,13 +128,13 @@ services:
127128
- '9090:9090'
128129

129130
localstack:
130-
image: 'localstack/localstack:0.8.10'
131+
image: 'localstack/localstack:3.6.0'
131132
ports:
132-
- '4576:4576'
133-
- '4575:4575'
133+
- "127.0.0.1:4566:4566" # LocalStack Gateway
134+
- "127.0.0.1:4510-4559:4510-4559" # external services port range
134135
environment:
135136
HOSTNAME_EXTERNAL: 'localstack'
136-
SERVICES: 'sqs,sns'
137+
SERVICES: 's3,sqs,sns'
137138

138139
influxdb:
139140
image: 'influxdb:latest'

‎docker/bin/test.sh

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ waitForService gearmand 4730 50
3939
waitForService kafka 9092 50
4040
waitForService mongo 27017 50
4141
waitForService thruway 9090 50
42-
waitForService localstack 4576 50
42+
waitForService localstack 4566 50
4343

4444
php docker/bin/refresh-mysql-database.php || exit 1
4545
php docker/bin/refresh-postgres-database.php || exit 1

‎docs/contribution.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,6 @@ Once everything is done open a pull request on official repository.
4949

5050
## WTF?!
5151

52-
* If you get `rabbitmqssl: forward host lookup failed: Unknown host, wait for service rabbitmqssl:5671` do `docker-compose down`.
52+
* If you get `rabbitmqssl: forward host lookup failed: Unknown host, wait for service rabbitmqssl:5671` do `docker compose down`.
5353

5454
[back to index](index.md)

‎phpunit.xml.dist

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -68,10 +68,6 @@
6868
<directory>pkg/sns/Tests</directory>
6969
</testsuite>
7070

71-
<testsuite name="snsqs transport">
72-
<directory>pkg/snsqs/Tests</directory>
73-
</testsuite>
74-
7571
<testsuite name="pheanstalk transport">
7672
<directory>pkg/pheanstalk/Tests</directory>
7773
</testsuite>
@@ -123,6 +119,10 @@
123119
<testsuite name="monitoring">
124120
<directory>pkg/monitoring/Tests</directory>
125121
</testsuite>
122+
123+
<testsuite name="snsqs transport">
124+
<directory>pkg/snsqs/Tests</directory>
125+
</testsuite>
126126
</testsuites>
127127

128128
<php>

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

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,7 @@ public function testProduceAndReceiveOneMessageSentDirectlyToTemporaryQueue()
112112
$queue = $this->amqpContext->createTemporaryQueue();
113113

114114
$message = $this->amqpContext->createMessage(__METHOD__);
115+
$message->setDeliveryTag(145);
115116

116117
$producer = $this->amqpContext->createProducer();
117118
$producer->send($queue, $message);
@@ -137,6 +138,7 @@ public function testProduceAndReceiveOneMessageSentDirectlyToTopic()
137138
$this->amqpContext->bind(new AmqpBind($topic, $queue));
138139

139140
$message = $this->amqpContext->createMessage(__METHOD__);
141+
$message->setDeliveryTag(145);
140142

141143
$producer = $this->amqpContext->createProducer();
142144
$producer->send($topic, $message);
@@ -158,10 +160,11 @@ public function testConsumerReceiveMessageFromTopicDirectly()
158160
$this->amqpContext->declareTopic($topic);
159161

160162
$consumer = $this->amqpContext->createConsumer($topic);
161-
//guard
163+
// guard
162164
$this->assertNull($consumer->receive(1000));
163165

164166
$message = $this->amqpContext->createMessage(__METHOD__);
167+
$message->setDeliveryTag(145);
165168

166169
$producer = $this->amqpContext->createProducer();
167170
$producer->send($topic, $message);
@@ -181,10 +184,11 @@ public function testConsumerReceiveMessageWithZeroTimeout()
181184
$this->amqpContext->declareTopic($topic);
182185

183186
$consumer = $this->amqpContext->createConsumer($topic);
184-
//guard
187+
// guard
185188
$this->assertNull($consumer->receive(1000));
186189

187190
$message = $this->amqpContext->createMessage(__METHOD__);
191+
$message->setDeliveryTag(145);
188192

189193
$producer = $this->amqpContext->createProducer();
190194
$producer->send($topic, $message);
@@ -205,6 +209,7 @@ public function testPurgeMessagesFromQueue()
205209
$consumer = $this->amqpContext->createConsumer($queue);
206210

207211
$message = $this->amqpContext->createMessage(__METHOD__);
212+
$message->setDeliveryTag(145);
208213

209214
$producer = $this->amqpContext->createProducer();
210215
$producer->send($queue, $message);

‎pkg/gps/GpsMessage.php

Lines changed: 24 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,11 @@ class GpsMessage implements Message, \JsonSerializable
2424
*/
2525
private $headers;
2626

27+
/**
28+
* @var array
29+
*/
30+
private $attributes;
31+
2732
/**
2833
* @var bool
2934
*/
@@ -34,11 +39,12 @@ class GpsMessage implements Message, \JsonSerializable
3439
*/
3540
private $nativeMessage;
3641

37-
public function __construct(string $body = '', array $properties = [], array $headers = [])
42+
public function __construct(string $body = '', array $properties = [], array $headers = [], array $attributes = [])
3843
{
3944
$this->body = $body;
4045
$this->properties = $properties;
4146
$this->headers = $headers;
47+
$this->attributes = $attributes;
4248

4349
$this->redelivered = false;
4450
}
@@ -103,7 +109,7 @@ public function isRedelivered(): bool
103109
return $this->redelivered;
104110
}
105111

106-
public function setCorrelationId(string $correlationId = null): void
112+
public function setCorrelationId(?string $correlationId = null): void
107113
{
108114
$this->setHeader('correlation_id', $correlationId);
109115
}
@@ -113,7 +119,7 @@ public function getCorrelationId(): ?string
113119
return $this->getHeader('correlation_id');
114120
}
115121

116-
public function setMessageId(string $messageId = null): void
122+
public function setMessageId(?string $messageId = null): void
117123
{
118124
$this->setHeader('message_id', $messageId);
119125
}
@@ -130,12 +136,12 @@ public function getTimestamp(): ?int
130136
return null === $value ? null : (int) $value;
131137
}
132138

133-
public function setTimestamp(int $timestamp = null): void
139+
public function setTimestamp(?int $timestamp = null): void
134140
{
135141
$this->setHeader('timestamp', $timestamp);
136142
}
137143

138-
public function setReplyTo(string $replyTo = null): void
144+
public function setReplyTo(?string $replyTo = null): void
139145
{
140146
$this->setHeader('reply_to', $replyTo);
141147
}
@@ -151,6 +157,7 @@ public function jsonSerialize(): array
151157
'body' => $this->getBody(),
152158
'properties' => $this->getProperties(),
153159
'headers' => $this->getHeaders(),
160+
'attributes' => $this->getAttributes(),
154161
];
155162
}
156163

@@ -161,16 +168,26 @@ public static function jsonUnserialize(string $json): self
161168
throw new \InvalidArgumentException(sprintf('The malformed json given. Error %s and message %s', json_last_error(), json_last_error_msg()));
162169
}
163170

164-
return new self($data['body'] ?? $json, $data['properties'] ?? [], $data['headers'] ?? []);
171+
return new self($data['body'] ?? $json, $data['properties'] ?? [], $data['headers'] ?? [], $data['attributes'] ?? []);
165172
}
166173

167174
public function getNativeMessage(): ?GoogleMessage
168175
{
169176
return $this->nativeMessage;
170177
}
171178

172-
public function setNativeMessage(GoogleMessage $message = null): void
179+
public function setNativeMessage(?GoogleMessage $message = null): void
173180
{
174181
$this->nativeMessage = $message;
175182
}
183+
184+
public function setAttributes(array $attributes): void
185+
{
186+
$this->attributes = $attributes;
187+
}
188+
189+
public function getAttributes(): array
190+
{
191+
return $this->attributes;
192+
}
176193
}

‎pkg/gps/GpsProducer.php

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -37,12 +37,17 @@ public function send(Destination $destination, Message $message): void
3737

3838
/** @var Topic $topic */
3939
$topic = $this->context->getClient()->topic($destination->getTopicName());
40-
$topic->publish([
41-
'data' => json_encode($message),
42-
]);
40+
41+
$params = ['data' => json_encode($message)];
42+
43+
if (count($message->getAttributes()) > 0) {
44+
$params['attributes'] = $message->getAttributes();
45+
}
46+
47+
$topic->publish($params);
4348
}
4449

45-
public function setDeliveryDelay(int $deliveryDelay = null): Producer
50+
public function setDeliveryDelay(?int $deliveryDelay = null): Producer
4651
{
4752
if (null === $deliveryDelay) {
4853
return $this;
@@ -56,7 +61,7 @@ public function getDeliveryDelay(): ?int
5661
return null;
5762
}
5863

59-
public function setPriority(int $priority = null): Producer
64+
public function setPriority(?int $priority = null): Producer
6065
{
6166
if (null === $priority) {
6267
return $this;
@@ -70,7 +75,7 @@ public function getPriority(): ?int
7075
return null;
7176
}
7277

73-
public function setTimeToLive(int $timeToLive = null): Producer
78+
public function setTimeToLive(?int $timeToLive = null): Producer
7479
{
7580
if (null === $timeToLive) {
7681
return $this;

‎pkg/gps/Tests/GpsMessageTest.php

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,18 +18,18 @@ public function testCouldSetGetNativeMessage()
1818

1919
public function testColdBeSerializedToJson()
2020
{
21-
$message = new GpsMessage('theBody', ['thePropFoo' => 'thePropFooVal'], ['theHeaderFoo' => 'theHeaderFooVal']);
21+
$message = new GpsMessage('theBody', ['thePropFoo' => 'thePropFooVal'], ['theHeaderFoo' => 'theHeaderFooVal'], ['theAttributeFoo' => 'theAttributeFooVal']);
2222

23-
$this->assertEquals('{"body":"theBody","properties":{"thePropFoo":"thePropFooVal"},"headers":{"theHeaderFoo":"theHeaderFooVal"}}', json_encode($message));
23+
$this->assertEquals('{"body":"theBody","properties":{"thePropFoo":"thePropFooVal"},"headers":{"theHeaderFoo":"theHeaderFooVal"},"attributes":{"theAttributeFoo":"theAttributeFooVal"}}', json_encode($message));
2424
}
2525

2626
public function testCouldBeUnserializedFromJson()
2727
{
28-
$message = new GpsMessage('theBody', ['thePropFoo' => 'thePropFooVal'], ['theHeaderFoo' => 'theHeaderFooVal']);
28+
$message = new GpsMessage('theBody', ['thePropFoo' => 'thePropFooVal'], ['theHeaderFoo' => 'theHeaderFooVal'], ['theAttributeFoo' => 'theAttributeFooVal']);
2929

3030
$json = json_encode($message);
3131

32-
//guard
32+
// guard
3333
$this->assertNotEmpty($json);
3434

3535
$unserializedMessage = GpsMessage::jsonUnserialize($json);
@@ -40,7 +40,7 @@ public function testCouldBeUnserializedFromJson()
4040

4141
public function testMessageEntityCouldBeUnserializedFromJson()
4242
{
43-
$json = '{"body":"theBody","properties":{"thePropFoo":"thePropFooVal"},"headers":{"theHeaderFoo":"theHeaderFooVal"}}';
43+
$json = '{"body":"theBody","properties":{"thePropFoo":"thePropFooVal"},"headers":{"theHeaderFoo":"theHeaderFooVal"},"attributes":{"theAttributeFoo":"theAttributeFooVal"}}';
4444

4545
$unserializedMessage = GpsMessage::jsonUnserialize($json);
4646

@@ -49,6 +49,7 @@ public function testMessageEntityCouldBeUnserializedFromJson()
4949
$this->assertEquals($decoded['body'], $unserializedMessage->getBody());
5050
$this->assertEquals($decoded['properties'], $unserializedMessage->getProperties());
5151
$this->assertEquals($decoded['headers'], $unserializedMessage->getHeaders());
52+
$this->assertEquals($decoded['attributes'], $unserializedMessage->getAttributes());
5253
}
5354

5455
public function testMessagePayloadCouldBeUnserializedFromJson()
@@ -61,6 +62,7 @@ public function testMessagePayloadCouldBeUnserializedFromJson()
6162
$this->assertEquals($json, $unserializedMessage->getBody());
6263
$this->assertEquals([], $unserializedMessage->getProperties());
6364
$this->assertEquals([], $unserializedMessage->getHeaders());
65+
$this->assertEquals([], $unserializedMessage->getAttributes());
6466
}
6567

6668
public function testThrowIfMalformedJsonGivenOnUnsterilizedFromJson()
@@ -70,4 +72,13 @@ public function testThrowIfMalformedJsonGivenOnUnsterilizedFromJson()
7072

7173
GpsMessage::jsonUnserialize('{]');
7274
}
75+
76+
public function testGetAttributes()
77+
{
78+
$message = new GpsMessage('the body', [], [], ['key1' => 'value1']);
79+
80+
$attributes = $message->getAttributes();
81+
82+
$this->assertSame(['key1' => 'value1'], $attributes);
83+
}
7384
}

0 commit comments

Comments
 (0)