Queue.php 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571
  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 license@zend.com so we can send you a copy immediately.
  14. *
  15. * @category Zend
  16. * @package Zend_Queue
  17. * @copyright Copyright (c) 2005-2009 Zend Technologies USA Inc. (http://www.zend.com)
  18. * @license http://framework.zend.com/license/new-bsd New BSD License
  19. * @version $Id$
  20. */
  21. /**
  22. * Class for connecting to queues performing common operations.
  23. *
  24. * @category Zend
  25. * @package Zend_Queue
  26. * @copyright Copyright (c) 2005-2009 Zend Technologies USA Inc. (http://www.zend.com)
  27. * @license http://framework.zend.com/license/new-bsd New BSD License
  28. */
  29. class Zend_Queue implements Countable
  30. {
  31. /**
  32. * Use the TIMEOUT constant in the config of a Zend_Queue
  33. */
  34. const TIMEOUT = 'timeout';
  35. /**
  36. * Default visibility passed to count
  37. */
  38. const VISIBILITY_TIMEOUT = 30;
  39. /**
  40. * Use the NAME constant in the config of Zend_Queue
  41. */
  42. const NAME = 'name';
  43. /**
  44. * @var Zend_Queue_Adapter_AdapterInterface
  45. */
  46. protected $_adapter = null;
  47. /**
  48. * User-provided configuration
  49. *
  50. * @var array
  51. */
  52. protected $_options = array();
  53. /**
  54. * Zend_Queue_Message class
  55. *
  56. * @var string
  57. */
  58. protected $_messageClass = 'Zend_Queue_Message';
  59. /**
  60. * Zend_Queue_Message_Iterator class
  61. *
  62. * @var string
  63. */
  64. protected $_messageSetClass = 'Zend_Queue_Message_Iterator';
  65. /**
  66. * @var Zend_Log
  67. */
  68. protected $_logger = null;
  69. /**
  70. * Constructor
  71. *
  72. * Can be called as
  73. * $queue = new Zend_Queue($config);
  74. * - or -
  75. * $queue = new Zend_Queue('array', $config);
  76. * - or -
  77. * $queue = new Zend_Queue(null, $config); // Zend_Queue->createQueue();
  78. *
  79. * @param string|Zend_Queue_Adapter|array|Zend_Config|null String or adapter instance, or options array or Zend_Config instance
  80. * @param Zend_Config|array $options Zend_Config or a configuration array
  81. * @return void
  82. */
  83. public function __construct($spec, $options = array())
  84. {
  85. $adapter = null;
  86. if ($spec instanceof Zend_Queue_Adapter_AdapterInterface) {
  87. $adapter = $spec;
  88. } elseif (is_string($spec)) {
  89. $adapter = $spec;
  90. } elseif ($spec instanceof Zend_Config) {
  91. $options = $spec->toArray();
  92. } elseif (is_array($spec)) {
  93. $options = $spec;
  94. }
  95. // last minute error checking
  96. if ((null === $adapter)
  97. && (!is_array($options) && (!$options instanceof Zend_Config))
  98. ) {
  99. require_once 'Zend/Queue/Exception.php';
  100. throw new Zend_Queue_Exception('No valid params passed to constructor');
  101. }
  102. // Now continue as we would if we were a normal constructor
  103. if ($options instanceof Zend_Config) {
  104. $options = $options->toArray();
  105. } elseif (!is_array($options)) {
  106. $options = array();
  107. }
  108. // Make sure we have some defaults to work with
  109. if (!isset($options[self::TIMEOUT])) {
  110. $options[self::TIMEOUT] = self::VISIBILITY_TIMEOUT;
  111. }
  112. // Make sure all defaults are appropriately set.
  113. if (!array_key_exists('timeout', $options)) {
  114. $options[self::TIMEOUT] = self::VISIBILITY_TIMEOUT;
  115. }
  116. if (array_key_exists('messageClass', $options)) {
  117. $this->setMessageClass($options['messageClass']);
  118. }
  119. if (array_key_exists('messageSetClass', $options)) {
  120. $this->setMessageSetClass($options['messageSetClass']);
  121. }
  122. $this->setOptions($options);
  123. // if we were passed an adapter we either build the $adapter or use it
  124. if (null !== $adapter) {
  125. $this->setAdapter($adapter);
  126. }
  127. }
  128. /**
  129. * Set queue options
  130. *
  131. * @param array $options
  132. * @return Zend_Queue
  133. */
  134. public function setOptions(array $options)
  135. {
  136. $this->_options = array_merge($this->_options, $options);
  137. return $this;
  138. }
  139. /**
  140. * Set an individual configuration option
  141. *
  142. * @param string $name
  143. * @param mixed $value
  144. * @return Zend_Queue
  145. */
  146. public function setOption($name, $value)
  147. {
  148. $this->_options[(string) $name] = $value;
  149. return $this;
  150. }
  151. /**
  152. * Returns the configuration options for the queue
  153. *
  154. * @return array
  155. */
  156. public function getOptions()
  157. {
  158. return $this->_options;
  159. }
  160. /**
  161. * Determine if a requested option has been defined
  162. *
  163. * @param string $name
  164. * @return bool
  165. */
  166. public function hasOption($name)
  167. {
  168. return array_key_exists($name, $this->_options);
  169. }
  170. /**
  171. * Retrieve a single option
  172. *
  173. * @param string $name
  174. * @return null|mixed Returns null if option does not exist; option value otherwise
  175. */
  176. public function getOption($name)
  177. {
  178. if ($this->hasOption($name)) {
  179. return $this->_options[$name];
  180. }
  181. return null;
  182. }
  183. /**
  184. * Set the adapter for this queue
  185. *
  186. * @param string|Zend_Queue_Adapter_AdapterInterface $adapter
  187. * @return Zend_Queue Provides a fluent interface
  188. */
  189. public function setAdapter($adapter)
  190. {
  191. if (is_string($adapter)) {
  192. if (null === ($adapterNamespace = $this->getOption('adapterNamespace'))) {
  193. $adapterNamespace = 'Zend_Queue_Adapter';
  194. }
  195. $adapterName = str_replace(
  196. ' ',
  197. '_',
  198. ucwords(
  199. str_replace(
  200. '_',
  201. ' ',
  202. strtolower($adapterNamespace . '_' . $adapter)
  203. )
  204. )
  205. );
  206. if (!class_exists($adapterName)) {
  207. require_once 'Zend/Loader.php';
  208. Zend_Loader::loadClass($adapterName);
  209. }
  210. /*
  211. * Create an instance of the adapter class.
  212. * Pass the configuration to the adapter class constructor.
  213. */
  214. $adapter = new $adapterName($this->getOptions(), $this);
  215. }
  216. if (!$adapter instanceof Zend_Queue_Adapter_AdapterInterface) {
  217. require_once 'Zend/Queue/Exception.php';
  218. throw new Zend_Queue_Exception("Adapter class '" . get_class($adapterName) . "' does not implement Zend_Queue_Adapter_AdapterInterface");
  219. }
  220. $this->_adapter = $adapter;
  221. $this->_adapter->setQueue($this);
  222. $this->_setName($this->getOption(self::NAME));
  223. return $this;
  224. }
  225. /**
  226. * Get the adapter for this queue
  227. *
  228. * @return Zend_Queue_Adapter_AdapterInterface
  229. */
  230. public function getAdapter()
  231. {
  232. return $this->_adapter;
  233. }
  234. /**
  235. * @param string $className
  236. * @return Zend_Queue Provides a fluent interface
  237. */
  238. public function setMessageClass($className)
  239. {
  240. $this->_messageClass = (string) $className;
  241. return $this;
  242. }
  243. /**
  244. * @return string
  245. */
  246. public function getMessageClass()
  247. {
  248. return $this->_messageClass;
  249. }
  250. /**
  251. * @param string $className
  252. * @return Zend_Queue Provides a fluent interface
  253. */
  254. public function setMessageSetClass($className)
  255. {
  256. $this->_messageSetClass = (string) $className;
  257. return $this;
  258. }
  259. /**
  260. * @return string
  261. */
  262. public function getMessageSetClass()
  263. {
  264. return $this->_messageSetClass;
  265. }
  266. /**
  267. * Get the name of the queue
  268. *
  269. * Note: _setName() used to exist, but it caused confusion with createQueue
  270. * Will evaluate later to see if we should add it back in.
  271. *
  272. * @return string
  273. */
  274. public function getName()
  275. {
  276. return $this->getOption(self::NAME);
  277. }
  278. /**
  279. * Create a new queue
  280. *
  281. * @param string $name queue name
  282. * @param integer $timeout default visibility timeout
  283. * @return Zend_Queue|false
  284. * @throws Zend_Queue_Exception
  285. */
  286. public function createQueue($name, $timeout = null)
  287. {
  288. if (!is_string($name)) {
  289. require_once 'Zend/Queue/Exception.php';
  290. throw new Zend_Queue_Exception('$name is not a string');
  291. }
  292. if ((null !== $timeout) && !is_integer($timeout)) {
  293. require_once 'Zend/Queue/Exception.php';
  294. throw new Zend_Queue_Exception('$timeout must be an integer');
  295. }
  296. // Default to standard timeout
  297. if (null === $timeout) {
  298. $timeout = $this->getOption(self::TIMEOUT);
  299. }
  300. // Some queues allow you to create on the fly, but cannot return
  301. // a list of queues. Stomp protocol for example.
  302. if ($this->isSupported('create')) {
  303. if ($this->getAdapter()->isExists($name)) {
  304. return false;
  305. }
  306. if (!$this->getAdapter()->create($name, $timeout)) {
  307. return false;
  308. }
  309. }
  310. $options = array(
  311. self::NAME => $name,
  312. 'timeout' => $timeout
  313. );
  314. return new self($this->getAdapter(), $options);
  315. }
  316. /**
  317. * Delete the queue this object is working on.
  318. *
  319. * This queue is disabled, regardless of the outcome of the deletion
  320. * of the queue, because the programmers intent is to disable this queue.
  321. *
  322. * @return boolean
  323. */
  324. public function deleteQueue()
  325. {
  326. if ($this->isSupported('delete')) {
  327. $deleted = $this->getAdapter()->delete($this->getName());
  328. }
  329. else {
  330. $deleted = true;
  331. }
  332. /**
  333. * @see Zend_Queue_Adapter_Null
  334. */
  335. require_once('Zend/Queue/Adapter/Null.php');
  336. $this->setAdapter(new Zend_Queue_Adapter_Null($this->getOptions()));
  337. return $deleted;
  338. }
  339. /**
  340. * Delete a message from the queue
  341. *
  342. * Returns true if the message is deleted, false if the deletion is
  343. * unsuccessful.
  344. *
  345. * Returns true if the adapter doesn't support message deletion.
  346. *
  347. * @param Zend_Queue_Message $message
  348. * @return boolean
  349. * @throws Zend_Queue_Exception
  350. */
  351. public function deleteMessage(Zend_Queue_Message $message)
  352. {
  353. if ($this->getAdapter()->isSupported('deleteMessage')) {
  354. return $this->getAdapter()->deleteMessage($message);
  355. }
  356. return true;
  357. }
  358. /**
  359. * Send a message to the queue
  360. *
  361. * @param string $message message
  362. * @return Zend_Queue_Message
  363. * @throws Zend_Queue_Exception
  364. */
  365. public function send($message)
  366. {
  367. if (!is_string($message)) {
  368. require_once 'Zend/Queue/Exception.php';
  369. throw new Zend_Queue_Exception('$message is not a string');
  370. }
  371. return $this->getAdapter()->send($message);
  372. }
  373. /**
  374. * Returns the approximate number of messages in the queue
  375. *
  376. * @return integer
  377. */
  378. public function count()
  379. {
  380. if ($this->getAdapter()->isSupported('count')) {
  381. return $this->getAdapter()->count();
  382. }
  383. return 0;
  384. }
  385. /**
  386. * Return the first element in the queue
  387. *
  388. * @param integer $maxMessages
  389. * @param integer $timeout
  390. * @return Zend_Queue_Message_Iterator
  391. */
  392. public function receive($maxMessages=null, $timeout=null)
  393. {
  394. if (($maxMessages !== null) && !is_integer($maxMessages)) {
  395. require_once 'Zend/Queue/Exception.php';
  396. throw new Zend_Queue_Exception('$maxMessages must be an integer or null');
  397. }
  398. if (($timeout !== null) && !is_integer($timeout)) {
  399. require_once 'Zend/Queue/Exception.php';
  400. throw new Zend_Queue_Exception('$timeout must be an integer or null');
  401. }
  402. // Default to returning only one message
  403. if ($maxMessages === null) {
  404. $maxMessages = 1;
  405. }
  406. // Default to standard timeout
  407. if ($timeout === null) {
  408. $timeout = $this->getOption(self::TIMEOUT);
  409. }
  410. return $this->getAdapter()->receive($maxMessages, $timeout);
  411. }
  412. /**
  413. * Return a list of queue capabilities functions
  414. *
  415. * $array['function name'] = true or false
  416. * true is supported, false is not supported.
  417. *
  418. * @param string $name
  419. * @return array
  420. */
  421. public function getCapabilities()
  422. {
  423. return $this->getAdapter()->getCapabilities();
  424. }
  425. /**
  426. * Indicates if a function is supported or not.
  427. *
  428. * @param string $name
  429. * @return boolean
  430. */
  431. public function isSupported($name)
  432. {
  433. $translation = array(
  434. 'deleteQueue' => 'delete',
  435. 'createQueue' => 'create'
  436. );
  437. if (isset($translation[$name])) {
  438. $name = $translation[$name];
  439. }
  440. return $this->getAdapter()->isSupported($name);
  441. }
  442. /**
  443. * Get an array of all available queues
  444. *
  445. * @return array
  446. * @throws Zend_Queue_Exception
  447. */
  448. public function getQueues()
  449. {
  450. if (!$this->isSupported('getQueues')) {
  451. throw new Zend_Queue_Exception( __FUNCTION__ . '() is not supported by ' . get_class($this->getAdapter()));
  452. }
  453. return $this->getAdapter()->getQueues();
  454. }
  455. /**
  456. * Set the name of the queue
  457. *
  458. * This is AN UNSUPPORTED FUNCTION
  459. *
  460. * @param string $name
  461. * @return Zend_Queue|false Provides a fluent interface
  462. */
  463. protected function _setName($name)
  464. {
  465. if (!is_string($name)) {
  466. /**
  467. * @see Zend_Queue_Exception
  468. */
  469. require_once 'Zend/Queue/Exception.php';
  470. throw new Zend_Queue_Exception("$name is not a string");
  471. }
  472. if ($this->getAdapter()->isSupported('create')) {
  473. if (!$this->getAdapter()->isExists($name)) {
  474. $timeout = $this->getOption(self::TIMEOUT);
  475. if (!$this->getAdapter()->create($name, $timeout)) {
  476. // Unable to create the new queue
  477. return false;
  478. }
  479. }
  480. }
  481. $this->setOption(self::NAME, $name);
  482. return $this;
  483. }
  484. /**
  485. * returns a listing of Zend_Queue details.
  486. * useful for debugging
  487. *
  488. * @return array
  489. */
  490. public function debugInfo()
  491. {
  492. $info = array();
  493. $info['self'] = get_class($this);
  494. $info['adapter'] = get_class($this->getAdapter());
  495. foreach ($this->getAdapter()->getCapabilities() as $feature => $supported) {
  496. $info['adapter-' . $feature] = ($supported) ? 'yes' : 'no';
  497. }
  498. $info['options'] = $this->getOptions();
  499. $info['options']['driverOptions'] = '[hidden]';
  500. $info['currentQueue'] = $this->getName();
  501. $info['messageClass'] = $this->getMessageClass();
  502. $info['messageSetClass'] = $this->getMessageSetClass();
  503. return $info;
  504. }
  505. }