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