[Summary view] [Print] [Text view]

   1  <?php
   2  /**
   3   * Zend Framework
   4   *
   5   * LICENSE
   6   *
   7   * This source file is subject to the new BSD license that is bundled
   8   * with this package in the file LICENSE.txt.
   9   * It is also available through the world-wide-web at this URL:
  10   * http://framework.zend.com/license/new-bsd
  11   * If you did not receive a copy of the license and are unable to
  12   * obtain it through the world-wide-web, please send an email
  13   * to [email protected] so we can send you a copy immediately.
  14   *
  15   * @category   Zend
  16   * @package    Zend_Service
  17   * @subpackage Amazon_Sqs
  18   * @copyright  Copyright (c) 2005-2010 Zend Technologies USA Inc. (http://www.zend.com)
  19   * @license    http://framework.zend.com/license/new-bsd     New BSD License
  20   * @version    $Id$
  21   */
  22  
  23  /**
  24   * @see Zend_Service_Amazon_Abstract
  25   */
  26  require_once 'Zend/Service/Amazon/Abstract.php';
  27  
  28  /**
  29   * @see Zend_Crypt_Hmac
  30   */
  31  require_once 'Zend/Crypt/Hmac.php';
  32  
  33  /**
  34   * Class for connecting to the Amazon Simple Queue Service (SQS)
  35   *
  36   * @category   Zend
  37   * @package    Zend_Service
  38   * @subpackage Amazon_Sqs
  39   * @copyright  Copyright (c) 2005-2010 Zend Technologies USA Inc. (http://www.zend.com)
  40   * @license    http://framework.zend.com/license/new-bsd     New BSD License
  41   * @see        http://aws.amazon.com/sqs/ Amazon Simple Queue Service
  42   */
  43  class Zend_Service_Amazon_Sqs extends Zend_Service_Amazon_Abstract
  44  {
  45      /**
  46       * Default timeout for createQueue() function
  47       */
  48      const CREATE_TIMEOUT_DEFAULT = 30;
  49  
  50      /**
  51       * HTTP end point for the Amazon SQS service
  52       */
  53      protected $_sqsEndpoint = 'queue.amazonaws.com';
  54  
  55      /**
  56       * The API version to use
  57       */
  58      protected $_sqsApiVersion = '2009-02-01';
  59  
  60      /**
  61       * Signature Version
  62       */
  63      protected $_sqsSignatureVersion = '2';
  64  
  65      /**
  66       * Signature Encoding Method
  67       */
  68      protected $_sqsSignatureMethod = 'HmacSHA256';
  69  
  70      /**
  71       * Constructor
  72       *
  73       * @param string $accessKey
  74       * @param string $secretKey
  75       * @param string $region
  76       */
  77      public function __construct($accessKey = null, $secretKey = null, $region = null)
  78      {
  79          parent::__construct($accessKey, $secretKey, $region);
  80      }
  81  
  82      /**
  83       * Create a new queue
  84       *
  85       * Visibility timeout is how long a message is left in the queue "invisible"
  86       * to other readers.  If the message is acknowleged (deleted) before the
  87       * timeout, then the message is deleted.  However, if the timeout expires
  88       * then the message will be made available to other queue readers.
  89       *
  90       * @param  string  $queue_name queue name
  91       * @param  integer $timeout    default visibility timeout
  92       * @return string|boolean
  93       * @throws Zend_Service_Amazon_Sqs_Exception
  94       */
  95      public function create($queue_name, $timeout = null)
  96      {
  97          $params = array();
  98          $params['QueueName'] = $queue_name;
  99          $timeout = ($timeout === null) ? self::CREATE_TIMEOUT_DEFAULT : (int)$timeout;
 100          $params['DefaultVisibilityTimeout'] = $timeout;
 101  
 102          $retry_count = 0;
 103  
 104          do {
 105              $retry  = false;
 106              $result = $this->_makeRequest(null, 'CreateQueue', $params);
 107  
 108              if ($result->CreateQueueResult->QueueUrl === null) {
 109                  if ($result->Error->Code == 'AWS.SimpleQueueService.QueueNameExists') {
 110                      return false;
 111                  } elseif ($result->Error->Code == 'AWS.SimpleQueueService.QueueDeletedRecently') {
 112                      // Must sleep for 60 seconds, then try re-creating the queue
 113                      sleep(60);
 114                      $retry = true;
 115                      $retry_count++;
 116                  } else {
 117                      require_once 'Zend/Service/Amazon/Sqs/Exception.php';
 118                      throw new Zend_Service_Amazon_Sqs_Exception($result->Error->Code);
 119                  }
 120              } else {
 121                  return (string) $result->CreateQueueResult->QueueUrl;
 122              }
 123  
 124          } while ($retry);
 125  
 126          return false;
 127      }
 128  
 129      /**
 130       * Delete a queue and all of it's messages
 131       *
 132       * Returns false if the queue is not found, true if the queue exists
 133       *
 134       * @param  string  $queue_url queue URL
 135       * @return boolean
 136       * @throws Zend_Service_Amazon_Sqs_Exception
 137       */
 138      public function delete($queue_url)
 139      {
 140          $result = $this->_makeRequest($queue_url, 'DeleteQueue');
 141  
 142          if ($result->Error->Code !== null) {
 143              require_once 'Zend/Service/Amazon/Sqs/Exception.php';
 144              throw new Zend_Service_Amazon_Sqs_Exception($result->Error->Code);
 145          }
 146  
 147          return true;
 148      }
 149  
 150      /**
 151       * Get an array of all available queues
 152       *
 153       * @return array
 154       * @throws Zend_Service_Amazon_Sqs_Exception
 155       */
 156      public function getQueues()
 157      {
 158          $result = $this->_makeRequest(null, 'ListQueues');
 159  
 160          if ($result->ListQueuesResult->QueueUrl === null) {
 161              require_once 'Zend/Service/Amazon/Sqs/Exception.php';
 162              throw new Zend_Service_Amazon_Sqs_Exception($result->Error->Code);
 163          }
 164  
 165          $queues = array();
 166          foreach ($result->ListQueuesResult->QueueUrl as $queue_url) {
 167              $queues[] = (string)$queue_url;
 168          }
 169  
 170          return $queues;
 171      }
 172  
 173      /**
 174       * Return the approximate number of messages in the queue
 175       *
 176       * @param  string  $queue_url Queue URL
 177       * @return integer
 178       * @throws Zend_Service_Amazon_Sqs_Exception
 179       */
 180      public function count($queue_url)
 181      {
 182          return (int)$this->getAttribute($queue_url, 'ApproximateNumberOfMessages');
 183      }
 184  
 185      /**
 186       * Send a message to the queue
 187       *
 188       * @param  string $queue_url Queue URL
 189       * @param  string $message   Message to send to the queue
 190       * @return string            Message ID
 191       * @throws Zend_Service_Amazon_Sqs_Exception
 192       */
 193      public function send($queue_url, $message)
 194      {
 195          $params = array();
 196          $params['MessageBody'] = urlencode($message);
 197  
 198          $checksum = md5($params['MessageBody']);
 199  
 200          $result = $this->_makeRequest($queue_url, 'SendMessage', $params);
 201  
 202          if ($result->SendMessageResult->MessageId === null) {
 203              require_once 'Zend/Service/Amazon/Sqs/Exception.php';
 204              throw new Zend_Service_Amazon_Sqs_Exception($result->Error->Code);
 205          } else if ((string) $result->SendMessageResult->MD5OfMessageBody != $checksum) {
 206              require_once 'Zend/Service/Amazon/Sqs/Exception.php';
 207              throw new Zend_Service_Amazon_Sqs_Exception('MD5 of body does not match message sent');
 208          }
 209  
 210          return (string) $result->SendMessageResult->MessageId;
 211      }
 212  
 213      /**
 214       * Get messages in the queue
 215       *
 216       * @param  string  $queue_url    Queue name
 217       * @param  integer $max_messages Maximum number of messages to return
 218       * @param  integer $timeout      Visibility timeout for these messages
 219       * @return array
 220       * @throws Zend_Service_Amazon_Sqs_Exception
 221       */
 222      public function receive($queue_url, $max_messages = null, $timeout = null)
 223      {
 224          $params = array();
 225  
 226          // If not set, the visibility timeout on the queue is used
 227          if ($timeout !== null) {
 228              $params['VisibilityTimeout'] = (int)$timeout;
 229          }
 230  
 231          // SQS will default to only returning one message
 232          if ($max_messages !== null) {
 233              $params['MaxNumberOfMessages'] = (int)$max_messages;
 234          }
 235  
 236          $result = $this->_makeRequest($queue_url, 'ReceiveMessage', $params);
 237  
 238          if ($result->ReceiveMessageResult->Message === null) {
 239              require_once 'Zend/Service/Amazon/Sqs/Exception.php';
 240              throw new Zend_Service_Amazon_Sqs_Exception($result->Error->Code);
 241          }
 242  
 243          $data = array();
 244          foreach ($result->ReceiveMessageResult->Message as $message) {
 245              $data[] = array(
 246                  'message_id' => (string)$message->MessageId,
 247                  'handle'     => (string)$message->ReceiptHandle,
 248                  'md5'        => (string)$message->MD5OfBody,
 249                  'body'       => urldecode((string)$message->Body),
 250              );
 251          }
 252  
 253          return $data;
 254      }
 255  
 256      /**
 257       * Delete a message from the queue
 258       *
 259       * Returns true if the message is deleted, false if the deletion is
 260       * unsuccessful.
 261       *
 262       * @param  string $queue_url  Queue URL
 263       * @param  string $handle     Message handle as returned by SQS
 264       * @return boolean
 265       * @throws Zend_Service_Amazon_Sqs_Exception
 266       */
 267      public function deleteMessage($queue_url, $handle)
 268      {
 269          $params = array();
 270          $params['ReceiptHandle'] = (string)$handle;
 271  
 272          $result = $this->_makeRequest($queue_url, 'DeleteMessage', $params);
 273  
 274          if ($result->Error->Code !== null) {
 275              return false;
 276          }
 277  
 278          // Will always return true unless ReceiptHandle is malformed
 279          return true;
 280      }
 281  
 282      /**
 283       * Get the attributes for the queue
 284       *
 285       * @param  string $queue_url  Queue URL
 286       * @param  string $attribute
 287       * @return string
 288       * @throws Zend_Service_Amazon_Sqs_Exception
 289       */
 290      public function getAttribute($queue_url, $attribute = 'All')
 291      {
 292          $params = array();
 293          $params['AttributeName'] = $attribute;
 294  
 295          $result = $this->_makeRequest($queue_url, 'GetQueueAttributes', $params);
 296  
 297          if ($result->GetQueueAttributesResult->Attribute === null) {
 298              require_once 'Zend/Service/Amazon/Sqs/Exception.php';
 299              throw new Zend_Service_Amazon_Sqs_Exception($result->Error->Code);
 300          }
 301          
 302          if(count($result->GetQueueAttributesResult->Attribute) > 1) {
 303              $attr_result = array();
 304              foreach($result->GetQueueAttributesResult->Attribute as $attribute) {
 305                  $attr_result[(string)$attribute->Name] = (string)$attribute->Value;
 306              }
 307              return $attr_result;
 308          } else {
 309              return (string) $result->GetQueueAttributesResult->Attribute->Value;
 310          }
 311      }
 312  
 313      /**
 314       * Make a request to Amazon SQS
 315       *
 316       * @param  string           $queue  Queue Name
 317       * @param  string           $action SQS action
 318       * @param  array            $params
 319       * @return SimpleXMLElement
 320       */
 321      private function _makeRequest($queue_url, $action, $params = array())
 322      {
 323          $params['Action'] = $action;
 324          $params = $this->addRequiredParameters($queue_url, $params);
 325  
 326          if ($queue_url === null) {
 327              $queue_url = '/';
 328          }
 329  
 330          $client = self::getHttpClient();
 331  
 332          switch ($action) {
 333              case 'ListQueues':
 334              case 'CreateQueue':
 335                  $client->setUri('http://'.$this->_sqsEndpoint);
 336                  break;
 337              default:
 338                  $client->setUri($queue_url);
 339                  break;
 340          }
 341  
 342          $retry_count = 0;
 343  
 344          do {
 345              $retry = false;
 346  
 347              $client->resetParameters();
 348              $client->setParameterGet($params);
 349  
 350              $response = $client->request('GET');
 351  
 352              $response_code = $response->getStatus();
 353  
 354              // Some 5xx errors are expected, so retry automatically
 355              if ($response_code >= 500 && $response_code < 600 && $retry_count <= 5) {
 356                  $retry = true;
 357                  $retry_count++;
 358                  sleep($retry_count / 4 * $retry_count);
 359              }
 360          } while ($retry);
 361  
 362          unset($client);
 363  
 364          return new SimpleXMLElement($response->getBody());
 365      }
 366  
 367      /**
 368       * Adds required authentication and version parameters to an array of
 369       * parameters
 370       *
 371       * The required parameters are:
 372       * - AWSAccessKey
 373       * - SignatureVersion
 374       * - Timestamp
 375       * - Version and
 376       * - Signature
 377       *
 378       * If a required parameter is already set in the <tt>$parameters</tt> array,
 379       * it is overwritten.
 380       *
 381       * @param  string $queue_url  Queue URL
 382       * @param  array  $parameters the array to which to add the required
 383       *                            parameters.
 384       * @return array
 385       */
 386      protected function addRequiredParameters($queue_url, array $parameters)
 387      {
 388          $parameters['AWSAccessKeyId']   = $this->_getAccessKey();
 389          $parameters['SignatureVersion'] = $this->_sqsSignatureVersion;
 390          $parameters['Timestamp']        = gmdate('Y-m-d\TH:i:s\Z', time()+10);
 391          $parameters['Version']          = $this->_sqsApiVersion;
 392          $parameters['SignatureMethod']  = $this->_sqsSignatureMethod;
 393          $parameters['Signature']        = $this->_signParameters($queue_url, $parameters);
 394  
 395          return $parameters;
 396      }
 397  
 398      /**
 399       * Computes the RFC 2104-compliant HMAC signature for request parameters
 400       *
 401       * This implements the Amazon Web Services signature, as per the following
 402       * specification:
 403       *
 404       * 1. Sort all request parameters (including <tt>SignatureVersion</tt> and
 405       *    excluding <tt>Signature</tt>, the value of which is being created),
 406       *    ignoring case.
 407       *
 408       * 2. Iterate over the sorted list and append the parameter name (in its
 409       *    original case) and then its value. Do not URL-encode the parameter
 410       *    values before constructing this string. Do not use any separator
 411       *    characters when appending strings.
 412       *
 413       * @param  string $queue_url  Queue URL
 414       * @param  array  $parameters the parameters for which to get the signature.
 415       *
 416       * @return string the signed data.
 417       */
 418      protected function _signParameters($queue_url, array $paramaters)
 419      {
 420          $data = "GET\n";
 421          $data .= $this->_sqsEndpoint . "\n";
 422          if ($queue_url !== null) {
 423              $data .= parse_url($queue_url, PHP_URL_PATH);
 424          }
 425          else {
 426              $data .= '/';
 427          }
 428          $data .= "\n";
 429  
 430          uksort($paramaters, 'strcmp');
 431          unset($paramaters['Signature']);
 432  
 433          $arrData = array();
 434          foreach($paramaters as $key => $value) {
 435              $arrData[] = $key . '=' . str_replace('%7E', '~', urlencode($value));
 436          }
 437  
 438          $data .= implode('&', $arrData);
 439  
 440          $hmac = Zend_Crypt_Hmac::compute($this->_getSecretKey(), 'SHA256', $data, Zend_Crypt_Hmac::BINARY);
 441  
 442          return base64_encode($hmac);
 443      }
 444  }