|
1
|
|
|
<?php |
|
2
|
|
|
|
|
3
|
|
|
namespace Kaliop\Queueing\Plugins\SQSBundle\Adapter\SQS; |
|
4
|
|
|
|
|
5
|
|
|
use Kaliop\QueueingBundle\Queue\ProducerInterface; |
|
6
|
|
|
use Aws\Sqs\SqsClient; |
|
7
|
|
|
use Aws\TraceMiddleware; |
|
8
|
|
|
|
|
9
|
|
|
class Producer implements ProducerInterface |
|
10
|
|
|
{ |
|
11
|
|
|
/** @var \Aws\Sqs\SqsClient */ |
|
12
|
|
|
protected $client; |
|
13
|
|
|
protected $queueUrl; |
|
14
|
|
|
protected $debug = false; |
|
15
|
|
|
protected $contentType = 'text/plain'; |
|
16
|
|
|
// The message attribute used to store content-type. To be kept in sync with the Consumer |
|
17
|
|
|
protected $contentTypeAttribute = 'contentType'; |
|
18
|
|
|
protected $routingKeyAttribute = 'routingKey'; |
|
19
|
|
|
protected $messageGroupId; |
|
20
|
|
|
|
|
21
|
|
|
/** |
|
22
|
|
|
* @param array $config - minimum seems to be: 'credentials', 'region', 'version' |
|
23
|
|
|
* @see \Aws\AwsClient::__construct for the full list |
|
24
|
|
|
* @see http://docs.aws.amazon.com/aws-sdk-php/v3/guide/guide/configuration.html |
|
25
|
|
|
*/ |
|
26
|
10 |
|
public function __construct(array $config) |
|
27
|
|
|
{ |
|
28
|
10 |
|
$this->client = new SqsClient($config); |
|
29
|
10 |
|
} |
|
30
|
|
|
|
|
31
|
|
|
/** |
|
32
|
|
|
* Enabled debug. At the moment can not disable it |
|
33
|
|
|
* |
|
34
|
|
|
* @param $debug |
|
35
|
|
|
* @return $this |
|
36
|
|
|
* |
|
37
|
|
|
* @todo test if using $handlerList->removeByInstance we can disable debug as well |
|
38
|
|
|
*/ |
|
39
|
10 |
View Code Duplication |
public function setDebug($debug) { |
|
|
|
|
|
|
40
|
10 |
|
if ($debug == $this->debug) { |
|
41
|
10 |
|
return $this; |
|
42
|
|
|
} |
|
43
|
|
|
if ($debug) { |
|
44
|
|
|
$handlerList = $this->client->getHandlerList(); |
|
45
|
|
|
$handlerList->interpose(new TraceMiddleware($debug === true ? [] : $debug)); |
|
46
|
|
|
} |
|
47
|
|
|
|
|
48
|
|
|
return $this; |
|
49
|
|
|
} |
|
50
|
|
|
|
|
51
|
|
|
/** |
|
52
|
|
|
* @param string $queueName NB: complete queue name as used by SQS |
|
|
|
|
|
|
53
|
|
|
* @param string $queueUrl |
|
54
|
|
|
* @return Producer |
|
55
|
|
|
* @todo test that we can successfully send messages to 2 queues using the same SqsClient |
|
56
|
|
|
*/ |
|
57
|
10 |
|
public function setQueueUrl($queueUrl) |
|
58
|
|
|
{ |
|
59
|
10 |
|
$this->queueUrl = $queueUrl; |
|
60
|
|
|
|
|
61
|
10 |
|
return $this; |
|
62
|
|
|
} |
|
63
|
|
|
|
|
64
|
|
|
/** |
|
65
|
|
|
* @return string the complete queue name as used by SQS |
|
66
|
|
|
*/ |
|
67
|
10 |
|
public function getQueueUrl() |
|
68
|
|
|
{ |
|
69
|
10 |
|
return $this->queueUrl; |
|
70
|
|
|
} |
|
71
|
|
|
|
|
72
|
1 |
|
public function setMessageGroupId($messageGroupId) |
|
73
|
|
|
{ |
|
74
|
1 |
|
$this->messageGroupId = $messageGroupId; |
|
75
|
|
|
|
|
76
|
1 |
|
return $this; |
|
77
|
|
|
} |
|
78
|
|
|
|
|
79
|
|
|
/** |
|
80
|
|
|
* Publishes the message and does nothing with the properties |
|
81
|
|
|
* |
|
82
|
|
|
* @param string $msgBody |
|
83
|
|
|
* @param string $routingKey |
|
84
|
|
|
* @param array $additionalProperties see https://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html#sendmessage |
|
85
|
|
|
* |
|
86
|
|
|
* @todo support custom message attributes (possible via $additionalProperties) |
|
87
|
|
|
* @todo support custom delaySeconds (possible via $additionalProperties) |
|
88
|
|
|
* @todo support custom MessageDeduplicationId (possible via $additionalProperties) |
|
89
|
|
|
*/ |
|
90
|
5 |
|
public function publish($msgBody, $routingKey = '', $additionalProperties = array()) |
|
91
|
|
|
{ |
|
92
|
5 |
|
$this->client->sendMessage(array_merge( |
|
93
|
|
|
array( |
|
94
|
5 |
|
'QueueUrl' => $this->queueUrl, |
|
95
|
5 |
|
'MessageBody' => $msgBody, |
|
96
|
|
|
), |
|
97
|
5 |
|
$this->getClientParams($routingKey, $additionalProperties) |
|
98
|
|
|
)); |
|
99
|
5 |
|
} |
|
100
|
|
|
|
|
101
|
|
|
/** |
|
102
|
|
|
* @see http://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html#sendmessagebatch |
|
103
|
|
|
* @param array[] $messages each element is an array that must contain: |
|
104
|
|
|
* - msgBody (string) |
|
105
|
|
|
* - routingKey (string, optional) |
|
106
|
|
|
* - additionalProperties (array, optional) |
|
107
|
|
|
*/ |
|
108
|
|
|
public function batchPublish(array $messages) |
|
109
|
|
|
{ |
|
110
|
|
|
$j = 0; |
|
111
|
|
|
for ($i = 0; $i < count($messages); $i += 10) { |
|
|
|
|
|
|
112
|
|
|
$entries = array(); |
|
113
|
|
|
$toSend = array_slice($messages, $i, 10); |
|
114
|
|
|
foreach($toSend as $message) { |
|
115
|
|
|
$entries[] = array_merge( |
|
116
|
|
|
array( |
|
117
|
|
|
'MessageBody' => $message['msgBody'], |
|
118
|
|
|
'Id' => $j++ |
|
119
|
|
|
), |
|
120
|
|
|
$this->getClientParams(@$message['routingKey'], @$message['additionalProperties']) |
|
121
|
|
|
); |
|
122
|
|
|
} |
|
123
|
|
|
|
|
124
|
|
|
$result = $this->client->sendMessageBatch( |
|
125
|
|
|
array( |
|
126
|
|
|
'QueueUrl' => $this->queueUrl, |
|
127
|
|
|
'Entries' => $entries, |
|
128
|
|
|
) |
|
129
|
|
|
); |
|
130
|
|
|
|
|
131
|
|
|
if (($ok = count($result->get('Successful'))) != ($tot = count($toSend))) { |
|
132
|
|
|
throw new \RuntimeException("Batch sending of messages failed - $ok ok out of $tot"); |
|
133
|
|
|
} |
|
134
|
|
|
} |
|
135
|
|
|
} |
|
136
|
|
|
|
|
137
|
|
|
/** |
|
138
|
|
|
* Allows callers to do whatever they want with the client - useful to the Queue Mgr |
|
139
|
|
|
* |
|
140
|
|
|
* @param string $method |
|
141
|
|
|
* @param array $args |
|
142
|
|
|
* @return mixed |
|
143
|
|
|
*/ |
|
144
|
10 |
|
public function call($method, array $args = array()) |
|
145
|
|
|
{ |
|
146
|
10 |
|
return $this->client->$method(array_merge($args, $this->getClientParams())); |
|
147
|
|
|
} |
|
148
|
|
|
|
|
149
|
|
|
/** |
|
150
|
|
|
* Prepares the extra parameters to be injected into calls made via the SQS Client |
|
151
|
|
|
* @param string $routingKey |
|
152
|
|
|
* @param array $additionalProperties |
|
153
|
|
|
* @return array |
|
154
|
|
|
* |
|
155
|
|
|
* @todo shall we throw if $additionalProperties['expiration'] is set, since we don't support it ? |
|
156
|
|
|
*/ |
|
157
|
10 |
|
protected function getClientParams($routingKey = '', array $additionalProperties = array()) |
|
158
|
|
|
{ |
|
159
|
|
|
$result = array( |
|
160
|
|
|
'MessageAttributes' => array( |
|
161
|
10 |
|
$this->contentTypeAttribute => array('StringValue' => $this->contentType, 'DataType' => 'String'), |
|
162
|
|
|
) |
|
163
|
|
|
); |
|
164
|
10 |
|
if ($routingKey != '') { |
|
165
|
3 |
|
$result['MessageAttributes'][$this->routingKeyAttribute] = array('StringValue' => $routingKey, 'DataType' => 'String'); |
|
166
|
|
|
} |
|
167
|
|
|
|
|
168
|
10 |
|
if ($this->messageGroupId != null) { |
|
169
|
1 |
|
$result['MessageGroupId'] = $this->messageGroupId; |
|
170
|
|
|
} |
|
171
|
|
|
|
|
172
|
10 |
|
$result = array_merge($result, $additionalProperties); |
|
173
|
|
|
|
|
174
|
10 |
|
return $result; |
|
175
|
|
|
} |
|
176
|
|
|
|
|
177
|
|
|
/** |
|
178
|
|
|
* @param string $contentType |
|
179
|
|
|
* @return Producer |
|
180
|
|
|
* @throws \Exception if unsupported contentType is used |
|
181
|
|
|
*/ |
|
182
|
5 |
|
public function setContentType($contentType) |
|
183
|
|
|
{ |
|
184
|
5 |
|
$this->contentType = $contentType; |
|
185
|
|
|
|
|
186
|
5 |
|
return $this; |
|
187
|
|
|
} |
|
188
|
|
|
} |
|
189
|
|
|
|
kaliop-uk /
kueueingbundle-sqs
Loading history...
Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation.
You can also find more detailed suggestions in the “Code” section of your repository.