Skip to content

Commit 97fe6fb

Browse files
author
DigitVE
committed
Merge branch 'master' into adding-new-getter-for-offset-in-rdkafka
2 parents bb5b633 + 182636b commit 97fe6fb

File tree

7 files changed

+76
-33
lines changed

7 files changed

+76
-33
lines changed

‎CHANGELOG.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,15 @@
11
# Change Log
22

3+
## [0.10.13](https://github.com/php-enqueue/enqueue-dev/tree/0.10.13) (2021-08-25)
4+
[Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.10.12...0.10.13)
5+
6+
**Merged pull requests:**
7+
8+
- \[SNSQS\] added possibility to send message attributes using snsqs transport [\#1195](https://github.com/php-enqueue/enqueue-dev/pull/1195) ([onatskyy](https://github.com/onatskyy))
9+
- Add in missing arg [\#1194](https://github.com/php-enqueue/enqueue-dev/pull/1194) ([gdsmith](https://github.com/gdsmith))
10+
- \#1190 add index on delivery\_id to prevent slow queries [\#1191](https://github.com/php-enqueue/enqueue-dev/pull/1191) ([commercewerft](https://github.com/commercewerft))
11+
- Add setTopicArn methods to SnsContext and SnsQsContext [\#1189](https://github.com/php-enqueue/enqueue-dev/pull/1189) ([gdsmith](https://github.com/gdsmith))
12+
313
## [0.10.11](https://github.com/php-enqueue/enqueue-dev/tree/0.10.11) (2021-04-28)
414
[Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.10.10...0.10.11)
515

‎pkg/dbal/DbalContext.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -238,6 +238,7 @@ public function createDataBaseTable(): void
238238

239239
$table->addIndex(['redeliver_after', 'delivery_id']);
240240
$table->addIndex(['time_to_live', 'delivery_id']);
241+
$table->addIndex(['delivery_id']);
241242

242243
$sm->createTable($table);
243244
}

‎pkg/sns/SnsContext.php‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,11 @@ public function declareTopic(SnsDestination $destination): void
7676
$this->topicArns[$destination->getTopicName()] = (string) $result->get('TopicArn');
7777
}
7878

79+
public function setTopicArn(SnsDestination $destination, string $arn): void
80+
{
81+
$this->topicArns[$destination->getTopicName()] = $arn;
82+
}
83+
7984
public function deleteTopic(SnsDestination $destination): void
8085
{
8186
$this->client->deleteTopic($this->getTopicArn($destination));

‎pkg/snsqs/SnsQsContext.php‎

Lines changed: 9 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -53,23 +53,15 @@ public function __construct($snsContext, $sqsContext)
5353
} elseif (is_callable($snsContext)) {
5454
$this->snsContextFactory = $snsContext;
5555
} else {
56-
throw new \InvalidArgumentException(sprintf(
57-
'The $snsContext argument must be either %s or callable that returns %s once called.',
58-
SnsContext::class,
59-
SnsContext::class
60-
));
56+
throw new \InvalidArgumentException(sprintf('The $snsContext argument must be either %s or callable that returns %s once called.', SnsContext::class, SnsContext::class));
6157
}
6258

6359
if ($sqsContext instanceof SqsContext) {
6460
$this->sqsContext = $sqsContext;
6561
} elseif (is_callable($sqsContext)) {
6662
$this->sqsContextFactory = $sqsContext;
6763
} else {
68-
throw new \InvalidArgumentException(sprintf(
69-
'The $sqsContext argument must be either %s or callable that returns %s once called.',
70-
SqsContext::class,
71-
SqsContext::class
72-
));
64+
throw new \InvalidArgumentException(sprintf('The $sqsContext argument must be either %s or callable that returns %s once called.', SqsContext::class, SqsContext::class));
7365
}
7466
}
7567

@@ -137,6 +129,11 @@ public function declareTopic(SnsQsTopic $topic): void
137129
$this->getSnsContext()->declareTopic($topic);
138130
}
139131

