diff options
Diffstat (limited to 'lib/queuehandler.php')
-rw-r--r-- | lib/queuehandler.php | 126 |
1 files changed, 22 insertions, 104 deletions
diff --git a/lib/queuehandler.php b/lib/queuehandler.php index d5e0150d9..c0f38f4e3 100644 --- a/lib/queuehandler.php +++ b/lib/queuehandler.php @@ -17,26 +17,34 @@ * 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', 0); + class QueueHandler extends Daemon { - var $_id = 'generic'; - function QueueHandler($id=null) + function __construct($id=null, $daemonize=true) { + parent::__construct($daemonize); + if ($id) { $this->set_id($id); } } + function timeout() + { + return 60; + } + function class_name() { return ucfirst($this->transport()) . 'Handler'; @@ -75,101 +83,21 @@ 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() { - require("Stomp.php"); - $con = new Stomp(common_config('queue','stomp_server')); - if (!$con->connect()) { - $this->log(LOG_ERR, 'Failed to connect to queue server'); - return false; - } - $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; - } else { - $this->log(LOG_WARNING, 'queue item for notice that does not exist'); - } - } - } 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(); - } + + $queue = $this->transport(); + $timeout = $this->timeout(); + + $qm = QueueManager::get(); + + $qm->service($queue, $this); + if (!$this->finish()) { return false; } @@ -178,21 +106,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); |