Index: beanstalkd.install
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/beanstalkd/Attic/beanstalkd.install,v
retrieving revision 1.1.2.2
diff -u -r1.1.2.2 beanstalkd.install
--- beanstalkd.install 1 Jan 2010 05:55:09 -0000 1.1.2.2
+++ beanstalkd.install 4 Jan 2011 03:53:02 -0000
@@ -15,16 +15,47 @@
if ($phase == 'runtime') {
drupal_queue_include();
-
- $queue = new BeanstalkdQueue(NULL);
-
- $requirements['beanstalkd_queue_count'] = array(
- 'title' => $t('Beanstalkd queue count'),
- 'value' => $queue->numberOfItems(),
- 'severity' => REQUIREMENT_INFO,
- 'description' => $t('Provides a count of all the number of items that are currently ready to be processed on the queue'),
+
+ $path = beanstalkd_pheanstalk_get_path();
+ if (!$path) {
+ $config = conf_path();
+ $profile = variable_get('install_profile', 'default');
+ $searchdir = array(
+ "profile/$profile/libraries",
+ 'sites/all/libraries',
+ "$config/libraries",
+ );
+ }
+ else {
+ $searchdir = array();
+ }
+ $requirements['beanstalkd_pheanstalk_path'] = array(
+ 'title' => $t('Beanstalkd Pheanstalk path'),
+ 'value' => $path ? $path : $t('Not found. Please ensure that Phpeanstalk is installed in one of the following paths; !paths', array('!paths' => '"' . implode('/pheanstalk", "', $searchdir) . '/pheanstalk"')),
+ 'severity' => $path ? REQUIREMENT_INFO : REQUIREMENT_ERROR,
+ 'description' => $t('Pheanstalk is a pure PHP 5.2+ client for the beanstalkd workqueue and provides the connection class to allow Drupal to talk to the beanstalkd daemon. See https://github.com/pda/pheanstalk for more information about Pheanstalk.'),
);
- }
+
+ if (class_exists('BeanstalkdQueue')) {
+ $queue = new BeanstalkdQueue(NULL);
+ $requirements['beanstalkd_queue_count'] = array(
+ 'title' => $t('Beanstalkd Queue count'),
+ 'value' => $queue->numberOfItems(),
+ 'severity' => REQUIREMENT_INFO,
+ 'description' => $t('Provides a count of all the number of items that are currently ready to be processed on the queue'),
+ );
+
+ if ($e = $queue->getError()) {
+ $requirements['beanstalkd_last_error'] = array(
+ 'title' => $t('Beanstalkd Queue count error'),
+ 'value' => $t('%message in %file on line %line.', array('%error' => $e->getCode(), '%message' => $e->getMessage(), '%file' => $e->getFile(), '%line' => $e->getLine())),
+ 'severity' => REQUIREMENT_ERROR,
+ 'description' => $t('An error has occurred while getting the current number of in the queue.'),
+ );
+ }
+ }
+ }
+
return $requirements;
}
\ No newline at end of file
Index: beanstalkd.module
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/beanstalkd/beanstalkd.module,v
retrieving revision 1.1.2.2
diff -u -r1.1.2.2 beanstalkd.module
--- beanstalkd.module 8 Dec 2009 11:49:58 -0000 1.1.2.2
+++ beanstalkd.module 4 Jan 2011 03:53:03 -0000
@@ -29,4 +29,69 @@
*/
function beanstalkd_drupal_queue_load_classes() {
module_load_include('inc', 'beanstalkd', 'beanstalkd.queue');
+}
+
+/**
+ * Get Pheanstalk path
+ */
+function beanstalkd_pheanstalk_get_path() {
+ static $path = NULL;
+
+ if (isset($path)) {
+ return $path;
+ }
+
+ $config = conf_path();
+ $profile = variable_get('install_profile', 'default');
+ $searchdir = array();
+
+ if (file_exists("profile/$profile/libraries")) {
+ $searchdir[] = "profile/$profile/libraries";
+ }
+
+ $searchdir[] = 'sites/all/libraries';
+
+ if (file_exists("$config/libraries")) {
+ $searchdir[] = "$config/libraries";
+ }
+
+ foreach ($searchdir as $dir) {
+ if (file_exists("$dir/pheanstalk/classes/Pheanstalk/ClassLoader.php")) {
+ $path = "$dir/pheanstalk";
+ return $path;
+ }
+ }
+
+ return FALSE;
+}
+
+/**
+ * Load Pheanstalk
+ */
+function beanstalkd_load_pheanstalk() {
+ if (!class_exists('Pheanstalk_ClassLoader') && ($path = beanstalkd_pheanstalk_get_path())) {
+ include_once "$path/classes/Pheanstalk/ClassLoader.php";
+
+ Pheanstalk_ClassLoader::register("$path/classes");
+ }
+
+ return class_exists('Pheanstalk_ClassLoader');
+}
+
+/**
+ * Get Queue Parameters
+ */
+function beanstalkd_get_queue_options($name) {
+ static $options = array();
+
+ if (!isset($options[$name])) {
+ $options[$name] = variable_get('beanstalk_queue_' . $name, array());
+ $options[$name] += array(
+ 'host' => variable_get('beanstalkd_host', 'localhost'),
+ 'port' => variable_get('beanstalkd_port', Pheanstalk::DEFAULT_PORT),
+ 'fork' => FALSE,
+ );
+ }
+
+ return $options[$name];
}
\ No newline at end of file
Index: beanstalkd.queue.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/beanstalkd/beanstalkd.queue.inc,v
retrieving revision 1.6.2.2
diff -u -r1.6.2.2 beanstalkd.queue.inc
--- beanstalkd.queue.inc 1 Jan 2010 05:55:09 -0000 1.6.2.2
+++ beanstalkd.queue.inc 4 Jan 2011 03:53:03 -0000
@@ -5,7 +5,7 @@
* @file
*/
-class BeanstalkdQueue implements DrupalQueueInterface {
+class BeanstalkdQueue implements DrupalReliableQueueInterface {
/**
*
*/
@@ -23,22 +23,35 @@
*/
public function __construct($name) {
$this->tube = $name;
- if (file_exists(drupal_get_path('module', 'beanstalkd') . '/pheanstalk/classes/Pheanstalk/ClassLoader.php')) {
- module_load_include('php', 'beanstalkd', 'pheanstalk/classes/Pheanstalk/ClassLoader');
- Pheanstalk_ClassLoader::register(drupal_get_path('module', 'beanstalkd') . '/pheanstalk/classes');
+ if (beanstalkd_load_pheanstalk()) {
+ $this->beanstalkd_params = beanstalkd_get_queue_options($name);
- $this->beanstalkd_queue = new Pheanstalk(variable_get('beanstalkd_host', 'localhost'), variable_get('beanstalkd_port', Pheanstalk::DEFAULT_PORT));
- if ($name) {
- // If a queue name is past then set this tube to be used and set it to be the
- // only tube to be watched.
- $tube = $this->_tubeName($name);
- $this->beanstalkd_queue
- ->useTube($tube)
- ->watch($tube)
- ->ignore('default');
+ try {
+ $this->beanstalkd_queue = new Pheanstalk($this->beanstalkd_params['host'], $this->beanstalkd_params['port']);
+ if ($name) {
+ // If a queue name is past then set this tube to be used and set it to be the
+ // only tube to be watched.
+ $tube = $this->_tubeName($name);
+ $this->beanstalkd_queue
+ ->useTube($tube)
+ ->watch($tube)
+ ->ignore('default');
+ }
+ else {
+ // be sure to establish the connection so that we can catch any
+ // errors
+ $this->beanstalkd_queue
+ ->stats();
+ }
+ }
+ catch (Exception $e) {
+ $this->beanstalkd_queue = FALSE;
+ $this->lastError = $e;
+ watchdog('beanstalk', '%message in %file on line %line.', array('%error' => $e->getCode(), '%message' => $e->getMessage(), '%file' => $e->getFile(), '%line' => $e->getLine()), WATCHDOG_ERROR);
}
}
else {
+ $this->beanstalkd_params = array();
$this->beanstalkd_queue = FALSE;
}
}
@@ -55,6 +68,10 @@
* far as we know, the item is now in the queue.
*/
public function createItem($data) {
+ if (!$this->beanstalkd_queue) {
+ return FALSE;
+ }
+
$record = new stdClass();
$record->name = $this->tube;
$record->data = $data;
@@ -78,6 +95,10 @@
* An integer estimate of the number of items in the queue.
*/
public function numberOfItems() {
+ if (!$this->beanstalkd_queue) {
+ return;
+ }
+
if ($this->tube) {
$stats = $this->beanstalkd_queue->statsTube($this->_tubeName($this->tube));
}
@@ -105,8 +126,12 @@
* and either the queue is empty or there is some other non-recoverable
* problem.
*/
- public function claimItem($lease_time = 3600) {
- $job = $this->beanstalkd_queue->reserve(0);
+ public function claimItem($lease_time = 3600, $timeout = 0) {
+ if (!$this->beanstalkd_queue) {
+ return FALSE;
+ }
+
+ $job = $this->beanstalkd_queue->reserve($timeout);
if ($job) {
$item = unserialize($job->getData());
$item->id = $job->getId();
@@ -119,16 +144,11 @@
/**
* Claim the next item on any of the tubes which are being watched. Since this is a blocking
* method it will not return until and item is claimed.
+ *
+ * This method has been depreciated. Use $this->claimItem(3600, NULL)
*/
public function claimItemBlocking() {
- $job = $this->beanstalkd_queue->reserve();
- if ($job) {
- $item = unserialize($job->getData());
- $item->id = $job->getId();
- $item->beanstalkd_job = $job;
- return $item;
- }
- return FALSE;
+ return $this->claimItem(3600, NULL);
}
/**
@@ -138,6 +158,10 @@
* The item returned by DrupalQueueInterface::claimItem().
*/
public function deleteItem($item) {
+ if (!$this->beanstalkd_queue) {
+ return;
+ }
+
$this->beanstalkd_queue->delete($item->beanstalkd_job);
}
@@ -157,16 +181,34 @@
}
/**
- * Delete a queue and every item in the queue.
+ * Delete a finished item from the queue.
+ *
+ * @param $item
+ * The item returned by DrupalQueueInterface::claimItem().
*/
public function deleteQueue() {
}
/**
+ * Release an item that the worker could not process, so another
+ * worker can come in and process it before the timeout expires.
+ *
+ * @param $item
+ * @return boolean
+ */
+ public function releaseItem($item) {
+
+ }
+
+ /**
* watch message queue
*/
public function watch($names) {
+ if (!$this->beanstalkd_queue) {
+ return;
+ }
+
$names = is_string($names) ? array($names) : $names;
$names = array_map(array($this, '_tubeName'), $names);
@@ -182,6 +224,10 @@
* Queues to ignore
*/
public function ignore($names) {
+ if (!$this->beanstalkd_queue) {
+ return;
+ }
+
$names = is_string($names) ? array($names) : $names;
$names = array_map(array($this, '_tubeName'), $names);
@@ -197,6 +243,10 @@
* Get a job from the Queue
*/
public function peekItem($job_id) {
+ if (!$this->beanstalkd_queue) {
+ return;
+ }
+
$job = $this->beanstalkd_queue->peek($job_id);
if ($job) {
@@ -214,6 +264,12 @@
return FALSE;
}
+ public function getError() {
+ if (isset($this->lastError)) {
+ return $this->lastError;
+ }
+ }
+
private function _tubeName($name) {
return str_replace('_', '-', $name);
}