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 require_once 'Stomp/Exception.php';
  33
  34
  35 class StompQueueManager extends QueueManager
  36 {
  37     protected $servers;
  38     protected $username;
  39     protected $password;
  40     protected $base;
  41     protected $control;
  42
  43     protected $useTransactions = true;
  44
  45     protected $sites = array();
  46     protected $subscriptions = array();
  47
  48     protected $cons = array(); // all open connections
  49     protected $disconnect = array();
  50     protected $transaction = array();
  51     protected $transactionCount = array();
  52     protected $defaultIdx = 0;
  53
  54     function __construct()
  55     {
  56         parent::__construct();
  57         $server = common_config('queue', 'stomp_server');
  58         if (is_array($server)) {
  59             $this->servers = $server;
  60         } else {
  61             $this->servers = array($server);
  62         }
  63         $this->username = common_config('queue', 'stomp_username');
  64         $this->password = common_config('queue', 'stomp_password');
  65         $this->base     = common_config('queue', 'queue_basename');
  66         $this->control  = common_config('queue', 'control_channel');
  67     }
  68
  69     /**
  70      * Tell the i/o master we only need a single instance to cover
  71      * all sites running in this process.
  72      */
  73     public static function multiSite()
  74     {
  75         return IoManager::INSTANCE_PER_PROCESS;
  76     }
  77
  78     /**
  79      * Record each site we'll be handling input for in this process,
  80      * so we can listen to the necessary queues for it.
  81      *
  82      * @fixme possibly actually do subscription here to save another
  83      *        loop over all sites later?
  84      * @fixme possibly don't assume it's the current site
  85      */
  86     public function addSite($server)
  87     {
  88         $this->sites[] = $server;
  89         $this->initialize();
  90     }
  91
  92     /**
  93      * Optional; ping any running queue handler daemons with a notification
  94      * such as announcing a new site to handle or requesting clean shutdown.
  95      * This avoids having to restart all the daemons manually to update configs
  96      * and such.
  97      *
  98      * Currently only relevant for multi-site queue managers such as Stomp.
  99      *
 100      * @param string $event event key
 101      * @param string $param optional parameter to append to key
 102      * @return boolean success
 103      */
 104     public function sendControlSignal($event, $param='')
 105     {
 106         $message = $event;
 107         if ($param != '') {
 108             $message .= ':' . $param;
 109         }
 110         $this->_connect();
 111         $result = $this->_send($this->control,
 112                                $message,
 113                                array ('created' => common_sql_now()));
 114         if ($result) {
 115             $this->_log(LOG_INFO, "Sent control ping to queue daemons: $message");
 116             return true;
 117         } else {
 118             $this->_log(LOG_ERR, "Failed sending control ping to queue daemons: $message");
 119             return false;
 120         }
 121     }
 122
 123     /**
 124      * Instantiate the appropriate QueueHandler class for the given queue.
 125      *
 126      * @param string $queue
 127      * @return mixed QueueHandler or null
 128      */
 129     function getHandler($queue)
 130     {
 131         $handlers = $this->handlers[$this->currentSite()];
 132         if (isset($handlers[$queue])) {
 133             $class = $handlers[$queue];
 134             if (class_exists($class)) {
 135                 return new $class();
 136             } else {
 137                 common_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'");
 138             }
 139         } else {
 140             common_log(LOG_ERR, "Requested handler for unkown queue '$queue'");
 141         }
 142         return null;
 143     }
 144
 145     /**
 146      * Get a list of all registered queue transport names.
 147      *
 148      * @return array of strings
 149      */
 150     function getQueues()
 151     {
 152         $group = $this->activeGroup();
 153         $site = $this->currentSite();
 154         if (empty($this->groups[$site][$group])) {
 155             return array();
 156         } else {
 157             return array_keys($this->groups[$site][$group]);
 158         }
 159     }
 160
 161     /**
 162      * Register a queue transport name and handler class for your plugin.
 163      * Only registered transports will be reliably picked up!
 164      *
 165      * @param string $transport
 166      * @param string $class
 167      * @param string $group
 168      */
 169     public function connect($transport, $class, $group='queuedaemon')
 170     {
 171         $this->handlers[$this->currentSite()][$transport] = $class;
 172         $this->groups[$this->currentSite()][$group][$transport] = $class;
 173     }
 174
 175     /**
 176      * Saves a notice object reference into the queue item table.
 177      * @return boolean true on success
 178      * @throws StompException on connection or send error
 179      */
 180     public function enqueue($object, $queue)
 181     {
 182         $this->_connect();
 183         return $this->_doEnqueue($object, $queue, $this->defaultIdx);
 184     }
 185
 186     /**
 187      * Saves a notice object reference into the queue item table
 188      * on the given connection.
 189      *
 190      * @return boolean true on success
 191      * @throws StompException on connection or send error
 192      */
 193     protected function _doEnqueue($object, $queue, $idx)
 194     {
 195         $msg = $this->encode($object);
 196         $rep = $this->logrep($object);
 197
 198         $props = array('created' => common_sql_now());
 199         if ($this->isPersistent($queue)) {
 200             $props['persistent'] = 'true';
 201         }
 202
 203         $con = $this->cons[$idx];
 204         $host = $con->getServer();
 205         $result = $con->send($this->queueName($queue), $msg, $props);
 206
 207         if (!$result) {
 208             common_log(LOG_ERR, "Error sending $rep to $queue queue on $host");
 209             return false;
 210         }
 211
 212         common_log(LOG_DEBUG, "complete remote queueing $rep for $queue on $host");
 213         $this->stats('enqueued', $queue);
 214         return true;
 215     }
 216
 217     /**
 218      * Determine whether messages to this queue should be marked as persistent.
 219      * Actual persistent storage depends on the queue server's configuration.
 220      * @param string $queue
 221      * @return bool
 222      */
 223     protected function isPersistent($queue)
 224     {
 225         $mode = common_config('queue', 'stomp_persistent');
 226         if (is_array($mode)) {
 227             return in_array($queue, $mode);
 228         } else {
 229             return (bool)$mode;
 230         }
 231     }
 232
 233     /**
 234      * Send any sockets we're listening on to the IO manager
 235      * to wait for input.
 236      *
 237      * @return array of resources
 238      */
 239     public function getSockets()
 240     {
 241         $sockets = array();
 242         foreach ($this->cons as $con) {
 243             if ($con) {
 244                 $sockets[] = $con->getSocket();
 245             }
 246         }
 247         return $sockets;
 248     }
 249
 250     /**
 251      * Get the Stomp connection object associated with the given socket.
 252      * @param resource $socket
 253      * @return int index into connections list
 254      * @throws Exception
 255      */
 256     protected function connectionFromSocket($socket)
 257     {
 258         foreach ($this->cons as $i => $con) {
 259             if ($con && $con->getSocket() === $socket) {
 260                 return $i;
 261             }
 262         }
 263         throw new Exception(__CLASS__ . " asked to read from unrecognized socket");
 264     }
 265
 266     /**
 267      * We've got input to handle on our socket!
 268      * Read any waiting Stomp frame(s) and process them.
 269      *
 270      * @param resource $socket
 271      * @return boolean ok on success
 272      */
 273     public function handleInput($socket)
 274     {
 275         $idx = $this->connectionFromSocket($socket);
 276         $con = $this->cons[$idx];
 277         $host = $con->getServer();
 278
 279         $ok = true;
 280         try {
 281             $frames = $con->readFrames();
 282         } catch (StompException $e) {
 283             common_log(LOG_ERR, "Lost connection to $host: " . $e->getMessage());
 284             $this->cons[$idx] = null;
 285             $this->transaction[$idx] = null;
 286             $this->disconnect[$idx] = time();
 287             return false;
 288         }
 289         foreach ($frames as $frame) {
 290             $dest = $frame->headers['destination'];
 291             if ($dest == $this->control) {
 292                 if (!$this->handleControlSignal($idx, $frame)) {
 293                     // We got a control event that requests a shutdown;
 294                     // close out and stop handling anything else!
 295                     break;
 296                 }
 297             } else {
 298                 $ok = $ok && $this->handleItem($idx, $frame);
 299             }
 300         }
 301         return $ok;
 302     }
 303
 304     /**
 305      * Attempt to reconnect in background if we lost a connection.
 306      */
 307     function idle()
 308     {
 309         $now = time();
 310         foreach ($this->cons as $idx => $con) {
 311             if (empty($con)) {
 312                 $age = $now - $this->disconnect[$idx];
 313                 if ($age >= 60) {
 314                     $this->_reconnect($idx);
 315                 }
 316             }
 317         }
 318         return true;
 319     }
 320
 321     /**
 322      * Initialize our connection and subscribe to all the queues
 323      * we're going to need to handle... If multiple queue servers
 324      * are configured for failover, we'll listen to all of them.
 325      *
 326      * Side effects: in multi-site mode, may reset site configuration.
 327      *
 328      * @param IoMaster $master process/event controller
 329      * @return bool return false on failure
 330      */
 331     public function start($master)
 332     {
 333         parent::start($master);
 334         $this->_connectAll();
 335
 336         common_log(LOG_INFO, "Subscribing to $this->control");
 337         foreach ($this->cons as $con) {
 338             if ($con) {
 339                 $con->subscribe($this->control);
 340             }
 341         }
 342         if ($this->sites) {
 343             foreach ($this->sites as $server) {
 344                 StatusNet::init($server);
 345                 $this->doSubscribe();
 346             }
 347         } else {
 348             $this->doSubscribe();
 349         }
 350         foreach ($this->cons as $i => $con) {
 351             if ($con) {
 352                 $this->begin($i);
 353             }
 354         }
 355         return true;
 356     }
 357
 358     /**
 359      * Subscribe to all the queues we're going to need to handle...
 360      *
 361      * Side effects: in multi-site mode, may reset site configuration.
 362      *
 363      * @return bool return false on failure
 364      */
 365     public function finish()
 366     {
 367         // If there are any outstanding delivered messages we haven't processed,
 368         // free them for another thread to take.
 369         foreach ($this->cons as $i => $con) {
 370             if ($con) {
 371                 $this->rollback($i);
 372                 $con->unsubscribe($this->control);
 373             }
 374         }
 375         if ($this->sites) {
 376             foreach ($this->sites as $server) {
 377                 StatusNet::init($server);
 378                 $this->doUnsubscribe();
 379             }
 380         } else {
 381             $this->doUnsubscribe();
 382         }
 383         return true;
 384     }
 385
 386     /**
 387      * Get identifier of the currently active site configuration
 388      * @return string
 389      */
 390     protected function currentSite()
 391     {
 392         return common_config('site', 'server'); // @fixme switch to nickname
 393     }
 394
 395     /**
 396      * Lazy open a single connection to Stomp queue server.
 397      * If multiple servers are configured, we let the Stomp client library
 398      * worry about finding a working connection among them.
 399      */
 400     protected function _connect()
 401     {
 402         if (empty($this->cons)) {
 403             $list = $this->servers;
 404             if (count($list) > 1) {
 405                 shuffle($list); // Randomize to spread load
 406                 $url = 'failover://(' . implode(',', $list) . ')';
 407             } else {
 408                 $url = $list[0];
 409             }
 410             $con = $this->_doConnect($url);
 411             $this->cons = array($con);
 412             $this->transactionCount = array(0);
 413             $this->transaction = array(null);
 414             $this->disconnect = array(null);
 415         }
 416     }
 417
 418     /**
 419      * Lazy open connections to all Stomp servers, if in manual failover
 420      * mode. This means the queue servers don't speak to each other, so
 421      * we have to listen to all of them to make sure we get all events.
 422      */
 423     protected function _connectAll()
 424     {
 425         if (!common_config('queue', 'stomp_manual_failover')) {
 426             return $this->_connect();
 427         }
 428         if (empty($this->cons)) {
 429             $this->cons = array();
 430             $this->transactionCount = array();
 431             $this->transaction = array();
 432             foreach ($this->servers as $idx => $server) {
 433                 try {
 434                     $this->cons[] = $this->_doConnect($server);
 435                     $this->disconnect[] = null;
 436                 } catch (Exception $e) {
 437                     // s'okay, we'll live
 438                     $this->cons[] = null;
 439                     $this->disconnect[] = time();
 440                 }
 441                 $this->transactionCount[] = 0;
 442                 $this->transaction[] = null;
 443             }
 444             if (empty($this->cons)) {
 445                 throw new ServerException("No queue servers reachable...");
 446                 return false;
 447             }
 448         }
 449     }
 450
 451     protected function _reconnect($idx)
 452     {
 453         try {
 454             $con = $this->_doConnect($this->servers[$idx]);
 455         } catch (Exception $e) {
 456             $this->_log(LOG_ERR, $e->getMessage());
 457             $con = null;
 458         }
 459         if ($con) {
 460             $this->cons[$idx] = $con;
 461             $this->disconnect[$idx] = null;
 462
 463             // now we have to listen to everything...
 464             // @fixme refactor this nicer. :P
 465             $host = $con->getServer();
 466             $this->_log(LOG_INFO, "Resubscribing to $this->control on $host");
 467             $con->subscribe($this->control);
 468             foreach ($this->subscriptions as $site => $queues) {
 469                 foreach ($queues as $queue) {
 470                     $this->_log(LOG_INFO, "Resubscribing to $queue on $host");
 471                     $con->subscribe($queue);
 472                 }
 473             }
 474             $this->begin($idx);
 475         } else {
 476             // Try again later...
 477             $this->disconnect[$idx] = time();
 478         }
 479     }
 480
 481     protected function _doConnect($server)
 482     {
 483         $this->_log(LOG_INFO, "Connecting to '$server' as '$this->username'...");
 484         $con = new LiberalStomp($server);
 485
 486         if ($con->connect($this->username, $this->password)) {
 487             $this->_log(LOG_INFO, "Connected.");
 488         } else {
 489             $this->_log(LOG_ERR, 'Failed to connect to queue server');
 490             throw new ServerException('Failed to connect to queue server');
 491         }
 492
 493         return $con;
 494     }
 495
 496     /**
 497      * Subscribe to all enabled notice queues for the current site.
 498      */
 499     protected function doSubscribe()
 500     {
 501         $site = $this->currentSite();
 502         $this->_connect();
 503         foreach ($this->getQueues() as $queue) {
 504             $rawqueue = $this->queueName($queue);
 505             $this->subscriptions[$site][$queue] = $rawqueue;
 506             $this->_log(LOG_INFO, "Subscribing to $rawqueue");
 507             foreach ($this->cons as $con) {
 508                 if ($con) {
 509                     $con->subscribe($rawqueue);
 510                 }
 511             }
 512         }
 513     }
 514
 515     /**
 516      * Subscribe from all enabled notice queues for the current site.
 517      */
 518     protected function doUnsubscribe()
 519     {
 520         $site = $this->currentSite();
 521         $this->_connect();
 522         if (!empty($this->subscriptions[$site])) {
 523             foreach ($this->subscriptions[$site] as $queue => $rawqueue) {
 524                 $this->_log(LOG_INFO, "Unsubscribing from $rawqueue");
 525                 foreach ($this->cons as $con) {
 526                     if ($con) {
 527                         $con->unsubscribe($rawqueue);
 528                     }
 529                 }
 530                 unset($this->subscriptions[$site][$queue]);
 531             }
 532         }
 533     }
 534
 535     /**
 536      * Handle and acknowledge an event that's come in through a queue.
 537      *
 538      * If the queue handler reports failure, the message is requeued for later.
 539      * Missing notices or handler classes will drop the message.
 540      *
 541      * Side effects: in multi-site mode, may reset site configuration to
 542      * match the site that queued the event.
 543      *
 544      * @param int $idx connection index
 545      * @param StompFrame $frame
 546      * @return bool
 547      */
 548     protected function handleItem($idx, $frame)
 549     {
 550         $this->defaultIdx = $idx;
 551
 552         list($site, $queue) = $this->parseDestination($frame->headers['destination']);
 553         if ($site != $this->currentSite()) {
 554             $this->stats('switch');
 555             StatusNet::init($site);
 556         }
 557
 558         $host = $this->cons[$idx]->getServer();
 559         if (is_numeric($frame->body)) {
 560             $id = intval($frame->body);
 561             $info = "notice $id posted at {$frame->headers['created']} in queue $queue from $host";
 562
 563             $notice = Notice::staticGet('id', $id);
 564             if (empty($notice)) {
 565                 $this->_log(LOG_WARNING, "Skipping missing $info");
 566                 $this->ack($idx, $frame);
 567                 $this->commit($idx);
 568                 $this->begin($idx);
 569                 $this->stats('badnotice', $queue);
 570                 return false;
 571             }
 572
 573             $item = $notice;
 574         } else {
 575             // @fixme should we serialize, or json, or what here?
 576             $info = "string posted at {$frame->headers['created']} in queue $queue from $host";
 577             $item = $frame->body;
 578         }
 579
 580         $handler = $this->getHandler($queue);
 581         if (!$handler) {
 582             $this->_log(LOG_ERR, "Missing handler class; skipping $info");
 583             $this->ack($idx, $frame);
 584             $this->commit($idx);
 585             $this->begin($idx);
 586             $this->stats('badhandler', $queue);
 587             return false;
 588         }
 589
 590         $ok = $handler->handle($item);
 591
 592         if (!$ok) {
 593             $this->_log(LOG_WARNING, "Failed handling $info");
 594             // FIXME we probably shouldn't have to do
 595             // this kind of queue management ourselves;
 596             // if we don't ack, it should resend...
 597             $this->ack($idx, $frame);
 598             $this->enqueue($item, $queue);
 599             $this->commit($idx);
 600             $this->begin($idx);
 601             $this->stats('requeued', $queue);
 602             return false;
 603         }
 604
 605         $this->_log(LOG_INFO, "Successfully handled $info");
 606         $this->ack($idx, $frame);
 607         $this->commit($idx);
 608         $this->begin($idx);
 609         $this->stats('handled', $queue);
 610         return true;
 611     }
 612
 613     /**
 614      * Process a control signal broadcast.
 615      *
 616      * @param int $idx connection index
 617      * @param array $frame Stomp frame
 618      * @return bool true to continue; false to stop further processing.
 619      */
 620     protected function handleControlSignal($idx, $frame)
 621     {
 622         $message = trim($frame->body);
 623         if (strpos($message, ':') !== false) {
 624             list($event, $param) = explode(':', $message, 2);
 625         } else {
 626             $event = $message;
 627             $param = '';
 628         }
 629
 630         $shutdown = false;
 631
 632         if ($event == 'shutdown') {
 633             $this->master->requestShutdown();
 634             $shutdown = true;
 635         } else if ($event == 'restart') {
 636             $this->master->requestRestart();
 637             $shutdown = true;
 638         } else if ($event == 'update') {
 639             $this->updateSiteConfig($param);
 640         } else {
 641             $this->_log(LOG_ERR, "Ignoring unrecognized control message: $message");
 642         }
 643
 644         $this->ack($idx, $frame);
 645         $this->commit($idx);
 646         $this->begin($idx);
 647         return $shutdown;
 648     }
 649
 650     /**
 651      * Set us up with queue subscriptions for a new site added at runtime,
 652      * triggered by a broadcast to the 'statusnet-control' topic.
 653      *
 654      * @param array $frame Stomp frame
 655      * @return bool true to continue; false to stop further processing.
 656      */
 657     protected function updateSiteConfig($nickname)
 658     {
 659         if (empty($this->sites)) {
 660             if ($nickname == common_config('site', 'nickname')) {
 661                 StatusNet::init(common_config('site', 'server'));
 662                 $this->doUnsubscribe();
 663                 $this->doSubscribe();
 664             } else {
 665                 $this->_log(LOG_INFO, "Ignoring update ping for other site $nickname");
 666             }
 667         } else {
 668             $sn = Status_network::staticGet($nickname);
 669             if ($sn) {
 670                 $server = $sn->getServerName(); // @fixme do config-by-nick
 671                 StatusNet::init($server);
 672                 if (empty($this->sites[$server])) {
 673                     $this->addSite($server);
 674                 }
 675                 $this->_log(LOG_INFO, "(Re)subscribing to queues for site $nickname / $server");
 676                 $this->doUnsubscribe();
 677                 $this->doSubscribe();
 678                 $this->stats('siteupdate');
 679             } else {
 680                 $this->_log(LOG_ERR, "Ignoring ping for unrecognized new site $nickname");
 681             }
 682         }
 683     }
 684
 685     /**
 686      * Combines the queue_basename from configuration with the
 687      * site server name and queue name to give eg:
 688      *
 689      * /queue/statusnet/identi.ca/sms
 690      *
 691      * @param string $queue
 692      * @return string
 693      */
 694     protected function queueName($queue)
 695     {
 696         return common_config('queue', 'queue_basename') .
 697             $this->currentSite() . '/' . $queue;
 698     }
 699
 700     /**
 701      * Returns the site and queue name from the server-side queue.
 702      *
 703      * @param string queue destination (eg '/queue/statusnet/identi.ca/sms')
 704      * @return array of site and queue: ('identi.ca','sms') or false if unrecognized
 705      */
 706     protected function parseDestination($dest)
 707     {
 708         $prefix = common_config('queue', 'queue_basename');
 709         if (substr($dest, 0, strlen($prefix)) == $prefix) {
 710             $rest = substr($dest, strlen($prefix));
 711             return explode("/", $rest, 2);
 712         } else {
 713             common_log(LOG_ERR, "Got a message from unrecognized stomp queue: $dest");
 714             return array(false, false);
 715         }
 716     }
 717
 718     function _log($level, $msg)
 719     {
 720         common_log($level, 'StompQueueManager: '.$msg);
 721     }
 722
 723     protected function begin($idx)
 724     {
 725         if ($this->useTransactions) {
 726             if (!empty($this->transaction[$idx])) {
 727                 throw new Exception("Tried to start transaction in the middle of a transaction");
 728             }
 729             $this->transactionCount[$idx]++;
 730             $this->transaction[$idx] = $this->master->id . '-' . $this->transactionCount[$idx] . '-' . time();
 731             $this->cons[$idx]->begin($this->transaction[$idx]);
 732         }
 733     }
 734
 735     protected function ack($idx, $frame)
 736     {
 737         if ($this->useTransactions) {
 738             if (empty($this->transaction[$idx])) {
 739                 throw new Exception("Tried to ack but not in a transaction");
 740             }
 741             $this->cons[$idx]->ack($frame, $this->transaction[$idx]);
 742         } else {
 743             $this->cons[$idx]->ack($frame);
 744         }
 745     }
 746
 747     protected function commit($idx)
 748     {
 749         if ($this->useTransactions) {
 750             if (empty($this->transaction[$idx])) {
 751                 throw new Exception("Tried to commit but not in a transaction");
 752             }
 753             $this->cons[$idx]->commit($this->transaction[$idx]);
 754             $this->transaction[$idx] = null;
 755         }
 756     }
 757
 758     protected function rollback($idx)
 759     {
 760         if ($this->useTransactions) {
 761             if (empty($this->transaction[$idx])) {
 762                 throw new Exception("Tried to rollback but not in a transaction");
 763             }
 764             $this->cons[$idx]->commit($this->transaction[$idx]);
 765             $this->transaction[$idx] = null;
 766         }
 767     }
 768 }
 769