SqsClient.php 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. <?php
  2. namespace Aws\Sqs;
  3. use Aws\AwsClient;
  4. use Aws\CommandInterface;
  5. use Aws\Sqs\Exception\SqsException;
  6. use GuzzleHttp\Psr7\Uri;
  7. use GuzzleHttp\Psr7\UriResolver;
  8. use Psr\Http\Message\RequestInterface;
  9. /**
  10. * Client used to interact Amazon Simple Queue Service (Amazon SQS)
  11. *
  12. * @method \Aws\Result addPermission(array $args = [])
  13. * @method \GuzzleHttp\Promise\Promise addPermissionAsync(array $args = [])
  14. * @method \Aws\Result changeMessageVisibility(array $args = [])
  15. * @method \GuzzleHttp\Promise\Promise changeMessageVisibilityAsync(array $args = [])
  16. * @method \Aws\Result changeMessageVisibilityBatch(array $args = [])
  17. * @method \GuzzleHttp\Promise\Promise changeMessageVisibilityBatchAsync(array $args = [])
  18. * @method \Aws\Result createQueue(array $args = [])
  19. * @method \GuzzleHttp\Promise\Promise createQueueAsync(array $args = [])
  20. * @method \Aws\Result deleteMessage(array $args = [])
  21. * @method \GuzzleHttp\Promise\Promise deleteMessageAsync(array $args = [])
  22. * @method \Aws\Result deleteMessageBatch(array $args = [])
  23. * @method \GuzzleHttp\Promise\Promise deleteMessageBatchAsync(array $args = [])
  24. * @method \Aws\Result deleteQueue(array $args = [])
  25. * @method \GuzzleHttp\Promise\Promise deleteQueueAsync(array $args = [])
  26. * @method \Aws\Result getQueueAttributes(array $args = [])
  27. * @method \GuzzleHttp\Promise\Promise getQueueAttributesAsync(array $args = [])
  28. * @method \Aws\Result getQueueUrl(array $args = [])
  29. * @method \GuzzleHttp\Promise\Promise getQueueUrlAsync(array $args = [])
  30. * @method \Aws\Result listDeadLetterSourceQueues(array $args = [])
  31. * @method \GuzzleHttp\Promise\Promise listDeadLetterSourceQueuesAsync(array $args = [])
  32. * @method \Aws\Result listQueues(array $args = [])
  33. * @method \GuzzleHttp\Promise\Promise listQueuesAsync(array $args = [])
  34. * @method \Aws\Result purgeQueue(array $args = [])
  35. * @method \GuzzleHttp\Promise\Promise purgeQueueAsync(array $args = [])
  36. * @method \Aws\Result receiveMessage(array $args = [])
  37. * @method \GuzzleHttp\Promise\Promise receiveMessageAsync(array $args = [])
  38. * @method \Aws\Result removePermission(array $args = [])
  39. * @method \GuzzleHttp\Promise\Promise removePermissionAsync(array $args = [])
  40. * @method \Aws\Result sendMessage(array $args = [])
  41. * @method \GuzzleHttp\Promise\Promise sendMessageAsync(array $args = [])
  42. * @method \Aws\Result sendMessageBatch(array $args = [])
  43. * @method \GuzzleHttp\Promise\Promise sendMessageBatchAsync(array $args = [])
  44. * @method \Aws\Result setQueueAttributes(array $args = [])
  45. * @method \GuzzleHttp\Promise\Promise setQueueAttributesAsync(array $args = [])
  46. */
  47. class SqsClient extends AwsClient
  48. {
  49. public function __construct(array $config)
  50. {
  51. parent::__construct($config);
  52. $list = $this->getHandlerList();
  53. $list->appendBuild($this->queueUrl(), 'sqs.queue_url');
  54. $list->appendSign($this->validateMd5(), 'sqs.md5');
  55. }
  56. /**
  57. * Converts a queue URL into a queue ARN.
  58. *
  59. * @param string $queueUrl The queue URL to perform the action on.
  60. * Retrieved when the queue is first created.
  61. *
  62. * @return string An ARN representation of the queue URL.
  63. */
  64. public function getQueueArn($queueUrl)
  65. {
  66. $queueArn = strtr($queueUrl, array(
  67. 'http://' => 'arn:aws:',
  68. 'https://' => 'arn:aws:',
  69. '.amazonaws.com' => '',
  70. '/' => ':',
  71. '.' => ':',
  72. ));
  73. // Cope with SQS' .fifo / :fifo arn inconsistency
  74. if (substr($queueArn, -5) === ':fifo') {
  75. $queueArn = substr_replace($queueArn, '.fifo', -5);
  76. }
  77. return $queueArn;
  78. }
  79. /**
  80. * Moves the URI of the queue to the URI in the input parameter.
  81. *
  82. * @return callable
  83. */
  84. private function queueUrl()
  85. {
  86. return static function (callable $handler) {
  87. return function (
  88. CommandInterface $c,
  89. RequestInterface $r = null
  90. ) use ($handler) {
  91. if ($c->hasParam('QueueUrl')) {
  92. $r = $r->withUri(UriResolver::resolve(
  93. $r->getUri(),
  94. new Uri($c['QueueUrl'])
  95. ));
  96. }
  97. return $handler($c, $r);
  98. };
  99. };
  100. }
  101. /**
  102. * Calculates the expected md5 hash of message attributes according to the encoding
  103. * scheme detailed in SQS documentation.
  104. *
  105. * @param array $message Message containing attributes for validation.
  106. * Retrieved when using MessageAttributeNames on
  107. * ReceiveMessage.
  108. *
  109. * @return string|null The md5 hash of the message attributes according to
  110. * the encoding scheme. Returns null when there are no
  111. * attributes.
  112. * @link http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-attributes.html#message-attributes-items-validation
  113. */
  114. private static function calculateMessageAttributesMd5($message)
  115. {
  116. if (empty($message['MessageAttributes'])
  117. || !is_array($message['MessageAttributes'])
  118. ) {
  119. return null;
  120. }
  121. ksort($message['MessageAttributes']);
  122. $attributeValues = "";
  123. foreach ($message['MessageAttributes'] as $name => $details) {
  124. $attributeValues .= self::getEncodedStringPiece($name);
  125. $attributeValues .= self::getEncodedStringPiece($details['DataType']);
  126. if (substr($details['DataType'], 0, 6) === 'Binary') {
  127. $attributeValues .= pack('c', 0x02);
  128. $attributeValues .= self::getEncodedBinaryPiece(
  129. $details['BinaryValue']
  130. );
  131. } else {
  132. $attributeValues .= pack('c', 0x01);
  133. $attributeValues .= self::getEncodedStringPiece(
  134. $details['StringValue']
  135. );
  136. }
  137. }
  138. return md5($attributeValues);
  139. }
  140. private static function calculateBodyMd5($message)
  141. {
  142. return md5($message['Body']);
  143. }
  144. private static function getEncodedStringPiece($piece)
  145. {
  146. $utf8Piece = iconv(
  147. mb_detect_encoding($piece, mb_detect_order(), true),
  148. "UTF-8",
  149. $piece
  150. );
  151. return self::getFourBytePieceLength($utf8Piece) . $utf8Piece;
  152. }
  153. private static function getEncodedBinaryPiece($piece)
  154. {
  155. return self::getFourBytePieceLength($piece) . $piece;
  156. }
  157. private static function getFourBytePieceLength($piece)
  158. {
  159. return pack('N', (int)strlen($piece));
  160. }
  161. /**
  162. * Validates ReceiveMessage body and message attribute MD5s.
  163. *
  164. * @return callable
  165. */
  166. private function validateMd5()
  167. {
  168. return static function (callable $handler) {
  169. return function (
  170. CommandInterface $c,
  171. RequestInterface $r = null
  172. ) use ($handler) {
  173. if ($c->getName() !== 'ReceiveMessage') {
  174. return $handler($c, $r);
  175. }
  176. return $handler($c, $r)
  177. ->then(
  178. function ($result) use ($c, $r) {
  179. foreach ((array) $result['Messages'] as $msg) {
  180. $bodyMd5 = self::calculateBodyMd5($msg);
  181. if (isset($msg['MD5OfBody'])
  182. && $bodyMd5 !== $msg['MD5OfBody']
  183. ) {
  184. throw new SqsException(
  185. sprintf(
  186. 'MD5 mismatch. Expected %s, found %s',
  187. $msg['MD5OfBody'],
  188. $bodyMd5
  189. ),
  190. $c,
  191. [
  192. 'code' => 'ClientChecksumMismatch',
  193. 'request' => $r
  194. ]
  195. );
  196. }
  197. if (isset($msg['MD5OfMessageAttributes'])) {
  198. $messageAttributesMd5 = self::calculateMessageAttributesMd5($msg);
  199. if ($messageAttributesMd5 !== $msg['MD5OfMessageAttributes']) {
  200. throw new SqsException(
  201. sprintf(
  202. 'Attribute MD5 mismatch. Expected %s, found %s',
  203. $msg['MD5OfMessageAttributes'],
  204. $messageAttributesMd5
  205. ? $messageAttributesMd5
  206. : 'No Attributes'
  207. ),
  208. $c,
  209. [
  210. 'code' => 'ClientChecksumMismatch',
  211. 'request' => $r
  212. ]
  213. );
  214. }
  215. } else if (isset($msg['MessageAttributes'])) {
  216. throw new SqsException(
  217. sprintf(
  218. 'No Attribute MD5 found. Expected %s',
  219. self::calculateMessageAttributesMd5($msg)
  220. ),
  221. $c,
  222. [
  223. 'code' => 'ClientChecksumMismatch',
  224. 'request' => $r
  225. ]
  226. );
  227. }
  228. }
  229. return $result;
  230. }
  231. );
  232. };
  233. };
  234. }
  235. }