Skip to content

Commit 4dbec68

Browse files
makasimmakasim
committed
[sqs] Multi region support.
1 parent caadd76 commit 4dbec68

10 files changed

Lines changed: 729 additions & 136 deletions

‎pkg/sqs/SqsClient.php‎

Lines changed: 24 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,12 @@ class SqsClient
2525
*/
2626
private $inputClient;
2727

28+
/**
29+
* @param AwsSqsClient|callable $inputClient
30+
*/
2831
public function __construct($inputClient)
2932
{
30-
$this->inputClient = is_callable($inputClient) ? $inputClient : function () use ($inputClient) {
31-
return $inputClient;
32-
};
33+
$this->inputClient = $inputClient;
3334
}
3435

3536
public function deleteMessage(array $args): Result
@@ -89,7 +90,13 @@ public function getAWSClient(): AwsSqsClient
8990

9091
private function callApi(string $name, array $args): Result
9192
{
93+
$this->resolveClient();
94+
9295
if ($this->singleClient) {
96+
if (false == empty($args['@region'])) {
97+
throw new \LogicException('Cannot send message to another region because transport is configured with single aws client');
98+
}
99+
93100
unset($args['@region']);
94101

95102
return call_user_func([$this->singleClient, $name], $args);
@@ -108,16 +115,27 @@ private function resolveClient(): void
108115
return;
109116
}
110117

111-
$client = call_user_func($this->inputClient);
118+
$client = $this->inputClient;
112119
if ($client instanceof MultiRegionClient) {
113120
$this->multiClient = $client;
114121

115122
return;
116-
}
117-
if ($client instanceof AwsSqsClient) {
123+
} elseif ($client instanceof AwsSqsClient) {
118124
$this->singleClient = $client;
119125

120126
return;
127+
} elseif (is_callable($client)) {
128+
$client = call_user_func($client);
129+
if ($client instanceof MultiRegionClient) {
130+
$this->multiClient = $client;
131+
132+
return;
133+
}
134+
if ($client instanceof AwsSqsClient) {
135+
$this->singleClient = $client;
136+
137+
return;
138+
}
121139
}
122140

123141
throw new \LogicException(sprintf(

‎pkg/sqs/SqsConnectionFactory.php‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,8 @@
44

55
namespace Enqueue\Sqs;
66

7-
use Aws\MultiRegionClient;
87
use Aws\Sdk;
9-
use Aws\Sqs\SqsClient;
8+
use Aws\Sqs\SqsClient as AwsSqsClient;
109
use Enqueue\Dsn\Dsn;
1110
use Interop\Queue\ConnectionFactory;
1211
use Interop\Queue\Context;
@@ -19,7 +18,7 @@ class SqsConnectionFactory implements ConnectionFactory
1918
private $config;
2019

2120
/**
22-
* @var MultiRegionClient
21+
* @var SqsClient
2322
*/
2423
private $client;
2524

@@ -45,8 +44,8 @@ class SqsConnectionFactory implements ConnectionFactory
4544
*/
4645
public function __construct($config = 'sqs:')
4746
{
48-
if ($config instanceof SqsClient) {
49-
$this->client = $config;
47+
if ($config instanceof AwsSqsClient) {
48+
$this->client = new SqsClient($config);
5049
$this->config = ['lazy' => false] + $this->defaultConfig();
5150

5251
return;
@@ -63,7 +62,7 @@ public function __construct($config = 'sqs:')
6362
unset($config['dsn']);
6463
}
6564
} else {
66-
throw new \LogicException(sprintf('The config must be either an array of options, a DSN string, null or instance of %s', SqsClient::class));
65+
throw new \LogicException(sprintf('The config must be either an array of options, a DSN string, null or instance of %s', AwsSqsClient::class));
6766
}
6867

6968
$this->config = array_replace($this->defaultConfig(), $config);
@@ -74,16 +73,10 @@ public function __construct($config = 'sqs:')
7473
*/
7574
public function createContext(): Context
7675
{
77-
if ($this->config['lazy']) {
78-
return new SqsContext(function () {
79-
return $this->establishConnection();
80-
}, $this->config);
81-
}
82-
8376
return new SqsContext($this->establishConnection(), $this->config);
8477
}
8578

86-
private function establishConnection(): MultiRegionClient
79+
private function establishConnection(): SqsClient
8780
{
8881
if ($this->client) {
8982
return $this->client;
@@ -110,7 +103,14 @@ private function establishConnection(): MultiRegionClient
110103
}
111104
}
112105

113-
$this->client = (new Sdk(['Sqs' => $config]))->createMultiRegionSqs();
106+
$establishConnection = function () use ($config) {
107+
return (new Sdk(['Sqs' => $config]))->createMultiRegionSqs();
108+
};
109+
110+
$this->client = $this->config['lazy'] ?
111+
new SqsClient($establishConnection) :
112+
new SqsClient($establishConnection())
113+
;
114114

115115
return $this->client;
116116
}

‎pkg/sqs/SqsConsumer.php‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -119,7 +119,7 @@ public function acknowledge(Message $message): void
119119
{
120120
InvalidMessageException::assertMessageInstanceOf($message, SqsMessage::class);
121121

122-
$this->context->getClient()->deleteMessage([
122+
$this->context->getSqsClient()->deleteMessage([
123123
'@region' => $this->queue->getRegion(),
124124
'QueueUrl' => $this->context->getQueueUrl($this->queue),
125125
'ReceiptHandle' => $message->getReceiptHandle(),
@@ -133,7 +133,7 @@ public function reject(Message $message, bool $requeue = false): void
133133
{
134134
InvalidMessageException::assertMessageInstanceOf($message, SqsMessage::class);
135135

136-
$this->context->getClient()->deleteMessage([
136+
$this->context->getSqsClient()->deleteMessage([
137137
'@region' => $this->queue->getRegion(),
138138
'QueueUrl' => $this->context->getQueueUrl($this->queue),
139139
'ReceiptHandle' => $message->getReceiptHandle(),
@@ -163,7 +163,7 @@ protected function receiveMessage(int $timeoutSeconds): ?SqsMessage
163163
$arguments['VisibilityTimeout'] = $this->visibilityTimeout;
164164
}
165165

166-
$result = $this->context->getClient()->receiveMessage($arguments);
166+
$result = $this->context->getSqsClient()->receiveMessage($arguments);
167167

168168
if ($result->hasKey('Messages')) {
169169
$this->messages = $result->get('Messages');

‎pkg/sqs/SqsContext.php‎

Lines changed: 26 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
namespace Enqueue\Sqs;
66

7-
use Aws\Sqs\SqsClient;
7+
use Aws\Sqs\SqsClient as AwsSqsClient;
88
use Interop\Queue\Consumer;
99
use Interop\Queue\Context;
1010
use Interop\Queue\Destination;
@@ -24,37 +24,19 @@ class SqsContext implements Context
2424
*/
2525
private $client;
2626

27-
/**
28-
* @var callable
29-
*/
30-
private $clientFactory;
31-
3227
/**
3328
* @var array
3429
*/
3530
private $queueUrls;
3631

37-
private $config;
38-
3932
/**
40-
* Callable must return instance of SqsClient once called.
41-
*
42-
* @param SqsClient|callable $client
33+
* @var array
4334
*/
44-
public function __construct($client, array $config)
45-
{
46-
if ($client instanceof SqsClient) {
47-
$this->client = $client;
48-
} elseif (is_callable($client)) {
49-
$this->clientFactory = $client;
50-
} else {
51-
throw new \InvalidArgumentException(sprintf(
52-
'The $client argument must be either %s or callable that returns %s once called.',
53-
SqsClient::class,
54-
SqsClient::class
55-
));
56-
}
35+
private $config;
5736

37+
public function __construct(SqsClient $client, array $config)
38+
{
39+
$this->client = $client;
5840
$this->config = $config;
5941
}
6042

@@ -118,7 +100,7 @@ public function purgeQueue(Queue $queue): void
118100
{
119101
InvalidDestinationException::assertDestinationInstanceOf($queue, SqsDestination::class);
120102

121-
$this->getClient()->purgeQueue([
103+
$this->client->purgeQueue([
122104
'@region' => $queue->getRegion(),
123105
'QueueUrl' => $this->getQueueUrl($queue),
124106
]);
@@ -129,24 +111,26 @@ public function createSubscriptionConsumer(): SubscriptionConsumer
129111
throw SubscriptionConsumerNotSupportedException::providerDoestNotSupportIt();
130112
}
131113

132-
public function getClient(): SqsClient
114+
public function getAwsSqsClient(): AwsSqsClient
133115
{
134-
if (false == $this->client) {
135-
$client = call_user_func($this->clientFactory);
136-
if (false == $client instanceof SqsClient) {
137-
throw new \LogicException(sprintf(
138-
'The factory must return instance of "%s". But it returns %s',
139-
SqsClient::class,
140-
is_object($client) ? get_class($client) : gettype($client)
141-
));
142-
}
143-
144-
$this->client = $client;
145-
}
116+
return $this->client->getAWSClient();
117+
}
146118

119+
public function getSqsClient(): SqsClient
120+
{
147121
return $this->client;
148122
}
149123

124+
/**
125+
* @deprecated use getAwsSqsClient method
126+
*/
127+
public function getClient(): AwsSqsClient
128+
{
129+
@trigger_error('The method is deprecated since 0.9.2. SqsContext::getAwsSqsClient() method should be used.', E_USER_DEPRECATED);
130+
131+
return $this->getAwsSqsClient();
132+
}
133+
150134
public function getQueueUrl(SqsDestination $destination): string
151135
{
152136
if (isset($this->queueUrls[$destination->getQueueName()])) {
@@ -155,7 +139,7 @@ public function getQueueUrl(SqsDestination $destination): string
155139

156140
$arguments = [
157141
'@region' => $destination->getRegion(),
158-
'QueueName' => $destination->getQueueName()
142+
'QueueName' => $destination->getQueueName(),
159143
];
160144

161145
if ($destination->getQueueOwnerAWSAccountId()) {
@@ -164,7 +148,7 @@ public function getQueueUrl(SqsDestination $destination): string
164148
$arguments['QueueOwnerAWSAccountId'] = $this->config['queue_owner_aws_account_id'];
165149
}
166150

167-
$result = $this->getClient()->getQueueUrl($arguments);
151+
$result = $this->client->getQueueUrl($arguments);
168152

169153
if (false == $result->hasKey('QueueUrl')) {
170154
throw new \RuntimeException(sprintf('QueueUrl cannot be resolved. queueName: "%s"', $destination->getQueueName()));
@@ -175,7 +159,7 @@ public function getQueueUrl(SqsDestination $destination): string
175159

176160
public function declareQueue(SqsDestination $dest): void
177161
{
178-
$result = $this->getClient()->createQueue([
162+
$result = $this->client->createQueue([
179163
'@region' => $dest->getRegion(),
180164
'Attributes' => $dest->getAttributes(),
181165
'QueueName' => $dest->getQueueName(),
@@ -190,7 +174,7 @@ public function declareQueue(SqsDestination $dest): void
190174

191175
public function deleteQueue(SqsDestination $dest): void
192176
{
193-
$this->getClient()->deleteQueue([
177+
$this->client->deleteQueue([
194178
'QueueUrl' => $this->getQueueUrl($dest),
195179
]);
196180

‎pkg/sqs/SqsProducer.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,7 @@ public function send(Destination $destination, Message $message): void
7171
$arguments['MessageGroupId'] = $message->getMessageGroupId();
7272
}
7373

74-
$result = $this->context->getClient()->sendMessage($arguments);
74+
$result = $this->context->getSqsClient()->sendMessage($arguments);
7575

7676
if (false == $result->hasKey('MessageId')) {
7777
throw new \RuntimeException('Message was not sent');

0 commit comments

Comments
 (0)