lib/stompqueuemanager.php

   1 <?php
   2 /**
   3  * StatusNet, the distributed open-source microblogging tool
   4  *
   5  * Abstract class for queue managers
   6  *
   7  * PHP version 5
   8  *
   9  * LICENCE: This program is free software: you can redistribute it and/or modify
  10  * it under the terms of the GNU Affero General Public License as published by
  11  * the Free Software Foundation, either version 3 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU Affero General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Affero General Public License
  20  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  21  *
  22  * @category  QueueManager
  23  * @package   StatusNet
  24  * @author    Evan Prodromou <evan@status.net>
  25  * @author    Sarven Capadisli <csarven@status.net>
  26  * @copyright 2009 StatusNet, Inc.
  27  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  28  * @link      http://status.net/
  29  */
  30
  31 require_once 'Stomp.php';
  32
  33
  34 class StompQueueManager extends QueueManager
  35 {
  36     var $server = null;
  37     var $username = null;
  38     var $password = null;
  39     var $base = null;
  40     var $con = null;
  41
  42     protected $sites = array();
  43
  44     protected $useTransactions = true;
  45     protected $transaction = null;
  46     protected $transactionCount = 0;
  47
  48     function __construct()
  49     {
  50         parent::__construct();
  51         $this->server   = common_config('queue', 'stomp_server');
  52         $this->username = common_config('queue', 'stomp_username');
  53         $this->password = common_config('queue', 'stomp_password');
  54         $this->base     = common_config('queue', 'queue_basename');
  55     }
  56
  57     /**
  58      * Tell the i/o master we only need a single instance to cover
  59      * all sites running in this process.
  60      */
  61     public static function multiSite()
  62     {
  63         return IoManager::INSTANCE_PER_PROCESS;
  64     }
  65
  66     /**
  67      * Record each site we'll be handling input for in this process,
  68      * so we can listen to the necessary queues for it.
  69      *
  70      * @fixme possibly actually do subscription here to save another
  71      *        loop over all sites later?
  72      * @fixme possibly don't assume it's the current site
  73      */
  74     public function addSite($server)
  75     {
  76         $this->sites[] = $server;
  77         $this->initialize();
  78     }
  79
  80
  81     /**
  82      * Instantiate the appropriate QueueHandler class for the given queue.
  83      *
  84      * @param string $queue
  85      * @return mixed QueueHandler or null
  86      */
  87     function getHandler($queue)
  88     {
  89         $handlers = $this->handlers[common_config('site', 'server')];
  90         if (isset($handlers[$queue])) {
  91             $class = $handlers[$queue];
  92             if (class_exists($class)) {
  93                 return new $class();
  94             } else {
  95                 common_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'");
  96             }
  97         } else {
  98             common_log(LOG_ERR, "Requested handler for unkown queue '$queue'");
  99         }
 100         return null;
 101     }
 102
 103     /**
 104      * Get a list of all registered queue transport names.
 105      *
 106      * @return array of strings
 107      */
 108     function getQueues()
 109     {
 110         $group = $this->activeGroup();
 111         $site = common_config('site', 'server');
 112         if (empty($this->groups[$site][$group])) {
 113             return array();
 114         } else {
 115             return array_keys($this->groups[$site][$group]);
 116         }
 117     }
 118
 119     /**
 120      * Register a queue transport name and handler class for your plugin.
 121      * Only registered transports will be reliably picked up!
 122      *
 123      * @param string $transport
 124      * @param string $class
 125      * @param string $group
 126      */
 127     public function connect($transport, $class, $group='queuedaemon')
 128     {
 129         $this->handlers[common_config('site', 'server')][$transport] = $class;
 130         $this->groups[common_config('site', 'server')][$group][$transport] = $class;
 131     }
 132
 133     /**
 134      * Saves a notice object reference into the queue item table.
 135      * @return boolean true on success
 136      */
 137     public function enqueue($object, $queue)
 138     {
 139         $msg = $this->encode($object);
 140         $rep = $this->logrep($object);
 141
 142         $this->_connect();
 143
 144         // XXX: serialize and send entire notice
 145
 146         $result = $this->con->send($this->queueName($queue),
 147                                    $msg,                // BODY of the message
 148                                    array ('created' => common_sql_now()));
 149
 150         if (!$result) {
 151             common_log(LOG_ERR, "Error sending $rep to $queue queue");
 152             return false;
 153         }
 154
 155         common_log(LOG_DEBUG, "complete remote queueing $rep for $queue");
 156         $this->stats('enqueued', $queue);
 157     }
 158
 159     /**
 160      * Send any sockets we're listening on to the IO manager
 161      * to wait for input.
 162      *
 163      * @return array of resources
 164      */
 165     public function getSockets()
 166     {
 167         return array($this->con->getSocket());
 168     }
 169
 170     /**
 171      * We've got input to handle on our socket!
 172      * Read any waiting Stomp frame(s) and process them.
 173      *
 174      * @param resource $socket
 175      * @return boolean ok on success
 176      */
 177     public function handleInput($socket)
 178     {
 179         assert($socket === $this->con->getSocket());
 180         $ok = true;
 181         $frames = $this->con->readFrames();
 182         foreach ($frames as $frame) {
 183             $ok = $ok && $this->_handleItem($frame);
 184         }
 185         return $ok;
 186     }
 187
 188     /**
 189      * Initialize our connection and subscribe to all the queues
 190      * we're going to need to handle...
 191      *
 192      * Side effects: in multi-site mode, may reset site configuration.
 193      *
 194      * @param IoMaster $master process/event controller
 195      * @return bool return false on failure
 196      */
 197     public function start($master)
 198     {
 199         parent::start($master);
 200         if ($this->sites) {
 201             foreach ($this->sites as $server) {
 202                 StatusNet::init($server);
 203                 $this->doSubscribe();
 204             }
 205         } else {
 206             $this->doSubscribe();
 207         }
 208         $this->begin();
 209         return true;
 210     }
 211
 212     /**
 213      * Subscribe to all the queues we're going to need to handle...
 214      *
 215      * Side effects: in multi-site mode, may reset site configuration.
 216      *
 217      * @return bool return false on failure
 218      */
 219     public function finish()
 220     {
 221         // If there are any outstanding delivered messages we haven't processed,
 222         // free them for another thread to take.
 223         $this->rollback();
 224         if ($this->sites) {
 225             foreach ($this->sites as $server) {
 226                 StatusNet::init($server);
 227                 $this->doUnsubscribe();
 228             }
 229         } else {
 230             $this->doUnsubscribe();
 231         }
 232         return true;
 233     }
 234
 235     /**
 236      * Lazy open connection to Stomp queue server.
 237      */
 238     protected function _connect()
 239     {
 240         if (empty($this->con)) {
 241             $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
 242             $this->con = new LiberalStomp($this->server);
 243
 244             if ($this->con->connect($this->username, $this->password)) {
 245                 $this->_log(LOG_INFO, "Connected.");
 246             } else {
 247                 $this->_log(LOG_ERR, 'Failed to connect to queue server');
 248                 throw new ServerException('Failed to connect to queue server');
 249             }
 250         }
 251     }
 252
 253     /**
 254      * Subscribe to all enabled notice queues for the current site.
 255      */
 256     protected function doSubscribe()
 257     {
 258         $this->_connect();
 259         foreach ($this->getQueues() as $queue) {
 260             $rawqueue = $this->queueName($queue);
 261             $this->_log(LOG_INFO, "Subscribing to $rawqueue");
 262             $this->con->subscribe($rawqueue);
 263         }
 264     }
 265
 266     /**
 267      * Subscribe from all enabled notice queues for the current site.
 268      */
 269     protected function doUnsubscribe()
 270     {
 271         $this->_connect();
 272         foreach ($this->getQueues() as $queue) {
 273             $this->con->unsubscribe($this->queueName($queue));
 274         }
 275     }
 276
 277     /**
 278      * Handle and acknowledge an event that's come in through a queue.
 279      *
 280      * If the queue handler reports failure, the message is requeued for later.
 281      * Missing notices or handler classes will drop the message.
 282      *
 283      * Side effects: in multi-site mode, may reset site configuration to
 284      * match the site that queued the event.
 285      *
 286      * @param StompFrame $frame
 287      * @return bool
 288      */
 289     protected function _handleItem($frame)
 290     {
 291         list($site, $queue) = $this->parseDestination($frame->headers['destination']);
 292         if ($site != common_config('site', 'server')) {
 293             $this->stats('switch');
 294             StatusNet::init($site);
 295         }
 296
 297         $item = $this->decode($frame->body);
 298
 299         $handler = $this->getHandler($queue);
 300         if (!$handler) {
 301             $this->_log(LOG_ERROR, "Missing handler class; skipping $info");
 302             $this->ack($frame);
 303             $this->commit();
 304             $this->begin();
 305             $this->stats('badhandler', $queue);
 306             return false;
 307         }
 308
 309         $ok = $handler->handle($item);
 310
 311         if (!$ok) {
 312             $this->_log(LOG_WARNING, "Failed handling $info");
 313             // FIXME we probably shouldn't have to do
 314             // this kind of queue management ourselves;
 315             // if we don't ack, it should resend...
 316             $this->ack($frame);
 317             $this->enqueue($item, $queue);
 318             $this->commit();
 319             $this->begin();
 320             $this->stats('requeued', $queue);
 321             return false;
 322         }
 323
 324         $this->_log(LOG_INFO, "Successfully handled $info");
 325         $this->ack($frame);
 326         $this->commit();
 327         $this->begin();
 328         $this->stats('handled', $queue);
 329         return true;
 330     }
 331
 332     /**
 333      * Combines the queue_basename from configuration with the
 334      * site server name and queue name to give eg:
 335      *
 336      * /queue/statusnet/identi.ca/sms
 337      *
 338      * @param string $queue
 339      * @return string
 340      */
 341     protected function queueName($queue)
 342     {
 343         return common_config('queue', 'queue_basename') .
 344             common_config('site', 'server') . '/' . $queue;
 345     }
 346
 347     /**
 348      * Returns the site and queue name from the server-side queue.
 349      *
 350      * @param string queue destination (eg '/queue/statusnet/identi.ca/sms')
 351      * @return array of site and queue: ('identi.ca','sms') or false if unrecognized
 352      */
 353     protected function parseDestination($dest)
 354     {
 355         $prefix = common_config('queue', 'queue_basename');
 356         if (substr($dest, 0, strlen($prefix)) == $prefix) {
 357             $rest = substr($dest, strlen($prefix));
 358             return explode("/", $rest, 2);
 359         } else {
 360             common_log(LOG_ERR, "Got a message from unrecognized stomp queue: $dest");
 361             return array(false, false);
 362         }
 363     }
 364
 365     function _log($level, $msg)
 366     {
 367         common_log($level, 'StompQueueManager: '.$msg);
 368     }
 369
 370     protected function begin()
 371     {
 372         if ($this->useTransactions) {
 373             if ($this->transaction) {
 374                 throw new Exception("Tried to start transaction in the middle of a transaction");
 375             }
 376             $this->transactionCount++;
 377             $this->transaction = $this->master->id . '-' . $this->transactionCount . '-' . time();
 378             $this->con->begin($this->transaction);
 379         }
 380     }
 381
 382     protected function ack($frame)
 383     {
 384         if ($this->useTransactions) {
 385             if (!$this->transaction) {
 386                 throw new Exception("Tried to ack but not in a transaction");
 387             }
 388         }
 389         $this->con->ack($frame, $this->transaction);
 390     }
 391
 392     protected function commit()
 393     {
 394         if ($this->useTransactions) {
 395             if (!$this->transaction) {
 396                 throw new Exception("Tried to commit but not in a transaction");
 397             }
 398             $this->con->commit($this->transaction);
 399             $this->transaction = null;
 400         }
 401     }
 402
 403     protected function rollback()
 404     {
 405         if ($this->useTransactions) {
 406             if (!$this->transaction) {
 407                 throw new Exception("Tried to rollback but not in a transaction");
 408             }
 409             $this->con->commit($this->transaction);
 410             $this->transaction = null;
 411         }
 412     }
 413 }
 414