* The queued_task model can now trigger a queueworker exit.

* queued_task model will trigger exit on Datasource error.
closes #6

Author: MSeven

.gitignore

@@ -1,2 +1,2 @@
-
+.svn
 *.bak

models/queued_task.php

@@ -13,6 +13,8 @@ class QueuedTask extends AppModel {
 
 	public $rateHistory = array();
 
+	public $exit = false;
+
 	/**
 	 * Add a new Job to the Queue
 	 *
@@ -32,6 +34,10 @@ public function createJob($jobName, $data, $notBefore = null) {
 		return ($this->save($this->create($data)));
 	}
 
+	public function onError() {
+		$this->exit = true;
+	}
+
 	/**
 	 * Look for a new job that can be processed with the current abilities.
 	 *

vendors/shells/queue.php

@@ -116,35 +116,40 @@ public function add() {
 	public function runworker() {
 		$exit = false;
 		$starttime = time();
+
 		while (!$exit) {
 			$this->out('Looking for Job....');
 			$data = $this->QueuedTask->requestJob($this->getTaskConf());
-			if ($data != false) {
-				$this->out('Running Job of type "' . $data['jobtype'] . '"');
-				$taskname = 'queue_' . strtolower($data['jobtype']);
-				$return = $this->{$taskname}->run(unserialize($data['data']));
-				if ($return == true) {
-					$this->QueuedTask->markJobDone($data['id']);
-					$this->out('Job Finished.');
+			if ($this->QueuedTask->exit === true) {
+				$exit = true;
+			} else {
+				if ($data !== false) {
+					$this->out('Running Job of type "' . $data['jobtype'] . '"');
+					$taskname = 'queue_' . strtolower($data['jobtype']);
+					$return = $this->{$taskname}->run(unserialize($data['data']));
+					if ($return == true) {
+						$this->QueuedTask->markJobDone($data['id']);
+						$this->out('Job Finished.');
+					} else {
+						$this->QueuedTask->markJobFailed($data['id']);
+						$this->out('Job did not finish, requeued.');
+					}
 				} else {
-					$this->QueuedTask->markJobFailed($data['id']);
-					$this->out('Job did not finish, requeued.');
+					$this->out('nothing to do, sleeping.');
+					sleep(Configure::read('queue.sleeptime'));
 				}
-			} else {
-				$this->out('nothing to do, sleeping.');
-				sleep(Configure::read('queue.sleeptime'));
-			}
 
-			// check if we are over the maximum runtime and end processing if so.
-			if (Configure::read('queue.workermaxruntime') != 0 && (time() - $starttime) >= Configure::read('queue.workermaxruntime')) {
-				$exit = true;
-				$this->out('Reached runtime of ' . (time() - $starttime) . ' Seconds (Max ' . Configure::read('queue.workermaxruntime') . '), terminating.');
-			}
-			if ($exit || rand(0, 100) > (100 - Configure::read('queue.gcprop'))) {
-				$this->out('Performing Old job cleanup.');
-				$this->QueuedTask->cleanOldJobs();
+				// check if we are over the maximum runtime and end processing if so.
+				if (Configure::read('queue.workermaxruntime') != 0 && (time() - $starttime) >= Configure::read('queue.workermaxruntime')) {
+					$exit = true;
+					$this->out('Reached runtime of ' . (time() - $starttime) . ' Seconds (Max ' . Configure::read('queue.workermaxruntime') . '), terminating.');
+				}
+				if ($exit || rand(0, 100) > (100 - Configure::read('queue.gcprop'))) {
+					$this->out('Performing Old job cleanup.');
+					$this->QueuedTask->cleanOldJobs();
+				}
+				$this->hr();
 			}
-			$this->hr();
 		}
 	}