X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=lib%2Fqueuehandler.php;h=2909cd83b100656efd354063323610590191d152;hb=4b4b763255ad3b2bff8f18da2bd3927b52a54e55;hp=c1c4f3309a24011b60e72de070a5fcef0267ad56;hpb=b10dba27982a94d73fb7f8d70cae611241871706;p=quix0rs-gnu-social.git diff --git a/lib/queuehandler.php b/lib/queuehandler.php index c1c4f3309a..2909cd83b1 100644 --- a/lib/queuehandler.php +++ b/lib/queuehandler.php @@ -1,7 +1,7 @@ . */ -define('CLAIM_TIMEOUT', 1200); +if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); } -if (!defined('LACONICA')) { exit(1); } - -require_once(INSTALLDIR.'/lib/daemon.php'); -require_once(INSTALLDIR.'/classes/Queue_item.php'); -require_once(INSTALLDIR.'/classes/Notice.php'); - -class QueueHandler extends Daemon +/** + * Base class for queue handlers. + * + * As of 0.9, queue handlers are short-lived for items as they are + * dequeued by a QueueManager running in an IoMaster in a daemon + * such as queuedaemon.php. + * + * Extensions requiring long-running maintenance or polling should + * register an IoManager. + * + * Subclasses must override at least the following methods: + * - transport + * - handle + */ +class QueueHandler { - var $_id = 'generic'; - - function __construct($id=null, $daemonize=true) - { - parent::__construct($daemonize); - - if ($id) { - $this->set_id($id); - } - } - - function class_name() - { - return ucfirst($this->transport()) . 'Handler'; - } - - function name() - { - return strtolower($this->class_name().'.'.$this->get_id()); - } - - function get_id() - { - return $this->_id; - } - - function set_id($id) - { - $this->_id = $id; - } + /** + * Return transport keyword which identifies items this queue handler + * services; must be defined for all subclasses. + * + * Must be 8 characters or less to fit in the queue_item database. + * ex "email", "jabber", "sms", "irc", ... + * + * @return string + */ function transport() { return null; } - function start() - { - } - - function finish() - { - } - - function handle_notice($notice) + /** + * Here's the meat of your queue handler -- you're handed a Notice + * or other object, which you may do as you will with. + * + * If this function indicates failure, a warning will be logged + * and the item is placed back in the queue to be re-run. + * + * @param mixed $object + * @return boolean true on success, false on failure + */ + function handle($object) { 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"); - - $server = common_config('queue','stomp_server'); - $username = common_config('queue', 'stomp_username'); - $password = common_config('queue', 'stomp_password'); - - $con = new Stomp($server); - - if (!$con->connect($username, $password)) { - $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(); - } - if (!$this->finish()) { - return false; - } - return true; - } - - function idle($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); - } }