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 class StompQueueManager extends QueueManager
  35 {
  36     protected $servers;
  37     protected $username;
  38     protected $password;
  39     protected $base;
  40     protected $control;
  41
  42     protected $useTransactions = true;
  43
  44     protected $sites = array();
  45     protected $subscriptions = array();
  46
  47     protected $cons = array(); // all open connections
  48     protected $disconnect = array();
  49     protected $transaction = array();
  50     protected $transactionCount = array();
  51     protected $defaultIdx = 0;
  52
  53     function __construct()
  54     {
  55         parent::__construct();
  56         $server = common_config('queue', 'stomp_server');
  57         if (is_array($server)) {
  58             $this->servers = $server;
  59         } else {
  60             $this->servers = array($server);
  61         }
  62         $this->username = common_config('queue', 'stomp_username');
  63         $this->password = common_config('queue', 'stomp_password');
  64         $this->base     = common_config('queue', 'queue_basename');
  65         $this->control  = common_config('queue', 'control_channel');
  66     }
  67
  68     /**
  69      * Tell the i/o master we only need a single instance to cover
  70      * all sites running in this process.
  71      */
  72     public static function multiSite()
  73     {
  74         return IoManager::INSTANCE_PER_PROCESS;
  75     }
  76
  77     /**
  78      * Record each site we'll be handling input for in this process,
  79      * so we can listen to the necessary queues for it.
  80      *
  81      * @fixme possibly actually do subscription here to save another
  82      *        loop over all sites later?
  83      * @fixme possibly don't assume it's the current site
  84      */
  85     public function addSite($server)
  86     {
  87         $this->sites[] = $server;
  88         $this->initialize();
  89     }
  90
  91     /**
  92      * Optional; ping any running queue handler daemons with a notification
  93      * such as announcing a new site to handle or requesting clean shutdown.
  94      * This avoids having to restart all the daemons manually to update configs
  95      * and such.
  96      *
  97      * Currently only relevant for multi-site queue managers such as Stomp.
  98      *
  99      * @param string $event event key
 100      * @param string $param optional parameter to append to key
 101      * @return boolean success
 102      */
 103     public function sendControlSignal($event, $param='')
 104     {
 105         $message = $event;
 106         if ($param != '') {
 107             $message .= ':' . $param;
 108         }
 109         $this->_connect();
 110         $con = $this->cons[$this->defaultIdx];
 111         $result = $con->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->disconnect();
 373                 $this->cons[$i] = null;
 374             }
 375         }
 376         return true;
 377     }
 378
 379     /**
 380      * Get identifier of the currently active site configuration
 381      * @return string
 382      */
 383     protected function currentSite()
 384     {
 385         return common_config('site', 'server'); // @fixme switch to nickname
 386     }
 387
 388     /**
 389      * Lazy open a single connection to Stomp queue server.
 390      * If multiple servers are configured, we let the Stomp client library
 391      * worry about finding a working connection among them.
 392      */
 393     protected function _connect()
 394     {
 395         if (empty($this->cons)) {
 396             $list = $this->servers;
 397             if (count($list) > 1) {
 398                 shuffle($list); // Randomize to spread load
 399                 $url = 'failover://(' . implode(',', $list) . ')';
 400             } else {
 401                 $url = $list[0];
 402             }
 403             $con = $this->_doConnect($url);
 404             $this->cons = array($con);
 405             $this->transactionCount = array(0);
 406             $this->transaction = array(null);
 407             $this->disconnect = array(null);
 408         }
 409     }
 410
 411     /**
 412      * Lazy open connections to all Stomp servers, if in manual failover
 413      * mode. This means the queue servers don't speak to each other, so
 414      * we have to listen to all of them to make sure we get all events.
 415      */
 416     protected function _connectAll()
 417     {
 418         if (!common_config('queue', 'stomp_manual_failover')) {
 419             return $this->_connect();
 420         }
 421         if (empty($this->cons)) {
 422             $this->cons = array();
 423             $this->transactionCount = array();
 424             $this->transaction = array();
 425             foreach ($this->servers as $idx => $server) {
 426                 try {
 427                     $this->cons[] = $this->_doConnect($server);
 428                     $this->disconnect[] = null;
 429                 } catch (Exception $e) {
 430                     // s'okay, we'll live
 431                     $this->cons[] = null;
 432                     $this->disconnect[] = time();
 433                 }
 434                 $this->transactionCount[] = 0;
 435                 $this->transaction[] = null;
 436             }
 437             if (empty($this->cons)) {
 438                 throw new ServerException("No queue servers reachable...");
 439                 return false;
 440             }
 441         }
 442     }
 443
 444     protected function _reconnect($idx)
 445     {
 446         try {
 447             $con = $this->_doConnect($this->servers[$idx]);
 448         } catch (Exception $e) {
 449             $this->_log(LOG_ERR, $e->getMessage());
 450             $con = null;
 451         }
 452         if ($con) {
 453             $this->cons[$idx] = $con;
 454             $this->disconnect[$idx] = null;
 455
 456             // now we have to listen to everything...
 457             // @fixme refactor this nicer. :P
 458             $host = $con->getServer();
 459             $this->_log(LOG_INFO, "Resubscribing to $this->control on $host");
 460             $con->subscribe($this->control);
 461             foreach ($this->subscriptions as $site => $queues) {
 462                 foreach ($queues as $queue) {
 463                     $this->_log(LOG_INFO, "Resubscribing to $queue on $host");
 464                     $con->subscribe($queue);
 465                 }
 466             }
 467             $this->begin($idx);
 468         } else {
 469             // Try again later...
 470             $this->disconnect[$idx] = time();
 471         }
 472     }
 473
 474     protected function _doConnect($server)
 475     {
 476         $this->_log(LOG_INFO, "Connecting to '$server' as '$this->username'...");
 477         $con = new LiberalStomp($server);
 478
 479         if ($con->connect($this->username, $this->password)) {
 480             $this->_log(LOG_INFO, "Connected.");
 481         } else {
 482             $this->_log(LOG_ERR, 'Failed to connect to queue server');
 483             throw new ServerException('Failed to connect to queue server');
 484         }
 485
 486         return $con;
 487     }
 488
 489     /**
 490      * Subscribe to all enabled notice queues for the current site.
 491      */
 492     protected function doSubscribe()
 493     {
 494         $site = $this->currentSite();
 495         $this->_connect();
 496         foreach ($this->getQueues() as $queue) {
 497             $rawqueue = $this->queueName($queue);
 498             $this->subscriptions[$site][$queue] = $rawqueue;
 499             $this->_log(LOG_INFO, "Subscribing to $rawqueue");
 500             foreach ($this->cons as $con) {
 501                 if ($con) {
 502                     $con->subscribe($rawqueue);
 503                 }
 504             }
 505         }
 506     }
 507
 508     /**
 509      * Subscribe from all enabled notice queues for the current site.
 510      */
 511     protected function doUnsubscribe()
 512     {
 513         $site = $this->currentSite();
 514         $this->_connect();
 515         if (!empty($this->subscriptions[$site])) {
 516             foreach ($this->subscriptions[$site] as $queue => $rawqueue) {
 517                 $this->_log(LOG_INFO, "Unsubscribing from $rawqueue");
 518                 foreach ($this->cons as $con) {
 519                     if ($con) {
 520                         $con->unsubscribe($rawqueue);
 521                     }
 522                 }
 523                 unset($this->subscriptions[$site][$queue]);
 524             }
 525         }
 526     }
 527
 528     /**
 529      * Handle and acknowledge an event that's come in through a queue.
 530      *
 531      * If the queue handler reports failure, the message is requeued for later.
 532      * Missing notices or handler classes will drop the message.
 533      *
 534      * Side effects: in multi-site mode, may reset site configuration to
 535      * match the site that queued the event.
 536      *
 537      * @param int $idx connection index
 538      * @param StompFrame $frame
 539      * @return bool
 540      */
 541     protected function handleItem($idx, $frame)
 542     {
 543         $this->defaultIdx = $idx;
 544
 545         list($site, $queue) = $this->parseDestination($frame->headers['destination']);
 546         if ($site != $this->currentSite()) {
 547             $this->stats('switch');
 548             StatusNet::init($site);
 549         }
 550
 551         $host = $this->cons[$idx]->getServer();
 552         $item = $this->decode($frame->body);
 553         if (empty($item)) {
 554             $this->_log(LOG_ERR, "Skipping empty or deleted item in queue $queue from $host");
 555             return true;
 556         }
 557         $info = $this->logrep($item) . " posted at " .
 558                 $frame->headers['created'] . " in queue $queue from $host";
 559         $this->_log(LOG_DEBUG, "Dequeued $info");
 560
 561         $handler = $this->getHandler($queue);
 562         if (!$handler) {
 563             $this->_log(LOG_ERR, "Missing handler class; skipping $info");
 564             $this->ack($idx, $frame);
 565             $this->commit($idx);
 566             $this->begin($idx);
 567             $this->stats('badhandler', $queue);
 568             return false;
 569         }
 570
 571         // If there's an exception when handling,
 572         // log the error and let it get requeued.
 573
 574         try {
 575             $ok = $handler->handle($item);
 576         } catch (Exception $e) {
 577             $this->_log(LOG_ERR, "Exception on queue $queue: " . $e->getMessage());
 578             $ok = false;
 579         }
 580
 581         if (!$ok) {
 582             $this->_log(LOG_WARNING, "Failed handling $info");
 583             // FIXME we probably shouldn't have to do
 584             // this kind of queue management ourselves;
 585             // if we don't ack, it should resend...
 586             $this->ack($idx, $frame);
 587             $this->enqueue($item, $queue);
 588             $this->commit($idx);
 589             $this->begin($idx);
 590             $this->stats('requeued', $queue);
 591             return false;
 592         }
 593
 594         $this->_log(LOG_INFO, "Successfully handled $info");
 595         $this->ack($idx, $frame);
 596         $this->commit($idx);
 597         $this->begin($idx);
 598         $this->stats('handled', $queue);
 599         return true;
 600     }
 601
 602     /**
 603      * Process a control signal broadcast.
 604      *
 605      * @param int $idx connection index
 606      * @param array $frame Stomp frame
 607      * @return bool true to continue; false to stop further processing.
 608      */
 609     protected function handleControlSignal($idx, $frame)
 610     {
 611         $message = trim($frame->body);
 612         if (strpos($message, ':') !== false) {
 613             list($event, $param) = explode(':', $message, 2);
 614         } else {
 615             $event = $message;
 616             $param = '';
 617         }
 618
 619         $shutdown = false;
 620
 621         if ($event == 'shutdown') {
 622             $this->master->requestShutdown();
 623             $shutdown = true;
 624         } else if ($event == 'restart') {
 625             $this->master->requestRestart();
 626             $shutdown = true;
 627         } else if ($event == 'update') {
 628             $this->updateSiteConfig($param);
 629         } else {
 630             $this->_log(LOG_ERR, "Ignoring unrecognized control message: $message");
 631         }
 632
 633         $this->ack($idx, $frame);
 634         $this->commit($idx);
 635         $this->begin($idx);
 636         return $shutdown;
 637     }
 638
 639     /**
 640      * Set us up with queue subscriptions for a new site added at runtime,
 641      * triggered by a broadcast to the 'statusnet-control' topic.
 642      *
 643      * @param array $frame Stomp frame
 644      * @return bool true to continue; false to stop further processing.
 645      */
 646     protected function updateSiteConfig($nickname)
 647     {
 648         if (empty($this->sites)) {
 649             if ($nickname == common_config('site', 'nickname')) {
 650                 StatusNet::init(common_config('site', 'server'));
 651                 $this->doUnsubscribe();
 652                 $this->doSubscribe();
 653             } else {
 654                 $this->_log(LOG_INFO, "Ignoring update ping for other site $nickname");
 655             }
 656         } else {
 657             $sn = Status_network::staticGet($nickname);
 658             if ($sn) {
 659                 $server = $sn->getServerName(); // @fixme do config-by-nick
 660                 StatusNet::init($server);
 661                 if (empty($this->sites[$server])) {
 662                     $this->addSite($server);
 663                 }
 664                 $this->_log(LOG_INFO, "(Re)subscribing to queues for site $nickname / $server");
 665                 $this->doUnsubscribe();
 666                 $this->doSubscribe();
 667                 $this->stats('siteupdate');
 668             } else {
 669                 $this->_log(LOG_ERR, "Ignoring ping for unrecognized new site $nickname");
 670             }
 671         }
 672     }
 673
 674     /**
 675      * Combines the queue_basename from configuration with the
 676      * site server name and queue name to give eg:
 677      *
 678      * /queue/statusnet/identi.ca/sms
 679      *
 680      * @param string $queue
 681      * @return string
 682      */
 683     protected function queueName($queue)
 684     {
 685         return common_config('queue', 'queue_basename') .
 686             $this->currentSite() . '/' . $queue;
 687     }
 688
 689     /**
 690      * Returns the site and queue name from the server-side queue.
 691      *
 692      * @param string queue destination (eg '/queue/statusnet/identi.ca/sms')
 693      * @return array of site and queue: ('identi.ca','sms') or false if unrecognized
 694      */
 695     protected function parseDestination($dest)
 696     {
 697         $prefix = common_config('queue', 'queue_basename');
 698         if (substr($dest, 0, strlen($prefix)) == $prefix) {
 699             $rest = substr($dest, strlen($prefix));
 700             return explode("/", $rest, 2);
 701         } else {
 702             common_log(LOG_ERR, "Got a message from unrecognized stomp queue: $dest");
 703             return array(false, false);
 704         }
 705     }
 706
 707     function _log($level, $msg)
 708     {
 709         common_log($level, 'StompQueueManager: '.$msg);
 710     }
 711
 712     protected function begin($idx)
 713     {
 714         if ($this->useTransactions) {
 715             if (!empty($this->transaction[$idx])) {
 716                 throw new Exception("Tried to start transaction in the middle of a transaction");
 717             }
 718             $this->transactionCount[$idx]++;
 719             $this->transaction[$idx] = $this->master->id . '-' . $this->transactionCount[$idx] . '-' . time();
 720             $this->cons[$idx]->begin($this->transaction[$idx]);
 721         }
 722     }
 723
 724     protected function ack($idx, $frame)
 725     {
 726         if ($this->useTransactions) {
 727             if (empty($this->transaction[$idx])) {
 728                 throw new Exception("Tried to ack but not in a transaction");
 729             }
 730             $this->cons[$idx]->ack($frame, $this->transaction[$idx]);
 731         } else {
 732             $this->cons[$idx]->ack($frame);
 733         }
 734     }
 735
 736     protected function commit($idx)
 737     {
 738         if ($this->useTransactions) {
 739             if (empty($this->transaction[$idx])) {
 740                 throw new Exception("Tried to commit but not in a transaction");
 741             }
 742             $this->cons[$idx]->commit($this->transaction[$idx]);
 743             $this->transaction[$idx] = null;
 744         }
 745     }
 746
 747     protected function rollback($idx)
 748     {
 749         if ($this->useTransactions) {
 750             if (empty($this->transaction[$idx])) {
 751                 throw new Exception("Tried to rollback but not in a transaction");
 752             }
 753             $this->cons[$idx]->commit($this->transaction[$idx]);
 754             $this->transaction[$idx] = null;
 755         }
 756     }
 757 }
 758