diff options
| -rw-r--r-- | lib/queuehandler.php | 151 |
1 files changed, 41 insertions, 110 deletions
diff --git a/lib/queuehandler.php b/lib/queuehandler.php index ae403c65e..045432ae5 100644 --- a/lib/queuehandler.php +++ b/lib/queuehandler.php @@ -17,17 +17,18 @@ * along with this program. If not, see <http://www.gnu.org/licenses/>. */ -define('CLAIM_TIMEOUT', 1200); - if (!defined('LACONICA')) { exit(1); } require_once(INSTALLDIR.'/lib/daemon.php'); require_once(INSTALLDIR.'/classes/Queue_item.php'); require_once(INSTALLDIR.'/classes/Notice.php'); +define('CLAIM_TIMEOUT', 1200); +define('QUEUE_HANDLER_MISS_IDLE', 10); +define('QUEUE_HANDLER_HIT_IDLE', 10); + class QueueHandler extends Daemon { - var $_id = 'generic'; function QueueHandler($id=null) @@ -37,6 +38,11 @@ class QueueHandler extends Daemon } } + function timeout() + { + return null; + } + function class_name() { return ucfirst($this->transport()) . 'Handler'; @@ -75,110 +81,45 @@ class QueueHandler extends Daemon return true; } - function db_dispatch() { - do { - $qi = Queue_item::top($this->transport()); - if ($qi) { - $this->log(LOG_INFO, 'Got item enqueued '.common_exact_date($qi->created)); - $notice = Notice::staticGet($qi->notice_id); - if ($notice) { - $this->log(LOG_INFO, 'broadcasting notice ID = ' . $notice->id); - # XXX: what to do if broadcast fails? - $result = $this->handle_notice($notice); - if (!$result) { - $this->log(LOG_WARNING, 'Failed broadcast for notice ID = ' . $notice->id); - $orig = $qi; - $qi->claimed = null; - $qi->update($orig); - $this->log(LOG_WARNING, 'Abandoned claim for notice ID = ' . $notice->id); - continue; - } - $this->log(LOG_INFO, 'finished broadcasting notice ID = ' . $notice->id); - $notice->free(); - unset($notice); - $notice = null; - } else { - $this->log(LOG_WARNING, 'queue item for notice that does not exist'); - } - $qi->delete(); - $qi->free(); - unset($qi); - $this->idle(0); - } else { - $this->clear_old_claims(); - $this->idle(5); - } - } while (true); - } - - function stomp_dispatch() { - - // use an external message queue system via STOMP - require_once("Stomp.php"); + function run() + { + if (!$this->start()) { + return false; + } - $server = common_config('queue','stomp_server'); - $username = common_config('queue', 'stomp_username'); - $password = common_config('queue', 'stomp_password'); + $this->log(LOG_INFO, 'checking for queued notices'); - $con = new Stomp($server); + $queue = $this->transport(); + $timeout = $this->timeout(); - if (!$con->connect($username, $password)) { - $this->log(LOG_ERR, 'Failed to connect to queue server'); - return false; - } + $qm = QueueManager::get(); - $queue_basename = common_config('queue','queue_basename'); - // subscribe to the relevant queue (format: basename-transport) - $con->subscribe('/queue/'.$queue_basename.'-'.$this->transport()); - - do { - $frame = $con->readFrame(); - if ($frame) { - $this->log(LOG_INFO, 'Got item enqueued '.common_exact_date($frame->headers['created'])); - - // XXX: Now the queue handler receives only the ID of the - // notice, and it has to get it from the DB - // A massive improvement would be avoid DB query by transmitting - // all the notice details via queue server... - $notice = Notice::staticGet($frame->body); - - if ($notice) { - $this->log(LOG_INFO, 'broadcasting notice ID = ' . $notice->id); - $result = $this->handle_notice($notice); - if ($result) { - // if the msg has been handled positively, ack it - // and the queue server will remove it from the queue - $con->ack($frame); - $this->log(LOG_INFO, 'finished broadcasting notice ID = ' . $notice->id); - } - else { - // no ack - $this->log(LOG_WARNING, 'Failed broadcast for notice ID = ' . $notice->id); - } - $notice->free(); - unset($notice); - $notice = null; + while (true) { + $this->log(LOG_DEBUG, 'Checking for notices...'); + $notice = $qm->nextItem($queue, $timeout); + if (empty($notice)) { + $this->log(LOG_DEBUG, 'No notices waiting; idling.'); + // Nothing in the queue. Do you + // have other tasks, like servicing your + // XMPP connection, to do? + $this->idle(QUEUE_HANDLER_MISS_IDLE); + } else { + $this->log(LOG_INFO, 'Got notice '. $notice->id); + // Yay! Got one! + if ($this->handle_notice($notice)) { + $this->log(LOG_INFO, 'Successfully handled notice '. $notice->id); + $qm->done($notice, $queue); } else { - $this->log(LOG_WARNING, 'queue item for notice that does not exist'); + $this->log(LOG_INFO, 'Failed to handle notice '. $notice->id); + $qm->fail($notice, $queue); } + // Chance to e.g. service your XMPP connection + $this->log(LOG_DEBUG, 'Idling after success.'); + $this->idle(QUEUE_HANDLER_HIT_IDLE); } - } while (true); - - $con->disconnect(); - } - - function run() - { - if (!$this->start()) { - return false; - } - $this->log(LOG_INFO, 'checking for queued notices'); - if (common_config('queue','subsystem') == 'stomp') { - $this->stomp_dispatch(); - } - else { - $this->db_dispatch(); + // XXX: when do we give up? } + if (!$this->finish()) { return false; } @@ -187,21 +128,11 @@ class QueueHandler extends Daemon function idle($timeout=0) { - if ($timeout>0) { + if ($timeout > 0) { sleep($timeout); } } - function clear_old_claims() - { - $qi = new Queue_item(); - $qi->transport = $this->transport(); - $qi->whereAdd('now() - claimed > '.CLAIM_TIMEOUT); - $qi->update(DB_DATAOBJECT_WHEREADD_ONLY); - $qi->free(); - unset($qi); - } - function log($level, $msg) { common_log($level, $this->class_name() . ' ('. $this->get_id() .'): '.$msg); |