1 <?php

     2 /**

     3  * LICENSE

     4  *

     5  * This source file is subject to the new BSD license that is bundled

     6  * with this package in the file LICENSE.txt.

     7  * It is also available through the world-wide-web at this URL:

     8  * http://framework.zend.com/license/new-bsd

     9  * If you did not receive a copy of the license and are unable to

    10  * obtain it through the world-wide-web, please send an email

    11  * to license@zend.com so we can send you a copy immediately.

    12  *

    13  * @category   Zend

    14  * @package    Zend_Cloud

    15  * @subpackage QueueService

    16  * @copyright  Copyright (c) 2005-2010 Zend Technologies USA Inc. (http://www.zend.com)

    17  * @license    http://framework.zend.com/license/new-bsd     New BSD License

    18  */

    19 

    20 require_once 'Zend/Cloud/QueueService/Adapter/AbstractAdapter.php';

    21 require_once 'Zend/Cloud/QueueService/Exception.php';

    22 require_once 'Zend/Service/WindowsAzure/Storage/Queue.php';

    23 

    24 /**

    25  * WindowsAzure adapter for simple queue service.

    26  *

    27  * @category   Zend

    28  * @package    Zend_Cloud

    29  * @subpackage QueueService

    30  * @copyright  Copyright (c) 2005-2010 Zend Technologies USA Inc. (http://www.zend.com)

    31  * @license    http://framework.zend.com/license/new-bsd     New BSD License

    32  */

    33 class Zend_Cloud_QueueService_Adapter_WindowsAzure 

    34     extends Zend_Cloud_QueueService_Adapter_AbstractAdapter

    35 {

    36     /**

    37      * Option array keys for the Windows Azure adapter.

    38      */

    39     const ACCOUNT_NAME      = 'storage_accountname';

    40     const ACCOUNT_KEY       = 'storage_accountkey';

    41     const HOST              = "storage_host";

    42     const PROXY_HOST        = "storage_proxy_host";

    43     const PROXY_PORT        = "storage_proxy_port";

    44     const PROXY_CREDENTIALS = "storage_proxy_credentials";

    45 

    46     /** list options */

    47     const LIST_PREFIX      = 'prefix';

    48     const LIST_MAX_RESULTS = 'max_results';

    49 

    50     /** message options */

    51     const MESSAGE_TTL = 'ttl';

    52     

    53     const DEFAULT_HOST = Zend_Service_WindowsAzure_Storage::URL_CLOUD_QUEUE;

    54 

    55     /**

    56      * Storage client

    57      * 

    58      * @var Zend_Service_WindowsAzure_Storage_Queue

    59      */

    60     protected $_storageClient = null;

    61     

    62     /**

    63      * Constructor

    64      * 

    65      * @param  array|Zend_Config $options 

    66      * @return void

    67      */

    68     public function __construct($options = array())

    69     {

    70         if ($options instanceof Zend_Config) {

    71             $options = $options->toArray();

    72         }

    73 

    74         if (!is_array($options)) {

    75             throw new Zend_Cloud_QueueService_Exception('Invalid options provided');

    76         }

    77 

    78         if (isset($options[self::MESSAGE_CLASS])) {

    79             $this->setMessageClass($options[self::MESSAGE_CLASS]);

    80         }

    81 

    82         if (isset($options[self::MESSAGESET_CLASS])) {

    83             $this->setMessageSetClass($options[self::MESSAGESET_CLASS]);

    84         }

    85 

    86         // Build Zend_Service_WindowsAzure_Storage_Blob instance

    87         if (!isset($options[self::HOST])) {

    88             $host = self::DEFAULT_HOST;

    89         } else {

    90             $host = $options[self::HOST];

    91         }

    92         if (! isset($options[self::ACCOUNT_NAME])) {

    93             throw new Zend_Cloud_Storage_Exception('No Windows Azure account name provided.');

    94         }

    95         if (! isset($options[self::ACCOUNT_KEY])) {

    96             throw new Zend_Cloud_Storage_Exception('No Windows Azure account key provided.');

    97         }

    98         try {

    99             // TODO: support $usePathStyleUri and $retryPolicy

   100             $this->_storageClient = new Zend_Service_WindowsAzure_Storage_Queue(

   101                 $host, $options[self::ACCOUNT_NAME], $options[self::ACCOUNT_KEY]);

   102             // Parse other options

   103             if (! empty($options[self::PROXY_HOST])) {

   104                 $proxyHost = $options[self::PROXY_HOST];

   105                 $proxyPort = isset($options[self::PROXY_PORT]) ? $options[self::PROXY_PORT] : 8080;

   106                 $proxyCredentials = isset($options[self::PROXY_CREDENTIALS]) ? $options[self::PROXY_CREDENTIALS] : '';

   107                 $this->_storageClient->setProxy(true, $proxyHost, $proxyPort, $proxyCredentials);

   108             }

   109             if (isset($options[self::HTTP_ADAPTER])) {

   110                 $this->_storageClient->setHttpClientChannel($httpAdapter);

   111             }

   112         } catch(Zend_Service_WindowsAzure_Exception $e) {

   113             throw new Zend_Cloud_QueueService_Exception('Error on create: '.$e->getMessage(), $e->getCode(), $e);

   114         }

   115             

   116     }

   117     

   118     /**

   119      * Create a queue. Returns the ID of the created queue (typically the URL).

   120      * It may take some time to create the queue. Check your vendor's

   121      * documentation for details.

   122      *

   123      * @param  string $name

   124      * @param  array  $options

   125      * @return string Queue ID (typically URL)

   126      */

   127     public function createQueue($name, $options = null)

   128     {

   129         try {

   130             $queue = $this->_storageClient->createQueue($name, $options);

   131             return $queue->Name;

   132         } catch (Zend_Service_WindowsAzure_Exception $e) {

   133             throw new Zend_Cloud_QueueService_Exception('Error on queue creation: '.$e->getMessage(), $e->getCode(), $e);

   134         }

   135     }

   136     

   137     /**

   138      * Delete a queue. All messages in the queue will also be deleted.

   139      *

   140      * @param  string $queueId

   141      * @param  array  $options

   142      * @return boolean true if successful, false otherwise

   143      */

   144     public function deleteQueue($queueId, $options = null)

   145     {

   146         try {

   147             if ($queueId instanceof Zend_Service_WindowsAzure_Storage_QueueInstance) {

   148                 $queueId = $queueId->Name;

   149             }

   150             return $this->_storageClient->deleteQueue($queueId);

   151         } catch (Zend_Service_WindowsAzure_Exception $e) {

   152             throw new Zend_Cloud_QueueService_Exception('Error on queue deletion: '.$e->getMessage(), $e->getCode(), $e);

   153         }

   154     }

   155     

   156     /**

   157      * List all queues.

   158      *

   159      * @param  array $options

   160      * @return array Queue IDs

   161      */

   162     public function listQueues($options = null)

   163     {

   164         $prefix = $maxResults = null;

   165         if (is_array($options)) {

   166             isset($options[self::LIST_PREFIX]) ? $prefix = $options[self::LIST_PREFIX] : null;

   167             isset($options[self::LIST_MAX_RESULTS]) ? $maxResults = $options[self::LIST_MAX_RESULTS] : null;

   168         }

   169         try {

   170             $queues =  $this->_storageClient->listQueues($prefix, $maxResults);

   171             $result = array();

   172             foreach ($queues as $queue) {

   173                 $result[] = $queue->Name;

   174             }

   175             return $result;

   176         } catch (Zend_Service_WindowsAzure_Exception $e) {

   177             throw new Zend_Cloud_QueueService_Exception('Error on listing queues: '.$e->getMessage(), $e->getCode(), $e);

   178         }

   179     }

   180     

   181     /**

   182      * Get a key/value array of metadata for the given queue.

   183      *

   184      * @param  string $queueId

   185      * @param  array  $options

   186      * @return array

   187      */

   188     public function fetchQueueMetadata($queueId, $options = null)

   189     {

   190         try {

   191             if ($queueId instanceof Zend_Service_WindowsAzure_Storage_QueueInstance) {

   192                 $queueId = $queueId->Name;

   193             }

   194             return $this->_storageClient->getQueueMetadata($queueId);

   195         } catch (Zend_Service_WindowsAzure_Exception $e) {

   196             throw new Zend_Cloud_QueueService_Exception('Error on fetching queue metadata: '.$e->getMessage(), $e->getCode(), $e);

   197         }

   198     }

   199     

   200     /**

   201      * Store a key/value array of metadata for the specified queue.

   202      * WARNING: This operation overwrites any metadata that is located at 

   203      * $destinationPath. Some adapters may not support this method.

   204      * 

   205      * @param  string $queueId

   206      * @param  array  $metadata

   207      * @param  array  $options

   208      * @return void

   209      */

   210     public function storeQueueMetadata($queueId, $metadata, $options = null)

   211     {

   212         try {

   213             if ($queueId instanceof Zend_Service_WindowsAzure_Storage_QueueInstance) {

   214                 $queueId = $queueId->Name;

   215             }

   216             return $this->_storageClient->setQueueMetadata($queueId, $metadata);

   217         } catch (Zend_Service_WindowsAzure_Exception $e) {

   218             throw new Zend_Cloud_QueueService_Exception('Error on setting queue metadata: '.$e->getMessage(), $e->getCode(), $e);

   219         }

   220     }

   221     

   222     /**

   223      * Send a message to the specified queue.

   224      * 

   225      * @param  string $queueId

   226      * @param  string $message

   227      * @param  array  $options

   228      * @return string Message ID

   229      */

   230     public function sendMessage($queueId, $message, $options = null)

   231     {

   232         try {

   233             if ($queueId instanceof Zend_Service_WindowsAzure_Storage_QueueInstance) {

   234                 $queueId = $queueId->Name;

   235             }

   236             return $this->_storageClient->putMessage(

   237                 $queueId, $message, $options[self::MESSAGE_TTL]

   238             );

   239         } catch (Zend_Service_WindowsAzure_Exception $e) {

   240             throw new Zend_Cloud_QueueService_Exception('Error on sending message: '.$e->getMessage(), $e->getCode(), $e);

   241         }

   242     }

   243     

   244     /**

   245      * Recieve at most $max messages from the specified queue and return the

   246      * message IDs for messages recieved.

   247      * 

   248      * @param  string $queueId

   249      * @param  int    $max

   250      * @param  array  $options

   251      * @return Zend_Cloud_QueueService_Message[]

   252      */

   253     public function receiveMessages($queueId, $max = 1, $options = null)

   254     {

   255         try {

   256             if ($queueId instanceof Zend_Service_WindowsAzure_Storage_QueueInstance) {

   257                 $queueId = $queueId->Name;

   258             }

   259             if (isset($options[self::VISIBILITY_TIMEOUT])) {

   260                 $visibility = $options[self::VISIBILITY_TIMEOUT];

   261             } else {

   262                 $visibility = self::DEFAULT_TIMEOUT;

   263             }

   264             return $this->_makeMessages($this->_storageClient->getMessages($queueId, $max, $visibility, false));

   265         } catch (Zend_Service_WindowsAzure_Exception $e) {

   266             throw new Zend_Cloud_QueueService_Exception('Error on recieving messages: '.$e->getMessage(), $e->getCode(), $e);

   267         }

   268     }

   269     

   270     /**

   271      * Create Zend_Cloud_QueueService_Message array for

   272      * Azure messages.

   273      *  

   274      * @param array $messages

   275      * @return Zend_Cloud_QueueService_Message[]

   276      */

   277     protected function _makeMessages($messages)

   278     {

   279         $messageClass = $this->getMessageClass();

   280         $setClass     = $this->getMessageSetClass();

   281         $result = array();

   282         foreach ($messages as $message) {

   283             $result[] = new $messageClass($message->MessageText, $message);

   284         }

   285         return new $setClass($result);

   286     }

   287 

   288     /**

   289      * Delete the specified message from the specified queue.

   290      * 

   291      * @param  string $queueId

   292      * @param  Zend_Cloud_QueueService_Message $message Message ID or message 

   293      * @param  array  $options

   294      * @return void

   295      */

   296     public function deleteMessage($queueId, $message, $options = null)

   297     {

   298         try {

   299             if ($queueId instanceof Zend_Service_WindowsAzure_Storage_QueueInstance) {

   300                 $queueId = $queueId->Name;

   301             }

   302             if ($message instanceof Zend_Cloud_QueueService_Message) {

   303                 $message = $message->getMessage();

   304             }

   305             if ($message instanceof Zend_Service_WindowsAzure_Storage_QueueMessage) {

   306                 return $this->_storageClient->deleteMessage($queueId, $message);

   307             } else {

   308                 throw new Zend_Cloud_QueueService_Exception('Cannot delete the message: message object required');

   309             }

   310         } catch (Zend_Service_WindowsAzure_Exception $e) {

   311             throw new Zend_Cloud_QueueService_Exception('Error on deleting a message: '.$e->getMessage(), $e->getCode(), $e);

   312         }

   313     }

   314 

   315     /**

   316      * Peek at the messages from the specified queue without removing them.

   317      *

   318      * @param  string $queueId

   319      * @param  int $num How many messages

   320      * @param  array  $options

   321      * @return Zend_Cloud_QueueService_Message[]

   322      */

   323     public function peekMessages($queueId, $num = 1, $options = null)

   324     {

   325         try {

   326             if ($queueId instanceof Zend_Service_WindowsAzure_Storage_QueueInstance) {

   327                 $queueId = $queueId->Name;

   328             }

   329             return $this->_makeMessages($this->_storageClient->peekMessages($queueId, $num));

   330         } catch (Zend_Service_WindowsAzure_Exception $e) {

   331             throw new Zend_Cloud_QueueService_Exception('Error on peeking messages: '.$e->getMessage(), $e->getCode(), $e);

   332         }

   333    }

   334     

   335     /**

   336      * Get Azure implementation

   337      * @return Zend_Service_Azure_Storage_Queue 

   338      */

   339     public function getClient()

   340     {

   341         return $this->_storageClient;

   342     }

   343 }