Queue.php 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569
  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-2015 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-2015 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. if (null !== ($name = $this->getOption(self::NAME))) {
  223. $this->_setName($name);
  224. }
  225. return $this;
  226. }
  227. /**
  228. * Get the adapter for this queue
  229. *
  230. * @return Zend_Queue_Adapter_AdapterInterface
  231. */
  232. public function getAdapter()
  233. {
  234. return $this->_adapter;
  235. }
  236. /**
  237. * @param string $className
  238. * @return Zend_Queue Provides a fluent interface
  239. */
  240. public function setMessageClass($className)
  241. {
  242. $this->_messageClass = (string) $className;
  243. return $this;
  244. }
  245. /**
  246. * @return string
  247. */
  248. public function getMessageClass()
  249. {
  250. return $this->_messageClass;
  251. }
  252. /**
  253. * @param string $className
  254. * @return Zend_Queue Provides a fluent interface
  255. */
  256. public function setMessageSetClass($className)
  257. {
  258. $this->_messageSetClass = (string) $className;
  259. return $this;
  260. }
  261. /**
  262. * @return string
  263. */
  264. public function getMessageSetClass()
  265. {
  266. return $this->_messageSetClass;
  267. }
  268. /**
  269. * Get the name of the queue
  270. *
  271. * Note: _setName() used to exist, but it caused confusion with createQueue
  272. * Will evaluate later to see if we should add it back in.
  273. *
  274. * @return string
  275. */
  276. public function getName()
  277. {
  278. return $this->getOption(self::NAME);
  279. }
  280. /**
  281. * Create a new queue
  282. *
  283. * @param string $name queue name
  284. * @param integer $timeout default visibility timeout
  285. * @return Zend_Queue|false
  286. * @throws Zend_Queue_Exception
  287. */
  288. public function createQueue($name, $timeout = null)
  289. {
  290. if (!is_string($name)) {
  291. require_once 'Zend/Queue/Exception.php';
  292. throw new Zend_Queue_Exception('$name is not a string');
  293. }
  294. if ((null !== $timeout) && !is_integer($timeout)) {
  295. require_once 'Zend/Queue/Exception.php';
  296. throw new Zend_Queue_Exception('$timeout must be an integer');
  297. }
  298. // Default to standard timeout
  299. if (null === $timeout) {
  300. $timeout = $this->getOption(self::TIMEOUT);
  301. }
  302. // Some queues allow you to create on the fly, but cannot return
  303. // a list of queues. Stomp protocol for example.
  304. if ($this->isSupported('create')) {
  305. if ($this->getAdapter()->isExists($name)) {
  306. return false;
  307. }
  308. if (!$this->getAdapter()->create($name, $timeout)) {
  309. return false;
  310. }
  311. }
  312. $options = array(
  313. self::NAME => $name,
  314. 'timeout' => $timeout
  315. );
  316. return new self($this->getAdapter(), $options);
  317. }
  318. /**
  319. * Delete the queue this object is working on.
  320. *
  321. * This queue is disabled, regardless of the outcome of the deletion
  322. * of the queue, because the programmers intent is to disable this queue.
  323. *
  324. * @return boolean
  325. */
  326. public function deleteQueue()
  327. {
  328. if ($this->isSupported('delete')) {
  329. $deleted = $this->getAdapter()->delete($this->getName());
  330. }
  331. else {
  332. $deleted = true;
  333. }
  334. /**
  335. * @see Zend_Queue_Adapter_Null
  336. */
  337. require_once('Zend/Queue/Adapter/Null.php');
  338. $this->setAdapter(new Zend_Queue_Adapter_Null($this->getOptions()));
  339. return $deleted;
  340. }
  341. /**
  342. * Delete a message from the queue
  343. *
  344. * Returns true if the message is deleted, false if the deletion is
  345. * unsuccessful.
  346. *
  347. * Returns true if the adapter doesn't support message deletion.
  348. *
  349. * @param Zend_Queue_Message $message
  350. * @return boolean
  351. * @throws Zend_Queue_Exception
  352. */
  353. public function deleteMessage(Zend_Queue_Message $message)
  354. {
  355. if ($this->getAdapter()->isSupported('deleteMessage')) {
  356. return $this->getAdapter()->deleteMessage($message);
  357. }
  358. return true;
  359. }
  360. /**
  361. * Send a message to the queue
  362. *
  363. * @param mixed $message message
  364. * @return Zend_Queue_Message
  365. * @throws Zend_Queue_Exception
  366. */
  367. public function send($message)
  368. {
  369. return $this->getAdapter()->send($message);
  370. }
  371. /**
  372. * Returns the approximate number of messages in the queue
  373. *
  374. * @return integer
  375. */
  376. public function count()
  377. {
  378. if ($this->getAdapter()->isSupported('count')) {
  379. return $this->getAdapter()->count();
  380. }
  381. return 0;
  382. }
  383. /**
  384. * Return the first element in the queue
  385. *
  386. * @param integer $maxMessages
  387. * @param integer $timeout
  388. * @return Zend_Queue_Message_Iterator
  389. */
  390. public function receive($maxMessages=null, $timeout=null)
  391. {
  392. if (($maxMessages !== null) && !is_integer($maxMessages)) {
  393. require_once 'Zend/Queue/Exception.php';
  394. throw new Zend_Queue_Exception('$maxMessages must be an integer or null');
  395. }
  396. if (($timeout !== null) && !is_integer($timeout)) {
  397. require_once 'Zend/Queue/Exception.php';
  398. throw new Zend_Queue_Exception('$timeout must be an integer or null');
  399. }
  400. // Default to returning only one message
  401. if ($maxMessages === null) {
  402. $maxMessages = 1;
  403. }
  404. // Default to standard timeout
  405. if ($timeout === null) {
  406. $timeout = $this->getOption(self::TIMEOUT);
  407. }
  408. return $this->getAdapter()->receive($maxMessages, $timeout);
  409. }
  410. /**
  411. * Return a list of queue capabilities functions
  412. *
  413. * $array['function name'] = true or false
  414. * true is supported, false is not supported.
  415. *
  416. * @param string $name
  417. * @return array
  418. */
  419. public function getCapabilities()
  420. {
  421. return $this->getAdapter()->getCapabilities();
  422. }
  423. /**
  424. * Indicates if a function is supported or not.
  425. *
  426. * @param string $name
  427. * @return boolean
  428. */
  429. public function isSupported($name)
  430. {
  431. $translation = array(
  432. 'deleteQueue' => 'delete',
  433. 'createQueue' => 'create'
  434. );
  435. if (isset($translation[$name])) {
  436. $name = $translation[$name];
  437. }
  438. return $this->getAdapter()->isSupported($name);
  439. }
  440. /**
  441. * Get an array of all available queues
  442. *
  443. * @return array
  444. * @throws Zend_Queue_Exception
  445. */
  446. public function getQueues()
  447. {
  448. if (!$this->isSupported('getQueues')) {
  449. throw new Zend_Queue_Exception( __FUNCTION__ . '() is not supported by ' . get_class($this->getAdapter()));
  450. }
  451. return $this->getAdapter()->getQueues();
  452. }
  453. /**
  454. * Set the name of the queue
  455. *
  456. * This is AN UNSUPPORTED FUNCTION
  457. *
  458. * @param string $name
  459. * @return Zend_Queue|false Provides a fluent interface
  460. */
  461. protected function _setName($name)
  462. {
  463. if (!is_string($name)) {
  464. /**
  465. * @see Zend_Queue_Exception
  466. */
  467. require_once 'Zend/Queue/Exception.php';
  468. throw new Zend_Queue_Exception("$name is not a string");
  469. }
  470. if ($this->getAdapter()->isSupported('create')) {
  471. if (!$this->getAdapter()->isExists($name)) {
  472. $timeout = $this->getOption(self::TIMEOUT);
  473. if (!$this->getAdapter()->create($name, $timeout)) {
  474. // Unable to create the new queue
  475. return false;
  476. }
  477. }
  478. }
  479. $this->setOption(self::NAME, $name);
  480. return $this;
  481. }
  482. /**
  483. * returns a listing of Zend_Queue details.
  484. * useful for debugging
  485. *
  486. * @return array
  487. */
  488. public function debugInfo()
  489. {
  490. $info = array();
  491. $info['self'] = get_class($this);
  492. $info['adapter'] = get_class($this->getAdapter());
  493. foreach ($this->getAdapter()->getCapabilities() as $feature => $supported) {
  494. $info['adapter-' . $feature] = ($supported) ? 'yes' : 'no';
  495. }
  496. $info['options'] = $this->getOptions();
  497. $info['options']['driverOptions'] = '[hidden]';
  498. $info['currentQueue'] = $this->getName();
  499. $info['messageClass'] = $this->getMessageClass();
  500. $info['messageSetClass'] = $this->getMessageSetClass();
  501. return $info;
  502. }
  503. }