132+
public function setTopicArn(SnsQsTopic $topic, string $arn): void
133+
{
134+
$this->getSnsContext()->setTopicArn($topic, $arn);
135+
}
136+
140137
public function deleteTopic(SnsQsTopic $topic): void
141138
{
142139
$this->getSnsContext()->deleteTopic($topic);
@@ -181,11 +178,7 @@ private function getSnsContext(): SnsContext
181178
if (null === $this->snsContext) {
182179
$context = call_user_func($this->snsContextFactory);
183180
if (false == $context instanceof SnsContext) {
184-
throw new \LogicException(sprintf(
185-
'The factory must return instance of %s. It returned %s',
186-
SnsContext::class,
187-
is_object($context) ? get_class($context) : gettype($context)
188-
));
181+
throw new \LogicException(sprintf('The factory must return instance of %s. It returned %s', SnsContext::class, is_object($context) ? get_class($context) : gettype($context)));
189182
}
190183

191184
$this->snsContext = $context;
@@ -199,11 +192,7 @@ private function getSqsContext(): SqsContext
199192
if (null === $this->sqsContext) {
200193
$context = call_user_func($this->sqsContextFactory);
201194
if (false == $context instanceof SqsContext) {
202-
throw new \LogicException(sprintf(
203-
'The factory must return instance of %s. It returned %s',
204-
SqsContext::class,
205-
is_object($context) ? get_class($context) : gettype($context)
206-
));
195+
throw new \LogicException(sprintf('The factory must return instance of %s. It returned %s', SqsContext::class, is_object($context) ? get_class($context) : gettype($context)));
207196
}
208197

209198
$this->sqsContext = $context;

‎pkg/snsqs/SnsQsMessage.php‎

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,27 @@ class SnsQsMessage implements Message
1717
*/
1818
private $sqsMessage;
1919

20-
public function __construct(string $body = '', array $properties = [], array $headers = [])
21-
{
20+
/**
21+
* @var array|null
22+
*/
23+
private $messageAttributes;
24+
25+
/**
26+
* See AWS documentation for message attribute structure.
27+
*
28+
* @see https://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sns-2010-03-31.html#shape-messageattributevalue
29+
*/
30+
public function __construct(
31+
string $body = '',
32+
array $properties = [],
33+
array $headers = [],
34+
array $messageAttributes = null
35+
) {
2236
$this->body = $body;
2337
$this->properties = $properties;
2438
$this->headers = $headers;
2539
$this->redelivered = false;
40+
$this->messageAttributes = $messageAttributes;
2641
}
2742

2843
public function setSqsMessage(SqsMessage $message): void
@@ -34,4 +49,14 @@ public function getSqsMessage(): SqsMessage
3449
{
3550
return $this->sqsMessage;
3651
}
52+
53+
public function getMessageAttributes(): ?array
54+
{
55+
return $this->messageAttributes;
56+
}
57+
58+
public function setMessageAttributes(?array $messageAttributes): void
59+
{
60+
$this->messageAttributes = $messageAttributes;
61+
}
3762
}

‎pkg/snsqs/SnsQsProducer.php‎

Lines changed: 2 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -51,11 +51,7 @@ public function send(Destination $destination, Message $message): void
5151
InvalidMessageException::assertMessageInstanceOf($message, SnsQsMessage::class);
5252

5353
if (false == $destination instanceof SnsQsTopic && false == $destination instanceof SnsQsQueue) {
54-
throw new InvalidDestinationException(sprintf(
55-
'The destination must be an instance of [%s|%s] but got %s.',
56-
SnsQsTopic::class, SnsQsQueue::class,
57-
is_object($destination) ? get_class($destination) : gettype($destination)
58-
));
54+
throw new InvalidDestinationException(sprintf('The destination must be an instance of [%s|%s] but got %s.', SnsQsTopic::class, SnsQsQueue::class, is_object($destination) ? get_class($destination) : gettype($destination)));
5955
}
6056

6157
if ($destination instanceof SnsQsTopic) {
@@ -64,6 +60,7 @@ public function send(Destination $destination, Message $message): void
6460
$message->getProperties(),
6561
$message->getHeaders()
6662
);
63+
$snsMessage->setMessageAttributes($message->getMessageAttributes());
6764

6865
$this->getSnsProducer()->send($destination, $snsMessage);
6966
} else {
@@ -79,10 +76,6 @@ public function send(Destination $destination, Message $message): void
7976

8077
/**
8178
* Delivery delay is supported by SQSProducer.
82-
*
83-
* @param int|null $deliveryDelay
84-
*
85-
* @return Producer
8679
*/
8780
public function setDeliveryDelay(int $deliveryDelay = null): Producer
8881
{
@@ -93,8 +86,6 @@ public function setDeliveryDelay(int $deliveryDelay = null): Producer
9386

9487
/**
9588
* Delivery delay is supported by SQSProducer.
96-
*
97-
* @return int|null
9889
*/
9990
public function getDeliveryDelay(): ?int
10091
{

‎pkg/snsqs/Tests/SnsQsProducerTest.php‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace Enqueue\SnsQs\Tests;
44

55
use Enqueue\Sns\SnsContext;
6+
use Enqueue\Sns\SnsMessage;
67
use Enqueue\Sns\SnsProducer;
78
use Enqueue\SnsQs\SnsQsMessage;
89
use Enqueue\SnsQs\SnsQsProducer;
@@ -91,6 +92,7 @@ public function testShouldGetDeliveryDelayFromSQSProducer()
9192
public function testShouldSendSnsTopicMessageToSnsProducer()
9293
{
9394
$snsMock = $this->createSnsContextMock();
95+
$snsMock->method('createMessage')->willReturn(new SnsMessage());
9496
$destination = new SnsQsTopic('');
9597

9698
$snsProducerStub = $this->prophesize(SnsProducer::class);
@@ -102,6 +104,26 @@ public function testShouldSendSnsTopicMessageToSnsProducer()
102104
$producer->send($destination, new SnsQsMessage());
103105
}
104106

107+
public function testShouldSendSnsTopicMessageWithAttributesToSnsProducer()
108+
{
109+
$snsMock = $this->createSnsContextMock();
110+
$snsMock->method('createMessage')->willReturn(new SnsMessage());
111+
$destination = new SnsQsTopic('');
112+
113+
$snsProducerStub = $this->prophesize(SnsProducer::class);
114+
$snsProducerStub->send(
115+
$destination,
116+
Argument::that(function (SnsMessage $snsMessage) {
117+
return $snsMessage->getMessageAttributes() === ['foo' => 'bar'];
118+
})
119+
)->shouldBeCalledOnce();
120+
121+
$snsMock->method('createProducer')->willReturn($snsProducerStub->reveal());
122+
123+
$producer = new SnsQsProducer($snsMock, $this->createSqsContextMock());
124+
$producer->send($destination, new SnsQsMessage('', [], [], ['foo' => 'bar']));
125+
}
126+
105127
public function testShouldSendSqsMessageToSqsProducer()
106128
{
107129
$sqsMock = $this->createSqsContextMock();

0 commit comments

Comments
 (0)