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@controlyourself.ca>
  25  * @author    Sarven Capadisli <csarven@controlyourself.ca>
  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://laconi.ca/
  29  */
  30
  31 require_once 'Stomp.php';
  32
  33 class LiberalStomp extends Stomp
  34 {
  35     function getSocket()
  36     {
  37         return $this->_socket;
  38     }
  39 }
  40
  41 class StompQueueManager
  42 {
  43     var $server = null;
  44     var $username = null;
  45     var $password = null;
  46     var $base = null;
  47     var $con = null;
  48
  49     function __construct()
  50     {
  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     function _connect()
  58     {
  59         if (empty($this->con)) {
  60             $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
  61             $this->con = new LiberalStomp($this->server);
  62
  63             if ($this->con->connect($this->username, $this->password)) {
  64                 $this->_log(LOG_INFO, "Connected.");
  65             } else {
  66                 $this->_log(LOG_ERR, 'Failed to connect to queue server');
  67                 throw new ServerException('Failed to connect to queue server');
  68             }
  69         }
  70     }
  71
  72     function enqueue($object, $queue)
  73     {
  74         $notice = $object;
  75
  76         $this->_connect();
  77
  78         // XXX: serialize and send entire notice
  79
  80         $result = $this->con->send($this->_queueName($queue),
  81                                    $notice->id,                 // BODY of the message
  82                                    array ('created' => $notice->created));
  83
  84         if (!$result) {
  85             common_log(LOG_ERR, 'Error sending to '.$queue.' queue');
  86             return false;
  87         }
  88
  89         common_log(LOG_DEBUG, 'complete remote queueing notice ID = '
  90                    . $notice->id . ' for ' . $queue);
  91     }
  92
  93     function service($queue, $handler)
  94     {
  95         $result = null;
  96
  97         $this->_connect();
  98
  99         $this->con->setReadTimeout($handler->timeout());
 100
 101         $this->con->subscribe($this->_queueName($queue));
 102
 103         while (true) {
 104
 105             // Wait for something on one of our sockets
 106
 107             $stompsock = $this->con->getSocket();
 108
 109             $handsocks = $handler->getSockets();
 110
 111             $socks = array_merge(array($stompsock), $handsocks);
 112
 113             $read = $socks;
 114             $write = array();
 115             $except = array();
 116
 117             $ready = stream_select($read, $write, $except, $handler->timeout(), 0);
 118
 119             if ($ready === false) {
 120                 $this->_log(LOG_ERR, "Error selecting on sockets");
 121             } else if ($ready > 0) {
 122                 if (in_array($stompsock, $read)) {
 123                     $this->_handleNotice($queue, $handler);
 124                 }
 125                 $handler->idle(QUEUE_HANDLER_HIT_IDLE);
 126             }
 127         }
 128
 129         $this->con->unsubscribe($this->_queueName($queue));
 130     }
 131
 132     function _handleNotice($queue, $handler)
 133     {
 134         $frame = $this->con->readFrame();
 135
 136         if (!empty($frame)) {
 137             $notice = Notice::staticGet('id', $frame->body);
 138
 139             if (empty($notice)) {
 140                 $this->_log(LOG_WARNING, 'Got ID '. $frame->body .' for non-existent notice in queue '. $queue);
 141                 $this->con->ack($frame);
 142             } else {
 143                 if ($handler->handle_notice($notice)) {
 144                     $this->_log(LOG_INFO, 'Successfully handled notice '. $notice->id .' posted at ' . $frame->headers['created'] . ' in queue '. $queue);
 145                     $this->con->ack($frame);
 146                 } else {
 147                     $this->_log(LOG_WARNING, 'Failed handling notice '. $notice->id .' posted at ' . $frame->headers['created']  . ' in queue '. $queue);
 148                     // FIXME we probably shouldn't have to do
 149                     // this kind of queue management ourselves
 150                     $this->con->ack($frame);
 151                     $this->enqueue($notice, $queue);
 152                 }
 153                 unset($notice);
 154             }
 155
 156             unset($frame);
 157         }
 158     }
 159
 160     function _queueName($queue)
 161     {
 162         return common_config('queue', 'queue_basename') . $queue;
 163     }
 164
 165     function _log($level, $msg)
 166     {
 167         common_log($level, 'StompQueueManager: '.$msg);
 168     }
 169 }